feat: add knowledge_query temporal API with CLI and prompt sync

webnovel-writer/agents/context-agent.md

@@ -69,6 +69,10 @@ python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" inde
 
 # 全量上下文（备选，兼容老项目）
 python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" extract-context --chapter {NNNN} --format json
+
+# 时序知识查询（查询某实体在指定章节时的状态和关系）
+python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" knowledge query-entity-state --entity "{entity_id}" --at-chapter {N}
+python -X utf8 "${SCRIPTS_DIR}/webnovel.py" --project-root "{project_root}" knowledge query-relationships --entity "{entity_id}" --at-chapter {N}
 ```
 
 参考资料（按需加载）：

webnovel-writer/scripts/data_modules/knowledge_query.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+from typing import Any, Dict, List
+
+
+class KnowledgeQuery:
+    def __init__(self, project_root: Path):
+        self.project_root = Path(project_root)
+        self._db_path = self.project_root / ".webnovel" / "index.db"
+
+    def entity_state_at_chapter(self, entity_id: str, chapter: int) -> Dict[str, Any]:
+        """查询实体在指定章节时的状态（从 state_changes 反推）。"""
+        conn = sqlite3.connect(str(self._db_path))
+        conn.row_factory = sqlite3.Row
+        try:
+            rows = conn.execute(
+                """
+                SELECT field, new_value
+                FROM state_changes
+                WHERE entity_id = ? AND chapter <= ?
+                ORDER BY chapter ASC, id ASC
+                """,
+                (entity_id, chapter),
+            ).fetchall()
+
+            state: Dict[str, str] = {}
+            for row in rows:
+                field = str(row["field"] or "").strip()
+                if field:
+                    state[field] = str(row["new_value"] or "").strip()
+
+            return {
+                "entity_id": entity_id,
+                "at_chapter": chapter,
+                "state_at_chapter": state,
+            }
+        finally:
+            conn.close()
+
+    def entity_relationships_at_chapter(self, entity_id: str, chapter: int) -> Dict[str, Any]:
+        """查询实体在指定章节时的所有关系。"""
+        conn = sqlite3.connect(str(self._db_path))
+        conn.row_factory = sqlite3.Row
+        try:
+            rows = conn.execute(
+                """
+                SELECT from_entity, to_entity, relationship_type, description, chapter
+                FROM relationship_events
+                WHERE (from_entity = ? OR to_entity = ?) AND chapter <= ?
+                ORDER BY chapter ASC, id ASC
+                """,
+                (entity_id, entity_id, chapter),
+            ).fetchall()
+
+            latest: Dict[str, Dict[str, Any]] = {}
+            for row in rows:
+                from_e = str(row["from_entity"] or "").strip()
+                to_e = str(row["to_entity"] or "").strip()
+                pair_key = tuple(sorted([from_e, to_e]))
+                latest[str(pair_key)] = {
+                    "from_entity": from_e,
+                    "to_entity": to_e,
+                    "relationship_type": str(row["relationship_type"] or "").strip(),
+                    "description": str(row["description"] or "").strip(),
+                    "since_chapter": int(row["chapter"] or 0),
+                }
+
+            return {
+                "entity_id": entity_id,
+                "at_chapter": chapter,
+                "relationships": list(latest.values()),
+            }
+        finally:
+            conn.close()

webnovel-writer/scripts/data_modules/tests/test_knowledge_query.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""KnowledgeQuery 时序查询测试。"""
+import json
+import sqlite3
+from pathlib import Path
+
+import pytest
+
+from data_modules.knowledge_query import KnowledgeQuery
+
+
+@pytest.fixture
+def setup_db(tmp_path):
+    db_path = tmp_path / ".webnovel" / "index.db"
+    db_path.parent.mkdir(parents=True)
+
+    conn = sqlite3.connect(str(db_path))
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS entities (
+            id TEXT PRIMARY KEY,
+            canonical_name TEXT,
+            type TEXT DEFAULT '角色',
+            current_json TEXT DEFAULT '{}',
+            created_at TEXT,
+            updated_at TEXT
+        )
+    """)
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS state_changes (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            entity_id TEXT,
+            field TEXT,
+            old_value TEXT,
+            new_value TEXT,
+            chapter INTEGER,
+            created_at TEXT
+        )
+    """)
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS relationship_events (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            from_entity TEXT,
+            to_entity TEXT,
+            relationship_type TEXT,
+            description TEXT,
+            chapter INTEGER,
+            created_at TEXT
+        )
+    """)
+
+    conn.execute(
+        "INSERT INTO entities (id, canonical_name, current_json) VALUES (?, ?, ?)",
+        ("hanli", "韩立", json.dumps({"realm": "筑基中期", "location": "乱星海"})),
+    )
+    conn.execute(
+        "INSERT INTO state_changes (entity_id, field, old_value, new_value, chapter) VALUES (?, ?, ?, ?, ?)",
+        ("hanli", "realm", "练气圆满", "筑基初期", 30),
+    )
+    conn.execute(
+        "INSERT INTO state_changes (entity_id, field, old_value, new_value, chapter) VALUES (?, ?, ?, ?, ?)",
+        ("hanli", "realm", "筑基初期", "筑基中期", 50),
+    )
+    conn.execute(
+        "INSERT INTO relationship_events (from_entity, to_entity, relationship_type, chapter) VALUES (?, ?, ?, ?)",
+        ("hanli", "陈巧倩", "同门", 20),
+    )
+    conn.execute(
+        "INSERT INTO relationship_events (from_entity, to_entity, relationship_type, chapter) VALUES (?, ?, ?, ?)",
+        ("hanli", "陈巧倩", "合作", 45),
+    )
+    conn.commit()
+    conn.close()
+    return tmp_path
+
+
+def test_entity_state_at_chapter_before_first_change(setup_db):
+    kq = KnowledgeQuery(setup_db)
+    result = kq.entity_state_at_chapter("hanli", 10)
+    assert result["entity_id"] == "hanli"
+    assert result["state_at_chapter"] == {}
+
+
+def test_entity_state_at_chapter_after_first_breakthrough(setup_db):
+    kq = KnowledgeQuery(setup_db)
+    result = kq.entity_state_at_chapter("hanli", 35)
+    assert result["state_at_chapter"]["realm"] == "筑基初期"
+
+
+def test_entity_state_at_chapter_after_second_breakthrough(setup_db):
+    kq = KnowledgeQuery(setup_db)
+    result = kq.entity_state_at_chapter("hanli", 60)
+    assert result["state_at_chapter"]["realm"] == "筑基中期"
+
+
+def test_relationships_at_chapter_before_any(setup_db):
+    kq = KnowledgeQuery(setup_db)
+    result = kq.entity_relationships_at_chapter("hanli", 10)
+    assert result["relationships"] == []
+
+
+def test_relationships_at_chapter_after_first(setup_db):
+    kq = KnowledgeQuery(setup_db)
+    result = kq.entity_relationships_at_chapter("hanli", 25)
+    assert len(result["relationships"]) == 1
+    assert result["relationships"][0]["to_entity"] == "陈巧倩"
+    assert result["relationships"][0]["relationship_type"] == "同门"
+
+
+def test_relationships_at_chapter_after_update(setup_db):
+    kq = KnowledgeQuery(setup_db)
+    result = kq.entity_relationships_at_chapter("hanli", 50)
+    rels = result["relationships"]
+    assert len(rels) == 1
+    assert rels[0]["relationship_type"] == "合作"

webnovel-writer/scripts/data_modules/tests/test_prompt_integrity.py

@@ -34,7 +34,7 @@ REGISTERED_CLI_SUBCOMMANDS = {
     "index", "state", "rag", "style", "entity", "context", "memory",
     "migrate", "status", "update-state", "backup", "archive",
     "init", "extract-context", "memory-contract", "review-pipeline",
-    "story-system", "chapter-commit", "story-events",
+    "story-system", "chapter-commit", "story-events", "knowledge",
 }
 
 

webnovel-writer/scripts/data_modules/webnovel.py

@@ -290,6 +290,17 @@ def main() -> None:
     p_review_pipeline.add_argument("--metrics-out", default="", help="metrics 输出文件")
     p_review_pipeline.add_argument("--report-file", default="", help="审查报告路径")
 
+    knowledge_parser = sub.add_parser("knowledge", help="时序知识查询")
+    knowledge_sub = knowledge_parser.add_subparsers(dest="knowledge_action")
+
+    qs_parser = knowledge_sub.add_parser("query-entity-state", help="查询实体在指定章节的状态")
+    qs_parser.add_argument("--entity", required=True, help="实体 ID")
+    qs_parser.add_argument("--at-chapter", type=int, required=True, help="目标章节号")
+
+    qr_parser = knowledge_sub.add_parser("query-relationships", help="查询实体在指定章节的关系")
+    qr_parser.add_argument("--entity", required=True, help="实体 ID")
+    qr_parser.add_argument("--at-chapter", type=int, required=True, help="目标章节号")
+
     # 兼容：允许 `--project-root` 出现在任意位置（减少 agents/skills 拼命令的出错率）
     from .cli_args import normalize_global_project_root
 
@@ -378,6 +389,19 @@ def main() -> None:
             return_args.extend(["--report-file", str(args.report_file)])
         raise SystemExit(_run_script("review_pipeline.py", return_args))
 
+    if tool == "knowledge":
+        from .knowledge_query import KnowledgeQuery
+        from .cli_output import print_success
+        kq = KnowledgeQuery(project_root)
+        if args.knowledge_action == "query-entity-state":
+            result = kq.entity_state_at_chapter(args.entity, args.at_chapter)
+            print_success(result, message="entity_state_at_chapter")
+            raise SystemExit(0)
+        elif args.knowledge_action == "query-relationships":
+            result = kq.entity_relationships_at_chapter(args.entity, args.at_chapter)
+            print_success(result, message="entity_relationships_at_chapter")
+            raise SystemExit(0)
+
     raise SystemExit(2)