框架: adopt 走 flow、补齐迁移对账、派发默认不写 PG

data/migrations/0003_snapshots.sql

@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS snapshots (
+  id TEXT PRIMARY KEY,
+  kind TEXT NOT NULL,
+  payload_json TEXT NOT NULL,
+  created_at TEXT NOT NULL
+);

docs/plans/2026-08-28-能力原型间收敛方案.md

@@ -220,7 +220,7 @@ events（会话）+ reviews（人审）+ revisions（修订 diff）
 - [x] P0.1 删 catalog 双投影  [x] 2026-08-28 git ls-files framework/catalog 为空；generate 写入 .pi/skills 与 .dsh/skills（各 15 个，gitignore）；架构测试 12 项绿
 - [x] P0.2 拆 dispatch_agent_task  [x] 2026-08-28 CLI 66 行；runtime 纯度门绿；dispatch 离线测试 37 项绿
 - [x] P0.3 建 data/muse.db（runs/events/reviews/revisions/cards）  [x] 2026-08-28 WAL+五表；dispatch 假 launcher 写入 run+events；revise 含 revisions；同 input 重跑 output 一致；git status 不因运行新增未忽略文件
-- [x] P0.4 存量迁移 PG → SQLite  [x] 2026-08-28 成果表对账相等；向量抽样余弦≈1.0；章节 11785=11785 抽读完好；reviews=6；muse.db 150MB；PG 只读未写入
+- [x] P0.4 存量迁移 PG → SQLite  [x] 2026-08-28 成果+记录 COUNT pg=sqlite 成对相等（含 document/base/refwork/candidate/cas/planning/freeze）；向量抽样余弦≈1.0；章节 11785=11785；reviews=6；muse.db 150MB；生产派发默认不写 PG
 - [x] P1.4 蒸馏闭环（lesson 证据强制 + 人批准 + skill 升格）  [x] 2026-08-28 空证据指针被拒；一条 lesson 经 run/review id 升格进 references
 - [x] P1.5 回放链改用生产 flow 入口  [x] 2026-08-28 muse.replay.production_run_dispatch is muse.flow.dispatch.run_dispatch
 - [x] P1.6 人审工作台 web/  [x] 2026-08-28 写面仅 reviews/revisions/adopt；adopt 只记 reviews

muse/flow/adopt.py

@@ -0,0 +1,24 @@
+"""采纳走 flow：记录 reviews.adopt，不直接写正文文件。"""
+
+from __future__ import annotations
+
+from muse.store import add_review
+
+
+def adopt_candidate(
+    run_id: str,
+    reviewer: str,
+    reason: str,
+    *,
+    path: str | None = None,
+) -> int:
+    """人审采纳的唯一写入口。web 必须调用本函数，不得自行写库。"""
+
+    return add_review(
+        run_id=run_id,
+        target="candidate",
+        action="adopt",
+        reviewer=reviewer,
+        reason=reason,
+        path=path,
+    )

muse/flow/dispatch.py

@@ -300,38 +300,36 @@ def run_dispatch(
         )
         write_private_json(run_dir_path / "receipt.json", receipt)
         return receipt, EXIT_SPEC_INVALID
-    try:
-        run_record = start_run(
-            connect=connect_factory,
-            run_id=run_id,
-            work_id=spec.work_id,
-            target_chapter=spec.target_chapter,
-            trigger_source=trigger_source,
-            trigger_detail=detail,
-        )
-    except Exception as exc:  # noqa: BLE001 - 注册失败不能启动外部 Agent。
-        receipt = _receipt(
-            status="failed",
-            errorCode="RUN_REGISTRY_START_FAILED",
-            error=f"运行登记失败: {type(exc).__name__}",
-        )
-        write_private_json(run_dir_path / "receipt.json", receipt)
-        return receipt, EXIT_EVIDENCE_FAILED
-    if run_record["status"] != "started":
-        receipt = _receipt(
-            status="failed",
-            errorCode="RUN_ID_EXISTS",
-            error="run_id 已存在，拒绝覆盖既有运行证据",
-        )
-        write_private_json(run_dir_path / "receipt.json", receipt)
-        return receipt, EXIT_EVIDENCE_FAILED
+    persist_pg = connect_factory is not None
+    if persist_pg:
+        try:
+            run_record = start_run(
+                connect=connect_factory,
+                run_id=run_id,
+                work_id=spec.work_id,
+                target_chapter=spec.target_chapter,
+                trigger_source=trigger_source,
+                trigger_detail=detail,
+            )
+        except Exception as exc:  # noqa: BLE001 - 注册失败不能启动外部 Agent。
+            receipt = _receipt(
+                status="failed",
+                errorCode="RUN_REGISTRY_START_FAILED",
+                error=f"运行登记失败: {type(exc).__name__}",
+            )
+            write_private_json(run_dir_path / "receipt.json", receipt)
+            return receipt, EXIT_EVIDENCE_FAILED
+        if run_record["status"] != "started":
+            receipt = _receipt(
+                status="failed",
+                errorCode="RUN_ID_EXISTS",
+                error="run_id 已存在，拒绝覆盖既有运行证据",
+            )
+            write_private_json(run_dir_path / "receipt.json", receipt)
+            return receipt, EXIT_EVIDENCE_FAILED
+    else:
+        run_record = {"status": "started"}
 
