fix: load knowledge in workbench context fallback

application/engine/services/query_service.py

@@ -31,6 +31,8 @@
 
 logger = logging.getLogger(__name__)
 
+_WORKBENCH_CONTEXT_BACKFILL_ATTEMPTED_KEY = "_workbench_context_backfill_v2_attempted"
+
 # 守护进程经 _update_shared_state 写入、/status 需透出的运行时字段（不在 NovelState 模型内）
 _RUNTIME_STATUS_KEYS: tuple[str, ...] = (
     "writing_substep",
@@ -314,6 +316,7 @@ class QueryService:
 
     def __init__(self, shared_state: Optional[SharedStateRepository] = None):
         self._shared = shared_state or get_shared_state_repository()
+        self._workbench_context_backfill_attempted: set[str] = set()
 
     # ==================== 小说状态 ====================
 
@@ -559,15 +562,60 @@ def get_workbench_context(self, novel_id: str) -> WorkbenchContextResponse:
         plot_arc = self._shared.get_plot_arc(novel_id)
         knowledge = self._shared.get_knowledge(novel_id)
         foreshadows = self._shared.get_foreshadows(novel_id)
+        triples = self._shared.get_triples(novel_id)
         chapters = self._shared.get_chapters(novel_id)
 
         # 如果共享内存中没有数据，降级到数据库查询
         if not storylines and not chapters:
             logger.debug(f"共享内存中没有小说 {novel_id} 的工作台数据，从数据库加载")
             return self._fallback_workbench_from_db(novel_id)
 
-        # 计算最大章节号
+        # Backfill optional workbench panels once when an older bootstrap path only warmed
+        # chapters/storylines. Empty lists/None may also be a legitimate result for early
+        # novels, so cache the attempt and avoid hitting SQLite on every workbench request.
+        raw_state = self._shared.get_raw_state(novel_id) or {}
+        backfill_attempted = (
+            novel_id in self._workbench_context_backfill_attempted
+            or bool(raw_state.get(_WORKBENCH_CONTEXT_BACKFILL_ATTEMPTED_KEY))
+        )
+        needs_backfill = (
+            not chronicles
+            or plot_arc is None
+            or knowledge is None
+            or not foreshadows
+            or not triples
+        )
+        if needs_backfill and not backfill_attempted:
+            try:
+                from application.engine.services.state_bootstrap import StateBootstrap
+
+                bootstrap = StateBootstrap()
+                if not triples:
+                    triples = bootstrap._load_triples(novel_id)
+                if not foreshadows:
+                    foreshadows = bootstrap._load_foreshadows(novel_id)
+                if plot_arc is None:
+                    plot_arc = bootstrap._load_plot_arc(novel_id)
+                if knowledge is None:
+                    knowledge = bootstrap._load_knowledge(novel_id)
+                if not chronicles:
+                    # Chronicles depend on Bible / snapshots / chapters, so hydrate them first.
+                    bootstrap._load_bible(novel_id)
+                    bootstrap._load_snapshots(novel_id)
+                    chapters = self._shared.get_chapters(novel_id) or chapters
+                    chronicles = bootstrap._load_chronicles(novel_id)
+            except Exception as e:
+                logger.debug("Workbench context backfill failed: novel=%s err=%s", novel_id, e)
+            else:
+                self._workbench_context_backfill_attempted.add(novel_id)
+                self._shared.merge_raw_state(
+                    novel_id,
+                    **{_WORKBENCH_CONTEXT_BACKFILL_ATTEMPTED_KEY: True},
+                )
+
+        # Calculate max chapter number
         max_ch = max((c.number for c in chapters), default=1)
+        knowledge_graph = self.get_knowledge_graph(novel_id)
 
         return WorkbenchContextResponse(
             novel_id=novel_id,
@@ -581,7 +629,7 @@ def get_workbench_context(self, novel_id: str) -> WorkbenchContextResponse:
             plot_arc=plot_arc,
             knowledge=knowledge,
             foreshadow_ledger=foreshadows,
-            knowledge_graph={"total_triples": 0, "by_source": {}},  # 需要单独处理
+            knowledge_graph=knowledge_graph,
             macro={"narrative_event_count": 0},  # 需要单独处理
             sandbox={"bible_character_count": 0},  # 需要单独处理
             chapters_digest=[c.to_dict() for c in chapters],
@@ -601,8 +649,18 @@ def _fallback_workbench_from_db(self, novel_id: str) -> WorkbenchContextResponse
             chapters_data = bootstrap._load_chapters(novel_id)
             foreshadows_data = bootstrap._load_foreshadows(novel_id)
             plot_arc_data = bootstrap._load_plot_arc(novel_id)
+            bootstrap._load_bible(novel_id)
+            triples_data = bootstrap._load_triples(novel_id)
+            bootstrap._load_snapshots(novel_id)
             knowledge_data = bootstrap._load_knowledge(novel_id)
             chronicles_data = bootstrap._load_chronicles(novel_id)
+            knowledge_graph = {
+                "total_triples": len(triples_data),
+                "by_source": {},
+            }
+            for t in triples_data:
+                src = t.get("source_type", "unknown")
+                knowledge_graph["by_source"][src] = knowledge_graph["by_source"].get(src, 0) + 1
 
             max_ch = max((c.number for c in chapters_data), default=1) if chapters_data else 1
 
@@ -618,7 +676,7 @@ def _fallback_workbench_from_db(self, novel_id: str) -> WorkbenchContextResponse
                 plot_arc=plot_arc_data,
                 knowledge=knowledge_data,
                 foreshadow_ledger=foreshadows_data,
-                knowledge_graph={"total_triples": 0, "by_source": {}},
+                knowledge_graph=knowledge_graph,
                 macro={"narrative_event_count": 0},
                 sandbox={"bible_character_count": 0},
                 chapters_digest=[c.to_dict() for c in chapters_data],

application/engine/services/state_bootstrap.py

@@ -84,6 +84,9 @@ def load_all(self) -> Dict[str, Any]:
                     # 加载 Bible
                     self._load_bible(novel_id)
 
+                    # 加载叙事知识（premise lock / 章后摘要 / facts）
+                    self._load_knowledge(novel_id)
+
                     # 加载三元组
                     self._load_triples(novel_id)
 
@@ -135,6 +138,21 @@ def load_novel(self, novel_id: str) -> bool:
             # 加载剧情弧光
             self._load_plot_arc(novel_id)
 
+            # 加载 Bible
+            self._load_bible(novel_id)
+
+            # 加载叙事知识
+            self._load_knowledge(novel_id)
+
+            # 加载三元组
+            self._load_triples(novel_id)
+
+            # 加载快照
+            self._load_snapshots(novel_id)
+
+            # 加载编年史
+            self._load_chronicles(novel_id)
+
             return True
 
         except Exception as e:
@@ -293,6 +311,31 @@ def _load_foreshadows(self, novel_id: str) -> List[Dict[str, Any]]:
                 self._shared.set_foreshadows(novel_id, entries)
                 return entries
 
+            rows = db.fetch_all(
+                """SELECT id, description, planted_chapter, due_chapter,
+                          resolved_chapter, status, importance, subtext_type
+                   FROM foreshadows
+                   WHERE novel_id = ?
+                   ORDER BY planted_chapter, importance DESC, id""",
+                (novel_id,),
+            )
+            entries = [
+                {
+                    "id": r["id"],
+                    "description": r["description"],
+                    "planted_chapter": r["planted_chapter"],
+                    "due_chapter": r["due_chapter"],
+                    "resolved_chapter": r["resolved_chapter"],
+                    "status": r["status"],
+                    "importance": r["importance"],
+                    "subtext_type": r["subtext_type"],
+                }
+                for r in rows
+            ]
+            if entries:
+                self._shared.set_foreshadows(novel_id, entries)
+                return entries
+
             return []
 
         except Exception as e:
@@ -425,6 +468,77 @@ def _load_bible(self, novel_id: str) -> Optional[Dict[str, Any]]:
             logger.debug(f"加载 Bible 失败（可能不存在）: {novel_id}, {e}")
             return None
 
+    def _load_knowledge(self, novel_id: str) -> Optional[Dict[str, Any]]:
+        """加载叙事知识到共享内存。
+
+        workbench-context 的 DB fallback 已经依赖该方法；缺失时会导致
+        “作品基础/知识库/章后摘要”在共享状态或降级读取中全部为空。
+        """
+        try:
+            from domain.novel.value_objects.novel_id import NovelId
+            from infrastructure.persistence.database.connection import get_database
+            from infrastructure.persistence.database.sqlite_knowledge_repository import (
+                SqliteKnowledgeRepository,
+            )
+
+            repo = SqliteKnowledgeRepository(get_database())
+            knowledge = repo.get_by_novel_id(NovelId(novel_id))
+
+            if not knowledge:
+                return None
+
+            chapters = [
+                {
+                    "chapter_id": ch.chapter_id,
+                    "summary": ch.summary,
+                    "key_events": ch.key_events,
+                    "open_threads": ch.open_threads,
+                    "consistency_note": ch.consistency_note,
+                    "beat_sections": list(ch.beat_sections or []),
+                    "micro_beats": list(ch.micro_beats or []),
+                    "sync_status": ch.sync_status,
+                }
+                for ch in (knowledge.chapters or [])
+            ]
+
+            facts = [
+                {
+                    "id": fact.id,
+                    "subject": fact.subject,
+                    "predicate": fact.predicate,
+                    "object": fact.object,
+                    "chapter_id": fact.chapter_id,
+                    "note": fact.note,
+                    "entity_type": fact.entity_type,
+                    "importance": fact.importance,
+                    "location_type": fact.location_type,
+                    "description": fact.description,
+                    "first_appearance": fact.first_appearance,
+                    "related_chapters": list(fact.related_chapters or []),
+                    "tags": list(fact.tags or []),
+                    "attributes": dict(fact.attributes or {}),
+                    "confidence": fact.confidence,
+                    "source_type": fact.source_type,
+                    "subject_entity_id": fact.subject_entity_id,
+                    "object_entity_id": fact.object_entity_id,
+                    "provenance": list(getattr(fact, "provenance", []) or []),
+                }
+                for fact in (knowledge.facts or [])
+            ]
+
+            knowledge_dict = {
+                "version": knowledge.version,
+                "premise_lock": knowledge.premise_lock,
+                "chapters": chapters,
+                "facts": facts,
+            }
+            self._shared.set_knowledge(novel_id, knowledge_dict)
+            return knowledge_dict
+
+        except Exception as e:
+            logger.debug(f"加载叙事知识失败（可能不存在）: {novel_id}, {e}")
+            return None
+
     def _load_triples(self, novel_id: str) -> List[Dict[str, Any]]:
         """加载三元组到共享内存"""
         try:
@@ -567,6 +681,13 @@ def refresh_narrative_contract_in_shared_state(novel_id: str) -> None:
     try:
         bootstrap = StateBootstrap()
         bootstrap._load_bible(novel_id)
+        bootstrap._load_chapters(novel_id)
+        bootstrap._load_foreshadows(novel_id)
         bootstrap._load_storylines(novel_id)
+        bootstrap._load_plot_arc(novel_id)
+        bootstrap._load_knowledge(novel_id)
+        bootstrap._load_triples(novel_id)
+        bootstrap._load_snapshots(novel_id)
+        bootstrap._load_chronicles(novel_id)
     except Exception as e:
         logger.debug("叙事契约共享状态刷新失败 novel=%s err=%s", novel_id, e)