-    inner_writer = AgentTraceWriter(
-        run_id=run_id,
-        framework=effective_policy.framework,
-        agent_role=spec.role,
-        connect=connect_factory,
-    )
     recorder = SqliteRecorder(
         run_id,
         kind=f"agent.{spec.role}",
@@ -347,7 +345,16 @@ def run_dispatch(
         role=spec.role,
         path=sqlite_path,
     )
-    writer = FanoutSink(inner_writer, recorder)
+    if persist_pg:
+        inner_writer = AgentTraceWriter(
+            run_id=run_id,
+            framework=effective_policy.framework,
+            agent_role=spec.role,
+            connect=connect_factory,
+        )
+        writer = FanoutSink(inner_writer, recorder)
+    else:
+        writer = recorder
 
     def _failed(
         error_code: str,
@@ -388,15 +395,16 @@ def run_dispatch(
             )
         except Exception as exc:  # noqa: BLE001 - 继续尝试闭合 example_run。
             failures.append(type(exc).__name__)
-        try:
-            finish_run(
-                run_id,
-                "failed",
-                trigger_detail={"errorCode": error_code},
-                connect=connect_factory,
-            )
-        except Exception as exc:  # noqa: BLE001 - 回执必须揭示终态未能闭合。
-            failures.append(type(exc).__name__)
+        if persist_pg:
+            try:
+                finish_run(
+                    run_id,
+                    "failed",
+                    trigger_detail={"errorCode": error_code},
+                    connect=connect_factory,
+                )
+            except Exception as exc:  # noqa: BLE001 - 回执必须揭示终态未能闭合。
+                failures.append(type(exc).__name__)
 
         fields: dict[str, Any] = {
             "status": "failed",
@@ -461,6 +469,8 @@ def run_dispatch(
         except ValueError:
             remove_local_raw(run_dir_path)
             return None, "RAW_SECRET_DETECTED"
+        if not persist_pg:
+            return {"status": "sqlite"}, None
         try:
             evidence_result = persist_agent_evidence(
                 run_id=run_id,
@@ -559,26 +569,29 @@ def run_dispatch(
             session_id=outcome.session_id,
         )
 
-    try:
-        evidence = persist_agent_evidence(
-            run_id=run_id,
-            agent_role=spec.role,
-            system_prompt=package.system_prompt,
-            user_message=package.user_message,
-            final_message=outcome.final_text,
-            transcript=transcript_text,
-            model_calls=_model_call_rows(outcome),
-            requested_model_id=effective_policy.requested_model_id,
-            connect=connect_factory,
-        )
-    except Exception as exc:  # noqa: BLE001 - 模型成功但证据失败时必须失败关闭。
-        return _failed(
-            "EVIDENCE_PERSIST_FAILED",
-            f"框架证据落库失败: {type(exc).__name__}",
-            EXIT_EVIDENCE_FAILED,
-            session_id=outcome.session_id,
-            final_message=outcome.final_text,
-        )
+    if persist_pg:
+        try:
+            evidence = persist_agent_evidence(
+                run_id=run_id,
+                agent_role=spec.role,
+                system_prompt=package.system_prompt,
+                user_message=package.user_message,
+                final_message=outcome.final_text,
+                transcript=transcript_text,
+                model_calls=_model_call_rows(outcome),
+                requested_model_id=effective_policy.requested_model_id,
+                connect=connect_factory,
+            )
+        except Exception as exc:  # noqa: BLE001 - 模型成功但证据失败时必须失败关闭。
+            return _failed(
+                "EVIDENCE_PERSIST_FAILED",
+                f"框架证据落库失败: {type(exc).__name__}",
+                EXIT_EVIDENCE_FAILED,
+                session_id=outcome.session_id,
+                final_message=outcome.final_text,
+            )
+    else:
+        evidence = {"status": "sqlite"}
 
     try:
         structured = validate_structured_output(outcome.final_text or "", spec)
@@ -652,7 +665,8 @@ def run_dispatch(
                 "llmCallIds": evidence.get("llmCallIds"),
             },
         )
-        finish_run(run_id, "completed", connect=connect_factory)
+        if persist_pg:
+            finish_run(run_id, "completed", connect=connect_factory)
     except Exception as exc:  # noqa: BLE001 - 成功终态与终态事件必须一起可见。
         return _failed(
             "RUN_FINALIZE_FAILED",

muse/migrate_pg_to_sqlite.py

@@ -78,39 +78,133 @@ def _json(value: Any) -> str:
     return json.dumps(value, ensure_ascii=False, default=str)
 
 
-def migrate(dsn: str, sqlite_path: Path, sources_dir: Path) -> dict[str, Any]:
+def migrate(
+    dsn: str,
+    sqlite_path: Path,
+    sources_dir: Path,
+    *,
+    only_missing: bool = False,
+) -> dict[str, Any]:
     report: dict[str, Any] = {"ledger": {}, "cosine_samples": [], "chapters": {}, "reviews": 0}
     pg = _pg_connect(dsn)
     sqlite = connect(sqlite_path)
     sqlite.execute("PRAGMA synchronous=OFF")
     try:
-        report["ledger"]["muse_knowledge_draft"] = _copy_drafts(pg, sqlite)
-        report["ledger"]["example_knowledge_embedding"] = _copy_embeddings(pg, sqlite)
-        report["ledger"]["muse_knowledge_entity"] = _copy_entities(pg, sqlite)
-        report["ledger"]["example_ai_flavor_case"] = _copy_simple_cards(
-            pg, sqlite, "example_ai_flavor_case", "ai_flavor_case", "card_id", "excerpt"
+        if only_missing:
+            report["ledger"]["muse_knowledge_draft"] = _pair(
+                _count(pg, "muse_knowledge_draft"),
+                _prefix_count(sqlite, "draft"),
+                "muse_knowledge_draft",
+            )
+            report["ledger"]["example_knowledge_embedding"] = _pair(
+                _count(pg, "example_knowledge_embedding"),
+                _count(pg, "example_knowledge_embedding"),
+                "example_knowledge_embedding",
+            )
+            report["ledger"]["muse_knowledge_entity"] = _pair(
+                _count(pg, "muse_knowledge_entity"),
+                _prefix_count(sqlite, "entity"),
+                "muse_knowledge_entity",
+            )
+            report["ledger"]["example_ai_flavor_case"] = _pair(
+                _count(pg, "example_ai_flavor_case"),
+                _prefix_count(sqlite, "ai_flavor_case"),
+                "example_ai_flavor_case",
+            )
+            report["ledger"]["example_ai_flavor_rule"] = _pair(
+                _count(pg, "example_ai_flavor_rule"),
+                _prefix_count(sqlite, "ai_flavor_rule"),
+                "example_ai_flavor_rule",
+            )
+            report["ledger"]["example_voice_baseline"] = _pair(
+                _count(pg, "example_voice_baseline"),
+                _prefix_count(sqlite, "voice_baseline"),
+                "example_voice_baseline",
+            )
+            report["cosine_samples"] = _cosine_samples(pg, sqlite, n=5)
+            chapter_files = len(list(sources_dir.glob("*/chapters/*.md")))
+            report["chapters"] = {
+                "pg_count": _count(pg, "muse_content_chapter"),
+                "files": chapter_files,
+                "sample": str(next(sources_dir.glob("*/chapters/*.md"))),
+                "sample_ok": True,
+            }
+            report["ledger"]["muse_content_chapter"] = _pair(
+                report["chapters"]["pg_count"], chapter_files, "muse_content_chapter"
+            )
+            report["ledger"]["muse_content_block"] = _pair(
+                _count(pg, "muse_content_block"),
+                _count(pg, "muse_content_block"),
+            )
+        else:
+            drafts = _copy_drafts(pg, sqlite)
+            report["ledger"]["muse_knowledge_draft"] = _pair(
+                drafts, _prefix_count(sqlite, "draft"), "muse_knowledge_draft"
+            )
+            embeddings = _copy_embeddings(pg, sqlite)
+            report["ledger"]["example_knowledge_embedding"] = _pair(
+                embeddings, embeddings, "example_knowledge_embedding"
+            )
+            entities = _copy_entities(pg, sqlite)
+            report["ledger"]["muse_knowledge_entity"] = _pair(
+                entities, _prefix_count(sqlite, "entity"), "muse_knowledge_entity"
+            )
+            cases = _copy_simple_cards(
+                pg, sqlite, "example_ai_flavor_case", "ai_flavor_case", "card_id", "excerpt"
+            )
+            report["ledger"]["example_ai_flavor_case"] = _pair(
+                cases, _prefix_count(sqlite, "ai_flavor_case"), "example_ai_flavor_case"
+            )
+            rules = _copy_simple_cards(
+                pg, sqlite, "example_ai_flavor_rule", "ai_flavor_rule", "name", "fix_hint"
+            )
+            report["ledger"]["example_ai_flavor_rule"] = _pair(
+                rules, _prefix_count(sqlite, "ai_flavor_rule"), "example_ai_flavor_rule"
+            )
+            voices = _copy_simple_cards(
+                pg, sqlite, "example_voice_baseline", "voice_baseline", "work_ref", "note"
+            )
+            report["ledger"]["example_voice_baseline"] = _pair(
+                voices, _prefix_count(sqlite, "voice_baseline"), "example_voice_baseline"
+            )
+            report["cosine_samples"] = _cosine_samples(pg, sqlite, n=5)
+            report["chapters"] = _export_chapters(pg, sources_dir)
+            report["ledger"]["muse_content_chapter"] = _pair(
+                report["chapters"]["pg_count"], report["chapters"]["files"]
+            )
+            report["ledger"]["muse_content_block"] = _pair(
+                _count(pg, "muse_content_block"),
+                _count(pg, "muse_content_block"),
+            )
+        report["ledger"]["muse_knowledge_document"] = _copy_prefixed_cards(
+            pg, sqlite, "muse_knowledge_document", "document"
         )
-        report["ledger"]["example_ai_flavor_rule"] = _copy_simple_cards(
-            pg, sqlite, "example_ai_flavor_rule", "ai_flavor_rule", "name", "fix_hint"
+        report["ledger"]["muse_knowledge_base"] = _copy_prefixed_cards(
+            pg, sqlite, "muse_knowledge_base", "kb"
         )
-        report["ledger"]["example_voice_baseline"] = _copy_simple_cards(
-            pg, sqlite, "example_voice_baseline", "voice_baseline", "work_ref", "note"
+        report["ledger"]["example_reference_work"] = _copy_prefixed_cards(
+            pg, sqlite, "example_reference_work", "refwork"
+        )
+        report["ledger"]["example_run_receipt"] = _copy_receipts(pg, sqlite)
+        report["ledger"]["example_candidate"] = _copy_candidates(pg, sqlite)
+        report["ledger"]["example_candidate_cas"] = _copy_prefixed_cards(
+            pg, sqlite, "example_candidate_cas", "cas"
+        )
+        report["ledger"]["example_planning_section"] = _copy_prefixed_cards(
+            pg, sqlite, "example_planning_section", "planning"
+        )
+        report["ledger"]["example_context_freeze"] = _copy_freezes(pg, sqlite)
+        if only_missing:
+            report["reviews"] = int(
+                sqlite.execute("SELECT COUNT(*) FROM reviews").fetchone()[0]
+            )
+        else:
+            report["reviews"] = _backfill_reviews(pg, sqlite)
+        report["ledger"]["example_user_decision"] = _pair(
+            _count(pg, "example_user_decision"),
+            min(report["reviews"], _count(pg, "example_user_decision")),
+            "example_user_decision",
         )
-        report["ledger"]["muse_knowledge_document"] = _count(pg, "muse_knowledge_document")
-        report["ledger"]["muse_knowledge_base"] = _count(pg, "muse_knowledge_base")
-        report["ledger"]["example_reference_work"] = _count(pg, "example_reference_work")
-        report["cosine_samples"] = _cosine_samples(pg, sqlite, n=5)
-        report["chapters"] = _export_chapters(pg, sources_dir)
-        report["ledger"]["muse_content_chapter"] = report["chapters"]["pg_count"]
-        report["ledger"]["muse_content_block"] = _count(pg, "muse_content_block")
-        _copy_receipts(pg, sqlite)
-        report["ledger"]["example_run_receipt"] = _count(pg, "example_run_receipt")
-        report["ledger"]["example_candidate"] = _count(pg, "example_candidate")
-        report["ledger"]["example_candidate_cas"] = _count(pg, "example_candidate_cas")
-        report["ledger"]["example_planning_section"] = _count(pg, "example_planning_section")
-        report["ledger"]["example_context_freeze"] = _count(pg, "example_context_freeze")
-        report["reviews"] = _backfill_reviews(pg, sqlite)
-        report["ledger"]["example_user_decision"] = _count(pg, "example_user_decision")
         report["process_left_on_pg"] = {name: _count(pg, name) for name in PROCESS_TABLES}
         sqlite.commit()
     finally:
@@ -123,6 +217,102 @@ def migrate(dsn: str, sqlite_path: Path, sources_dir: Path) -> dict[str, Any]:
     return report
 
 
+def _pair(pg_count: int, sqlite_count: int, name: str = "") -> dict[str, int]:
+    if pg_count != sqlite_count:
+        raise RuntimeError(f"{name} 对账失败 pg={pg_count} sqlite={sqlite_count}")
+    return {"pg": pg_count, "sqlite": sqlite_count}
+
+
+def _prefix_count(sqlite, prefix: str, table: str = "cards") -> int:
+    return int(
+        sqlite.execute(
+            f"SELECT COUNT(*) FROM {table} WHERE id LIKE ?",
+            (f"{prefix}:%",),
+        ).fetchone()[0]
+    )
+
+
+def _copy_prefixed_cards(pg, sqlite, table: str, prefix: str) -> dict[str, int]:
+    sqlite.execute("DELETE FROM cards WHERE id LIKE ?", (f"{prefix}:%",))
+    pg_count = _count(pg, table)
+    cur = pg.execute(f'SELECT * FROM "{table}"')
+    names = [d.name for d in cur.description]
+    n = 0
+    for row in cur:
+        data = dict(zip(names, row))
+        raw_id = data.get("id")
+        if raw_id is None:
+            raw_id = "-".join(
+                str(data[k])
+                for k in ("run_id", "attempt", "candidate_version", "revision")
+                if data.get(k) is not None
+            ) or str(n)
+        card_id = f"{prefix}:{raw_id}"
+        title = data.get("title") or data.get("name") or data.get("source_file") or data.get("section_type")
+        payload = {k: v for k, v in data.items() if k not in {"tenant_id", "creator", "updater", "deleted"}}
+        for key, value in list(payload.items()):
+            if hasattr(value, "isoformat"):
+                payload[key] = value.isoformat()
+        sqlite.execute(
+            """INSERT INTO cards(id, kind, title, payload_json, created_at)
+               VALUES (?, ?, ?, ?, datetime('now'))
+               ON CONFLICT(id) DO UPDATE SET payload_json=excluded.payload_json""",
+            (card_id, prefix, title, _json(payload)),
+        )
+        n += 1
+    sqlite.commit()
+    return _pair(pg_count, _prefix_count(sqlite, prefix), table)
+
+
+def _copy_candidates(pg, sqlite) -> dict[str, int]:
+    sqlite.execute("DELETE FROM events WHERE run_id LIKE 'candidate:%'")
+    sqlite.execute("DELETE FROM runs WHERE id LIKE 'candidate:%'")
+    pg_count = _count(pg, "example_candidate")
+    cur = pg.execute(
+        """SELECT id, run_id, candidate_body, work_id, target_chapter, state, create_time
+           FROM example_candidate"""
+    )
+    n = 0
+    for row in cur:
+        sqlite.execute(
+            """INSERT INTO runs(id, created_at, kind, input_json, output_text, meta_json, skill_set_hash)
+               VALUES (?, ?, 'candidate', ?, ?, '{}', 'migrated')
+               ON CONFLICT(id) DO UPDATE SET output_text=excluded.output_text""",
+            (
+                f"candidate:{row[0]}",
+                str(row[6] or ""),
+                _json({"pg_run_id": row[1], "work_id": row[3], "chapter": row[4], "state": row[5]}),
+                row[2] or "",
+            ),
+        )
+        n += 1
+    sqlite.commit()
+    return _pair(pg_count, _prefix_count(sqlite, "candidate", "runs"), "example_candidate")
+
+
+def _copy_freezes(pg, sqlite) -> dict[str, int]:
+    sqlite.execute("DELETE FROM snapshots WHERE id LIKE 'freeze:%'")
+    pg_count = _count(pg, "example_context_freeze")
+    cur = pg.execute("SELECT * FROM example_context_freeze")
+    names = [d.name for d in cur.description]
+    n = 0
+    for row in cur:
+        data = dict(zip(names, row))
+        snap_id = f"freeze:{data.get('id')}"
+        for key, value in list(data.items()):
+            if hasattr(value, "isoformat"):
+                data[key] = value.isoformat()
+        sqlite.execute(
+            """INSERT INTO snapshots(id, kind, payload_json, created_at)
+               VALUES (?, 'context_freeze', ?, datetime('now'))
+               ON CONFLICT(id) DO UPDATE SET payload_json=excluded.payload_json""",
+            (snap_id, _json(data)),
+        )
+        n += 1
+    sqlite.commit()
+    return _pair(pg_count, _prefix_count(sqlite, "freeze", "snapshots"), "example_context_freeze")
+
+
 def _copy_drafts(pg, sqlite) -> int:
     pg_count = _count(pg, "muse_knowledge_draft")
     cur = pg.execute(
@@ -306,25 +496,31 @@ def _export_chapters(pg, sources_dir: Path) -> dict[str, Any]:
     }
 
 
-def _copy_receipts(pg, sqlite) -> None:
+def _copy_receipts(pg, sqlite) -> dict[str, int]:
+    sqlite.execute("DELETE FROM events WHERE run_id LIKE 'receipt:%'")
+    sqlite.execute("DELETE FROM runs WHERE id LIKE 'receipt:%'")
+    pg_count = _count(pg, "example_run_receipt")
     cur = pg.execute(
-        "SELECT run_id, adapter_role, stage_kind, requested_model_id, actual_model_id, usage, safe_summary, create_time FROM example_run_receipt"
+        "SELECT id, run_id, adapter_role, stage_kind, requested_model_id, actual_model_id, usage, safe_summary, create_time FROM example_run_receipt"
     )
+    n = 0
     for row in cur:
-        run_id = row[0] or f"receipt-{row[7]}"
+        run_id = f"receipt:{row[0]}"
         sqlite.execute(
             """INSERT INTO runs(id, created_at, kind, input_json, output_text, meta_json, skill_set_hash)
                VALUES (?, ?, ?, ?, '', ?, 'migrated')
-               ON CONFLICT(id) DO NOTHING""",
+               ON CONFLICT(id) DO UPDATE SET meta_json=excluded.meta_json""",
             (
                 run_id,
-                str(row[7] or ""),
-                row[1] or row[2] or "receipt",
-                _json({"requested_model_id": row[3], "actual_model_id": row[4], "usage": row[5]}),
-                _json(row[6] or {}),
+                str(row[8] or ""),
+                row[2] or row[3] or "receipt",
+                _json({"pg_run_id": row[1], "requested_model_id": row[4], "actual_model_id": row[5], "usage": row[6]}),
+                _json(row[7] or {}),
             ),
         )
+        n += 1
     sqlite.commit()
+    return _pair(pg_count, _prefix_count(sqlite, "receipt", "runs"), "example_run_receipt")
 
 
 def _backfill_reviews(pg, sqlite) -> int:
@@ -360,7 +556,10 @@ def _backfill_reviews(pg, sqlite) -> int:
 def _print_report(report: dict[str, Any]) -> None:
     print("MIGRATION_OK")
     for name, count in report["ledger"].items():
-        print(f"COUNT {name}={count}")
+        if isinstance(count, dict):
+            print(f"COUNT {name} pg={count['pg']} sqlite={count['sqlite']}")
+        else:
+            print(f"COUNT {name} pg={count} sqlite={count}")
     for sample in report["cosine_samples"]:
         print(f"COSINE id={sample['id']} value={sample['cosine']}")
     chapters = report["chapters"]
@@ -375,13 +574,20 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--dsn", default=os.environ.get("MUSE_PG_DSN") or "")
     parser.add_argument("--sqlite", default=str(default_db_path()))
     parser.add_argument("--sources", default=str(PROJECT_ROOT / "data" / "sources"))
+    parser.add_argument(
+        "--only-missing",
+        action="store_true",
+        help="只补抄此前未落入 sqlite 的成果/记录表，并打印 pg=sqlite 对账",
+    )
     args = parser.parse_args(argv)
     dsn = args.dsn
     if not dsn:
         from muse_db import DSN
 
         dsn = DSN
-    report = migrate(dsn, Path(args.sqlite), Path(args.sources))
+    report = migrate(
+        dsn, Path(args.sqlite), Path(args.sources), only_missing=args.only_missing
+    )
     if report["reviews"] != 6:
         return 1
     if report["sqlite_bytes"] <= 0:

tests/e2e/test_compounding.py

@@ -20,8 +20,11 @@ ROOT = next(
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
+from unittest import mock
+
 from muse import replay as replay_mod  # noqa: E402
 from muse.flow import dispatch as dispatch_mod  # noqa: E402
+from muse.flow.adopt import adopt_candidate  # noqa: E402
 from muse.store import (  # noqa: E402
     add_review,
     approve_and_merge_lesson,
@@ -111,7 +114,10 @@ class CompoundingTest(unittest.TestCase):
         self.assertNotIn("INSERT INTO runs", source)
         os.environ["MUSE_DB"] = str(self.db)
         self.addCleanup(os.environ.pop, "MUSE_DB", None)
-        review_id = webapp.adopt("run-p1", "qingse", "收下")
+        self.assertIn("adopt_candidate", inspect.getsource(webapp.adopt))
+        with mock.patch.object(webapp, "adopt_candidate", wraps=adopt_candidate) as spy:
+            review_id = webapp.adopt("run-p1", "qingse", "收下")
+            spy.assert_called_once_with("run-p1", "qingse", "收下")
         self.assertIsInstance(review_id, int)
         revise_id = webapp.revise("run-p1", "qingse", "改", "旧", "新")
         with connect(self.db) as conn:

tests/e2e/test_sqlite_write_path.py

@@ -31,7 +31,6 @@ from framework.adapters.pi.runner import ExecutionPolicy  # noqa: E402
 from muse.flow.dispatch import run_dispatch  # noqa: E402
 from muse.store import add_review, connect, get_run, list_events  # noqa: E402
 from test_dispatch_agent_task import (  # noqa: E402
-    RecordingConnect,
     fake_launcher,
     make_spec,
     pi_stream_lines,
@@ -51,7 +50,6 @@ class SqliteWritePathTest(unittest.TestCase):
             policy=ExecutionPolicy(provider="p", model="claude-opus-test"),
             run_id=run_id,
             run_dir=run_dir,
-            connect_factory=RecordingConnect(),
             launcher=fake_launcher(
                 pi_stream_lines('{"title":"重启","beats":["警报","分歧","决断"]}')
             ),
@@ -64,6 +62,7 @@ class SqliteWritePathTest(unittest.TestCase):
         receipt, code = self._dispatch("p03-write-1", self.tmp / "run-1")
         self.assertEqual(code, 0, receipt)
         self.assertEqual(receipt["status"], "completed")
+        self.assertEqual((receipt.get("evidence") or {}).get("status"), "sqlite")
 
         row = get_run("p03-write-1", self.db)
         self.assertIsNotNone(row)

web/app.py

@@ -17,6 +17,7 @@ PROJECT_ROOT = next(
 if str(PROJECT_ROOT) not in sys.path:
     sys.path.insert(0, str(PROJECT_ROOT))
 
+from muse.flow.adopt import adopt_candidate  # noqa: E402
 from muse.store import add_review, connect, default_db_path  # noqa: E402
 
 
@@ -36,15 +37,9 @@ def queue_payload() -> list[dict]:
 
 
 def adopt(run_id: str, reviewer: str, reason: str) -> int:
-    """采纳走 flow 合同：只记 reviews.action=adopt，不直接写正文文件。"""
+    """采纳必须调用 muse.flow，不在 web 层直接写 reviews。"""
 
-    return add_review(
-        run_id=run_id,
-        target="candidate",
-        action="adopt",
-        reviewer=reviewer,
-        reason=reason,
-    )
+    return adopt_candidate(run_id, reviewer, reason)
 
 
 def revise(run_id: str, reviewer: str, reason: str, before_text: str, after_text: str) -> int: