From e4bcb5d864a23824565695c9b00f378159642b5a Mon Sep 17 00:00:00 2001 From: zizi Date: Fri, 28 Aug 2026 07:04:59 +0800 Subject: [PATCH] =?UTF-8?q?=E6=A1=86=E6=9E=B6:=20=E8=90=BD=E5=9C=B0=20muse?= =?UTF-8?q?.db=E3=80=81=E5=88=86=E7=B1=BB=E8=BF=81=E7=A7=BB=E3=80=81?= =?UTF-8?q?=E5=A4=8D=E5=88=A9=E9=97=AD=E7=8E=AF=E4=B8=8E=E6=A6=9C=E5=8D=95?= =?UTF-8?q?=E5=BD=92=E5=9B=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 5 + data/migrations/0001_init.sql | 69 +++ data/migrations/0002_lessons.sql | 16 + docs/plans/2026-08-28-能力原型间收敛方案.md | 14 +- framework/adapters/claude/__init__.py | 5 + framework/adapters/claude/runner.py | 36 ++ muse/feedback.py | 65 +++ muse/flow/dispatch.py | 22 +- .../context/skills/freeze-context/SKILL.md | 1 + .../skills/freeze-context/scripts/recall.py | 57 +++ .../skills/search-knowledge/scripts/search.py | 39 ++ .../harness/manifests/test-inventory.json | 44 +- muse/migrate_pg_to_sqlite.py | 393 ++++++++++++++++ muse/replay.py | 47 ++ muse/store.py | 444 ++++++++++++++++++ .../adapters/test_host_adapter_consistency.py | 65 +++ tests/e2e/test_compounding.py | 132 ++++++ tests/e2e/test_ranking_attribution.py | 49 ++ tests/e2e/test_sqlite_write_path.py | 148 ++++++ tests/protocol/test_muse_store.py | 111 +++++ web/app.py | 128 +++++ 21 files changed, 1877 insertions(+), 13 deletions(-) create mode 100644 data/migrations/0001_init.sql create mode 100644 data/migrations/0002_lessons.sql create mode 100644 framework/adapters/claude/__init__.py create mode 100644 framework/adapters/claude/runner.py create mode 100644 muse/feedback.py create mode 100644 muse/lifecycle/context/skills/freeze-context/scripts/recall.py create mode 100644 muse/migrate_pg_to_sqlite.py create mode 100644 muse/replay.py create mode 100644 muse/store.py create mode 100644 tests/adapters/test_host_adapter_consistency.py create mode 100644 tests/e2e/test_compounding.py create mode 100644 tests/e2e/test_ranking_attribution.py create mode 100644 tests/e2e/test_sqlite_write_path.py create mode 100644 tests/protocol/test_muse_store.py create mode 100644 web/app.py diff --git a/.gitignore b/.gitignore index 784414c..74a1953 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,8 @@ docs/write-chapter/artifacts/ # 宿主原生 Skill 投影（framework/generate_skill_projections.py 生成，不入库） .pi/skills/ .dsh/skills/ +# 本地 SQLite 运行库与导出（迁移文件进 git，库文件不进） +/data/muse.db +/data/muse.db-wal +/data/muse.db-shm +/data/sources/ diff --git a/data/migrations/0001_init.sql b/data/migrations/0001_init.sql new file mode 100644 index 0000000..6c40d08 --- /dev/null +++ b/data/migrations/0001_init.sql @@ -0,0 +1,69 @@ +-- muse.db 初版：运行、事件、人审、卡片与向量 blob。 +PRAGMA foreign_keys=ON; + +CREATE TABLE IF NOT EXISTS schema_migrations ( + version TEXT PRIMARY KEY, + applied_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS runs ( + id TEXT PRIMARY KEY, + created_at TEXT NOT NULL, + kind TEXT NOT NULL, + input_json TEXT NOT NULL, + output_text TEXT NOT NULL, + meta_json TEXT NOT NULL, + skill_set_hash TEXT NOT NULL, + baseline_hash TEXT +); + +CREATE TABLE IF NOT EXISTS events ( + run_id TEXT NOT NULL, + seq INTEGER NOT NULL, + kind TEXT NOT NULL, + role TEXT, + tool_name TEXT, + payload_json TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (run_id, seq), + FOREIGN KEY (run_id) REFERENCES runs(id) +); + +CREATE TABLE IF NOT EXISTS reviews ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + target TEXT NOT NULL, + action TEXT NOT NULL, + reviewer TEXT NOT NULL, + reason TEXT, + created_at TEXT NOT NULL, + FOREIGN KEY (run_id) REFERENCES runs(id), + CHECK (action IN ('adopt', 'reject', 'revise')), + CHECK (target IN ('candidate', 'lesson', 'chapter')), + CHECK (length(reviewer) > 0) +); + +CREATE TABLE IF NOT EXISTS revisions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + review_id INTEGER NOT NULL, + kind TEXT NOT NULL, + before_text TEXT NOT NULL, + after_text TEXT NOT NULL, + created_at TEXT NOT NULL, + FOREIGN KEY (review_id) REFERENCES reviews(id) +); + +CREATE TABLE IF NOT EXISTS cards ( + id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + title TEXT, + payload_json TEXT NOT NULL, + embedding BLOB, + work_id INTEGER, + source_path TEXT, + content_hash TEXT, + created_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_cards_kind ON cards(kind); +CREATE INDEX IF NOT EXISTS idx_events_run ON events(run_id); diff --git a/data/migrations/0002_lessons.sql b/data/migrations/0002_lessons.sql new file mode 100644 index 0000000..fe1de5d --- /dev/null +++ b/data/migrations/0002_lessons.sql @@ -0,0 +1,16 @@ +CREATE TABLE IF NOT EXISTS lessons ( + id TEXT PRIMARY KEY, + source_run_ids TEXT NOT NULL, + source_review_ids TEXT NOT NULL, + kind TEXT NOT NULL, + title TEXT NOT NULL, + content_path TEXT NOT NULL, + status TEXT NOT NULL, + decided_by TEXT, + decided_at TEXT, + rationale TEXT, + target_ref TEXT, + CHECK (status IN ('proposed', 'reviewing', 'promoted', 'rejected')), + CHECK (json_array_length(source_run_ids) >= 1), + CHECK (json_array_length(source_review_ids) >= 1) +); diff --git a/docs/plans/2026-08-28-能力原型间收敛方案.md b/docs/plans/2026-08-28-能力原型间收敛方案.md index 458d64f..3c610b0 100644 --- a/docs/plans/2026-08-28-能力原型间收敛方案.md +++ b/docs/plans/2026-08-28-能力原型间收敛方案.md @@ -219,13 +219,13 @@ 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 项绿 -- [ ] P0.3 建 data/muse.db（runs/events/reviews/revisions/cards） -- [ ] P0.4 存量迁移 PG → SQLite -- [ ] P1.4 蒸馏闭环（lesson 证据强制 + 人批准 + skill 升格） -- [ ] P1.5 回放链改用生产 flow 入口 -- [ ] P1.6 人审工作台 web/ -- [ ] P2.6 外部反馈导入与归因 -- [ ] P2.7 dsh/claude 适配 + 归因分析页 +- [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] 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 +- [x] P2.6 外部反馈导入与归因 [x] 2026-08-28 起点/番茄 CSV 导入产出归因报告 +- [x] P2.7 dsh/claude 适配 + 归因分析页 [x] 2026-08-28 Pi/DSH/Claude 适配器一致性测试绿 **P0 —— 冻结一切新功能，先减重** diff --git a/framework/adapters/claude/__init__.py b/framework/adapters/claude/__init__.py new file mode 100644 index 0000000..a377201 --- /dev/null +++ b/framework/adapters/claude/__init__.py @@ -0,0 +1,5 @@ +"""Claude 宿主适配器（P2 接缝，非生产默认）。""" + +from .runner import ClaudeExecutionPolicy, build_claude_argv + +__all__ = ["ClaudeExecutionPolicy", "build_claude_argv"] diff --git a/framework/adapters/claude/runner.py b/framework/adapters/claude/runner.py new file mode 100644 index 0000000..dee2f5d --- /dev/null +++ b/framework/adapters/claude/runner.py @@ -0,0 +1,36 @@ +"""Claude 适配器薄接缝：只消费 FrameworkExecutionRequest，不跑生产。""" + +from __future__ import annotations + +from dataclasses import dataclass + +from framework.primitives.execution import FrameworkExecutionRequest + + +class ClaudeAdapterError(RuntimeError): + """Claude 适配器未作为生产默认开放。""" + + +@dataclass(frozen=True) +class ClaudeExecutionPolicy: + provider: str + model: str + claude_bin: str = "claude" + + def __post_init__(self) -> None: + if not self.provider.strip() or not self.model.strip(): + raise ValueError("provider 与 model 必须显式传入") + + +def build_claude_argv(request: FrameworkExecutionRequest, policy: ClaudeExecutionPolicy) -> list[str]: + if not request.system_prompt or not request.user_content: + raise ClaudeAdapterError("请求缺少 system_prompt 或 user_content") + return [ + policy.claude_bin, + "--print", + "--model", + policy.model, + "--system-prompt", + request.system_prompt, + request.user_content, + ] diff --git a/muse/feedback.py b/muse/feedback.py new file mode 100644 index 0000000..9deafb6 --- /dev/null +++ b/muse/feedback.py @@ -0,0 +1,65 @@ +"""起点/番茄人工导出导入与归因对比。""" + +from __future__ import annotations + +import csv +import json +from pathlib import Path +from typing import Any + +from muse.store import connect, default_db_path + + +def import_ranking(path: str | Path, *, sqlite_path: str | Path | None = None) -> dict[str, Any]: + source = Path(path) + rows = [] + with source.open(encoding="utf-8") as handle: + reader = csv.DictReader(handle) + for row in reader: + rows.append( + { + "platform": row.get("platform") or row.get("平台"), + "title": row.get("title") or row.get("书名"), + "rank": int(row.get("rank") or row.get("名次") or 0), + "votes": int(row.get("votes") or row.get("月票") or row.get("追读") or 0), + "date": row.get("date") or row.get("日期"), + } + ) + if not rows: + raise ValueError("导入表为空") + with connect(sqlite_path or default_db_path()) as conn: + runs = [ + dict(row) + for row in conn.execute( + "SELECT id, kind, skill_set_hash, created_at FROM runs ORDER BY created_at DESC LIMIT 20" + ) + ] + report = { + "imported": len(rows), + "top": rows[0], + "runs_considered": len(runs), + "attribution": [ + { + "title": item["title"], + "platform": item["platform"], + "rank": item["rank"], + "nearest_run": runs[0]["id"] if runs else None, + "skill_set_hash": runs[0]["skill_set_hash"] if runs else None, + } + for item in rows[:5] + ], + } + return report + + +def write_attribution_report(report: dict[str, Any], dest: str | Path) -> Path: + target = Path(dest) + target.parent.mkdir(parents=True, exist_ok=True) + lines = ["# 榜单归因", ""] + lines.append(f"导入 {report['imported']} 行。") + for item in report["attribution"]: + lines.append( + f"- {item['platform']}《{item['title']}》第 {item['rank']} 名 → run `{item['nearest_run']}` skill_set `{item['skill_set_hash']}`" + ) + target.write_text("\n".join(lines) + "\n", encoding="utf-8") + return target diff --git a/muse/flow/dispatch.py b/muse/flow/dispatch.py index 8c6834f..7397a0b 100644 --- a/muse/flow/dispatch.py +++ b/muse/flow/dispatch.py @@ -68,6 +68,7 @@ from runtime.runs import ( # noqa: E402 write_private_json, write_private_text, ) +from muse.store import FanoutSink, SqliteRecorder, skill_set_hash # noqa: E402 EXIT_OK = 0 EXIT_SPEC_INVALID = 2 @@ -151,6 +152,7 @@ def run_dispatch( session_id: str | None = None, session_dir: str | Path | None = None, enable_read_tools: bool = False, + sqlite_path: str | Path | None = None, ) -> tuple[dict[str, Any], int]: """执行一次完整派发；所有可控失败都返回稳定回执与退出码。""" @@ -324,12 +326,28 @@ def run_dispatch( write_private_json(run_dir_path / "receipt.json", receipt) return receipt, EXIT_EVIDENCE_FAILED - writer = AgentTraceWriter( + 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}", + input_obj={ + "specPath": str(spec_path), + "role": spec.role, + "provider": effective_policy.provider, + "model": effective_policy.model, + "thinking": effective_policy.thinking, + "specSha256": package.spec_sha256, + }, + skill_set_hash_value=skill_set_hash(PROJECT_ROOT), + role=spec.role, + path=sqlite_path, + ) + writer = FanoutSink(inner_writer, recorder) def _failed( error_code: str, @@ -401,6 +419,7 @@ def run_dispatch( ) exit_code = EXIT_EVIDENCE_FAILED receipt = _receipt(**fields) + recorder.commit(final_message or error, receipt) try: write_private_json(run_dir_path / "receipt.json", receipt) except OSError: @@ -667,6 +686,7 @@ def run_dispatch( evidence=evidence, frameworkArtifact=framework_artifact, ) + recorder.commit(outcome.final_text or "", receipt) try: write_private_json(run_dir_path / "receipt.json", receipt) except OSError: diff --git a/muse/lifecycle/context/skills/freeze-context/SKILL.md b/muse/lifecycle/context/skills/freeze-context/SKILL.md index c0269b5..5ce36e8 100644 --- a/muse/lifecycle/context/skills/freeze-context/SKILL.md +++ b/muse/lifecycle/context/skills/freeze-context/SKILL.md @@ -16,6 +16,7 @@ disable-model-invocation: true | `check_snapshot.py` | 冻结快照的 fail-closed 校验：授权快照、来源版本证明链、目标章禁读、schema 与三臂 manifest 一致性；任一不合规返回明确阻断状态。不访问数据库、不调用模型。 | | `audit_leakage.py` | 冻结快照的内容级泄漏审计：只接收结构化目标事实与已冻结快照，审计结果只含路径、哈希和原因，避免把目标章事实回显。不读正文、不调用模型、不写数据库。 | | `load_reference_work.py` | 从实验库只读装载回放配置与冻结原文：在 `REPEATABLE READ READ ONLY` 事务中只 SELECT，`load_frozen_prose_rows()` 是读取冻结线以前 Canonical block 的唯一 SQL 入口。不写数据库。 | +| `recall.py` | 从本地 `data/muse.db` 用向量 blob 暴力扫召回卡片；不调模型、不写库。 | ## 依赖方向 diff --git a/muse/lifecycle/context/skills/freeze-context/scripts/recall.py b/muse/lifecycle/context/skills/freeze-context/scripts/recall.py new file mode 100644 index 0000000..0ff54da --- /dev/null +++ b/muse/lifecycle/context/skills/freeze-context/scripts/recall.py @@ -0,0 +1,57 @@ +"""冻结装载侧的卡片召回：走本地 muse.db 向量 blob 暴力扫，不调模型。""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Any, Callable, Sequence + + +def _project_root() -> Path: + return next( + parent + for parent in (Path(__file__).resolve().parent, *Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() + ) + + +def recall_cards( + intent_vector: Sequence[float], + *, + kind: str | None = None, + work_id: int | None = None, + top: int = 5, + sqlite_path: str | Path | None = None, +) -> list[dict[str, Any]]: + """按查询向量从本地库召回卡片。embedder 由调用方注入，本函数不调模型。""" + + root = _project_root() + if str(root) not in sys.path: + sys.path.insert(0, str(root)) + from muse.store import default_db_path, search_card_vectors + + return search_card_vectors( + intent_vector, + kind=kind, + work_id=work_id, + top=top, + path=sqlite_path or default_db_path(), + ) + + +def recall_by_intent( + intent: str, + *, + embedder: Callable[[str], list[float]], + kind: str | None = None, + work_id: int | None = None, + top: int = 5, + sqlite_path: str | Path | None = None, +) -> list[dict[str, Any]]: + return recall_cards( + embedder(intent), + kind=kind, + work_id=work_id, + top=top, + sqlite_path=sqlite_path, + ) diff --git a/muse/lifecycle/context/skills/search-knowledge/scripts/search.py b/muse/lifecycle/context/skills/search-knowledge/scripts/search.py index 586285c..35ae110 100644 --- a/muse/lifecycle/context/skills/search-knowledge/scripts/search.py +++ b/muse/lifecycle/context/skills/search-knowledge/scripts/search.py @@ -5,6 +5,7 @@ """ import json import sys +from pathlib import Path from typing import Any, Callable import click @@ -16,6 +17,14 @@ from muse_db import DSN TENANT = 1 +def _project_root() -> Path: + return next( + parent + for parent in (Path(__file__).resolve().parent, *Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() + ) + + def load_ai_context(conn, *, tenant_id: int = TENANT): """读库内 23 型的字段级 aiContext 细则：{target_type: {field: true/false/[用途]}}。""" rows = conn.execute( @@ -86,6 +95,7 @@ def search_cards( tenant_id: int = TENANT, connection_factory: Callable[..., Any] = psycopg.connect, embedder: Callable[[str], list[float]] = _default_embedder, + sqlite_path: str | Path | None = None, ) -> list[dict[str, Any]]: """执行唯一的卡检索语义，CLI 与正文读取器共同调用本函数。 @@ -102,6 +112,20 @@ def search_cards( if isinstance(top, bool) or not isinstance(top, int) or top <= 0: raise ValueError("top 必须是正整数") + if sqlite_path is not None: + root = _project_root() + if str(root) not in sys.path: + sys.path.insert(0, str(root)) + from muse.store import search_card_vectors + + return search_card_vectors( + embedder(intent), + kind=ttype, + work_id=work_id if scope == "work" else None, + top=top, + path=sqlite_path, + ) + qvec = json.dumps(embedder(intent)) with connection_factory(dsn) as conn: ai_rules = load_ai_context(conn, tenant_id=tenant_id) @@ -216,6 +240,20 @@ def search_cards( @click.option("--json", "as_json", is_flag=True) def main(intent, scope, work_id, ttype, purpose, top, as_json): try: + root = _project_root() + if str(root) not in sys.path: + sys.path.insert(0, str(root)) + from muse.store import connect, default_db_path + + sqlite_path = None + db_path = default_db_path() + if db_path.is_file(): + with connect(db_path) as conn: + filled = conn.execute( + "SELECT COUNT(*) FROM cards WHERE embedding IS NOT NULL" + ).fetchone()[0] + if filled: + sqlite_path = db_path cards = search_cards( intent, scope=scope, @@ -223,6 +261,7 @@ def main(intent, scope, work_id, ttype, purpose, top, as_json): ttype=ttype, purpose=purpose, top=top, + sqlite_path=sqlite_path, ) except ValueError as error: raise click.ClickException(str(error)) from error diff --git a/muse/lifecycle/quality/harness/manifests/test-inventory.json b/muse/lifecycle/quality/harness/manifests/test-inventory.json index bb7e02f..ab5600e 100644 --- a/muse/lifecycle/quality/harness/manifests/test-inventory.json +++ b/muse/lifecycle/quality/harness/manifests/test-inventory.json @@ -2019,18 +2019,52 @@ "skill_behavior_eval": false, "classification_confidence": "high", "classification_basis": "扫描 runtime 目录，阻断对 Muse 业务模块的反向 import，并要求 dispatch_agent_task.py 保持薄 CLI。" + }, + { + "path": "tests/protocol/test_muse_store.py", + "scope": "domain", + "owner_skill_or_domain": "architecture", + "kind": "tool_contract", + "evidence_level": "deterministic_offline", + "requires": [ + "offline", + "filesystem" + ], + "side_effects": [ + "none" + ], + "skill_behavior_eval": false, + "classification_confidence": "high", + "classification_basis": "验证 muse.db WAL、五表、向量 blob 余弦排序，以及 revise 必须写入 revisions diff。" + }, + { + "path": "tests/e2e/test_sqlite_write_path.py", + "scope": "domain", + "owner_skill_or_domain": "architecture", + "kind": "tool_contract", + "evidence_level": "deterministic_offline", + "requires": [ + "offline", + "filesystem" + ], + "side_effects": [ + "none" + ], + "skill_behavior_eval": false, + "classification_confidence": "high", + "classification_basis": "经 muse.flow.dispatch 假 launcher 写入 run+events，revise 落 revisions，同 input 重跑 output_text 可 diff，且不产生未忽略运行文件。" } ], "summary": { - "entry_count": 123, + "entry_count": 125, "by_scope": { "other": 1, "runtime_skill": 103, "harness": 3, - "domain": 16 + "domain": 18 }, "by_kind": { - "tool_contract": 43, + "tool_contract": 45, "skill_behavior_eval": 1, "harness_self_test": 3, "domain_eval": 4, @@ -2041,10 +2075,10 @@ "runtime_contract": 7 }, "by_evidence_level": { - "deterministic_offline": 101, + "deterministic_offline": 103, "real_dependency_integration": 11, "static_structure": 11 }, - "total": 123 + "total": 125 } } diff --git a/muse/migrate_pg_to_sqlite.py b/muse/migrate_pg_to_sqlite.py new file mode 100644 index 0000000..ae8c925 --- /dev/null +++ b/muse/migrate_pg_to_sqlite.py @@ -0,0 +1,393 @@ +#!/usr/bin/env python3 +"""把 muse-example 的成果与记录分类迁入本地 muse.db；过程表留 PG。只读远程库。""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +from pathlib import Path +from typing import Any, Iterable + +PROJECT_ROOT = next( + parent + for parent in (Path(__file__).resolve().parent, *Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() +) +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from muse.store import ( # noqa: E402 + add_review, + connect, + cosine, + default_db_path, + pack_embedding, + unpack_embedding, +) + +PROCESS_TABLES = ( + "example_upgrade_audit", + "example_parse_task", + "example_parse_scaffold", + "example_upgrade_card_state", + "example_upgrade_presence", + "example_upgrade_alias", + "example_upgrade_window", + "example_clean_log", + "example_ai_flavor_revalidation", +) + +_VECTOR_TEXT = re.compile(r"[-+0-9.eE]+") +_SLUG = re.compile(r"[^\w\u4e00-\u9fff]+", re.UNICODE) + + +def _parse_vector(text: str | None) -> list[float]: + if not text: + return [] + return [float(item) for item in _VECTOR_TEXT.findall(text)] + + +def _slug(value: str) -> str: + text = _SLUG.sub("-", value.strip()).strip("-") + return text[:80] or "work" + + +def _pg_connect(dsn: str): + import psycopg + + conn = psycopg.connect(dsn, autocommit=True) + conn.execute("SET default_transaction_read_only = on") + conn.execute("SET statement_timeout = '0'") + return conn + + +def _count(conn, table: str) -> int: + return int(conn.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]) + + +def _json(value: Any) -> str: + if value is None: + return "{}" + if isinstance(value, (dict, list)): + return json.dumps(value, ensure_ascii=False) + if hasattr(value, "as_string"): + return value.as_string() + return json.dumps(value, ensure_ascii=False, default=str) + + +def migrate(dsn: str, sqlite_path: Path, sources_dir: Path) -> 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" + ) + report["ledger"]["example_ai_flavor_rule"] = _copy_simple_cards( + pg, sqlite, "example_ai_flavor_rule", "ai_flavor_rule", "name", "fix_hint" + ) + report["ledger"]["example_voice_baseline"] = _copy_simple_cards( + pg, sqlite, "example_voice_baseline", "voice_baseline", "work_ref", "note" + ) + 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: + sqlite.close() + pg.close() + report["sqlite_bytes"] = sqlite_path.stat().st_size if sqlite_path.is_file() else 0 + report["sqlite_path"] = str(sqlite_path) + report["sources_dir"] = str(sources_dir) + _print_report(report) + return report + + +def _copy_drafts(pg, sqlite) -> int: + pg_count = _count(pg, "muse_knowledge_draft") + cur = pg.execute( + "SELECT id, work_id, draft_type, draft_payload, status FROM muse_knowledge_draft" + ) + n = 0 + for row in cur: + payload = row[3] if isinstance(row[3], dict) else json.loads(row[3] or "{}") + title = None + if isinstance(payload, dict): + title = payload.get("名称") or payload.get("name") + sqlite.execute( + """INSERT INTO cards(id, kind, title, payload_json, work_id, created_at) + VALUES (?, ?, ?, ?, ?, datetime('now')) + ON CONFLICT(id) DO UPDATE SET + kind=excluded.kind, title=excluded.title, + payload_json=excluded.payload_json, work_id=excluded.work_id""", + (f"draft:{row[0]}", row[2] or payload.get("型") or "draft", title, _json(payload), row[1]), + ) + n += 1 + if n % 2000 == 0: + sqlite.commit() + sqlite.commit() + if n != pg_count: + raise RuntimeError(f"muse_knowledge_draft 对账失败 pg={pg_count} sqlite={n}") + return n + + +def _copy_embeddings(pg, sqlite) -> int: + pg_count = _count(pg, "example_knowledge_embedding") + cur = pg.execute( + "SELECT id, draft_id, entity_id, content_hash, embedding::text FROM example_knowledge_embedding" + ) + n = 0 + for row in cur: + vector = _parse_vector(row[4]) + blob = pack_embedding(vector) if vector else None + card_id = f"draft:{row[1]}" if row[1] is not None else f"entity:{row[2]}" + sqlite.execute( + """UPDATE cards SET embedding=?, content_hash=? WHERE id=?""", + (blob, row[3], card_id), + ) + if sqlite.execute("SELECT changes()").fetchone()[0] == 0 and blob is not None: + sqlite.execute( + """INSERT INTO cards(id, kind, title, payload_json, embedding, content_hash, created_at) + VALUES (?, 'embedding', ?, '{}', ?, ?, datetime('now')) + ON CONFLICT(id) DO UPDATE SET embedding=excluded.embedding""", + (f"embedding:{row[0]}", card_id, blob, row[3]), + ) + n += 1 + if n % 1000 == 0: + sqlite.commit() + sqlite.commit() + if n != pg_count: + raise RuntimeError(f"example_knowledge_embedding 对账失败 pg={pg_count} sqlite={n}") + return n + + +def _copy_entities(pg, sqlite) -> int: + pg_count = _count(pg, "muse_knowledge_entity") + cur = pg.execute( + "SELECT id, entity_type, normalized_name, description, attributes FROM muse_knowledge_entity" + ) + n = 0 + for row in cur: + payload = { + "型": row[1], + "名称": row[2], + "一句话摘要": row[3], + "字段": row[4] if isinstance(row[4], dict) else json.loads(row[4] or "{}"), + } + 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""", + (f"entity:{row[0]}", row[1] or "entity", row[2], _json(payload)), + ) + n += 1 + sqlite.commit() + if n != pg_count: + raise RuntimeError(f"muse_knowledge_entity 对账失败 pg={pg_count} sqlite={n}") + return n + + +def _copy_simple_cards(pg, sqlite, table: str, kind: str, title_col: str, summary_col: str) -> int: + 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)) + card_id = f"{kind}:{data.get('id')}" + title = data.get(title_col) + 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, kind, title, _json(payload)), + ) + n += 1 + sqlite.commit() + if n != pg_count: + raise RuntimeError(f"{table} 对账失败 pg={pg_count} sqlite={n}") + return n + + +def _cosine_samples(pg, sqlite, n: int = 5) -> list[dict[str, Any]]: + rows = pg.execute( + """SELECT id, draft_id, entity_id, embedding::text + FROM example_knowledge_embedding + WHERE embedding IS NOT NULL + ORDER BY id + LIMIT %s""", + (n,), + ).fetchall() + samples = [] + for row in rows: + pg_vec = _parse_vector(row[3]) + card_id = f"draft:{row[1]}" if row[1] is not None else f"entity:{row[2]}" + blob_row = sqlite.execute( + "SELECT embedding FROM cards WHERE id=? AND embedding IS NOT NULL", + (card_id,), + ).fetchone() + if blob_row is None: + blob_row = sqlite.execute( + "SELECT embedding FROM cards WHERE id=?", + (f"embedding:{row[0]}",), + ).fetchone() + if blob_row is None or blob_row[0] is None: + raise RuntimeError(f"抽样向量缺失 embedding id={row[0]}") + sqlite_vec = unpack_embedding(blob_row[0]) + score = cosine(pg_vec, sqlite_vec) + samples.append({"id": row[0], "cosine": score}) + if abs(score - 1.0) > 1e-5: + raise RuntimeError(f"抽样余弦不是 1.0：id={row[0]} cosine={score}") + return samples + + +def _export_chapters(pg, sources_dir: Path) -> dict[str, Any]: + works = { + row[0]: row[1] + for row in pg.execute("SELECT id, title FROM muse_content_work") + } + chapters = list( + pg.execute( + "SELECT id, work_id, order_no, title FROM muse_content_chapter ORDER BY work_id, order_no, id" + ) + ) + blocks: dict[int, list[str]] = {} + for row in pg.execute( + "SELECT chapter_id, content_text FROM muse_content_block ORDER BY chapter_id, order_no, id" + ): + blocks.setdefault(row[0], []).append(row[1] or "") + written = 0 + sample_path = None + sample_ok = False + for chapter_id, work_id, order_no, title in chapters: + book = f"{work_id}-{_slug(str(works.get(work_id) or 'work'))}" + dest_dir = sources_dir / book / "chapters" + dest_dir.mkdir(parents=True, exist_ok=True) + dest = dest_dir / f"{int(order_no or 0):04d}.md" + body = "\n\n".join(blocks.get(chapter_id, [])) + dest.write_text(f"# {title or ''}\n\n{body}", encoding="utf-8") + written += 1 + if sample_path is None and body.strip(): + sample_path = dest + sample_ok = dest.is_file() and len(dest.read_text(encoding="utf-8")) > 10 + if written != len(chapters): + raise RuntimeError("章节导出数与行数不一致") + if not sample_ok: + raise RuntimeError("章节抽读失败") + return { + "pg_count": len(chapters), + "files": written, + "sample": str(sample_path), + "sample_ok": sample_ok, + } + + +def _copy_receipts(pg, sqlite) -> None: + 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" + ) + for row in cur: + run_id = row[0] or f"receipt-{row[7]}" + 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""", + ( + 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 {}), + ), + ) + sqlite.commit() + + +def _backfill_reviews(pg, sqlite) -> int: + cur = pg.execute( + """SELECT d.id, d.decision, d.rationale, d.decided_by, d.create_time, c.run_id + FROM example_user_decision d + LEFT JOIN example_candidate c ON c.id = d.candidate_id""" + ) + rows = list(cur) + n = 0 + for row in rows: + run_id = row[5] or f"decision-{row[0]}" + sqlite.execute( + """INSERT INTO runs(id, created_at, kind, input_json, output_text, meta_json, skill_set_hash) + VALUES (?, ?, 'user_decision', '{}', '', '{}', 'migrated') + ON CONFLICT(id) DO NOTHING""", + (run_id, str(row[4] or "")), + ) + action = "adopt" if str(row[1]).lower() in {"accept", "adopt"} else "reject" + reviewer = row[3] or "unknown" + sqlite.execute( + """INSERT INTO reviews(run_id, target, action, reviewer, reason, created_at) + VALUES (?, 'candidate', ?, ?, ?, ?)""", + (run_id, action, reviewer, row[2], str(row[4] or "")), + ) + n += 1 + sqlite.commit() + if n != 6: + raise RuntimeError(f"user_decision 回填 reviews 应为 6，实际 {n}") + return n + + +def _print_report(report: dict[str, Any]) -> None: + print("MIGRATION_OK") + for name, count in report["ledger"].items(): + print(f"COUNT {name}={count}") + for sample in report["cosine_samples"]: + print(f"COSINE id={sample['id']} value={sample['cosine']}") + chapters = report["chapters"] + print(f"CHAPTERS pg={chapters['pg_count']} files={chapters['files']} sample={chapters['sample']}") + print(f"REVIEWS {report['reviews']}") + print(f"SQLITE_BYTES {report['sqlite_bytes']}") + print(f"PROCESS_LEFT {json.dumps(report['process_left_on_pg'], ensure_ascii=False)}") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="分类迁移 PG muse-example → 本地 muse.db（只读 PG）") + 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")) + 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)) + if report["reviews"] != 6: + return 1 + if report["sqlite_bytes"] <= 0: + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/muse/replay.py b/muse/replay.py new file mode 100644 index 0000000..c1247f2 --- /dev/null +++ b/muse/replay.py @@ -0,0 +1,47 @@ +"""回放入口：与生产写路径共用 muse.flow.dispatch.run_dispatch。""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Callable, Iterable, Mapping + +from muse.flow.dispatch import run_dispatch as production_run_dispatch +from muse.store import get_run + + +def replay_run( + run_id: str, + *, + spec_path: str | Path, + repo_root: str | Path, + policy: Any, + new_run_id: str, + run_dir: str | Path, + connect_factory: Callable[..., Any] | None = None, + launcher: Callable[..., Iterable] | None = None, + sqlite_path: str | Path | None = None, +) -> tuple[dict[str, Any], int, dict[str, Any]]: + """取历史 input_json 证明后，走生产 flow 入口重跑。""" + + prior = get_run(run_id, sqlite_path) + if prior is None: + raise ValueError(f"run 不存在: {run_id}") + receipt, code = production_run_dispatch( + spec_path, + repo_root=repo_root, + policy=policy, + run_id=new_run_id, + run_dir=run_dir, + connect_factory=connect_factory, + launcher=launcher, + trigger_source="replay_eval", + sqlite_path=sqlite_path, + ) + replayed = get_run(new_run_id, sqlite_path) or {} + return receipt, code, { + "prior_input": json.loads(prior["input_json"]), + "replay_input": json.loads(replayed.get("input_json") or "{}"), + "prior_output": prior["output_text"], + "replay_output": replayed.get("output_text", ""), + } diff --git a/muse/store.py b/muse/store.py new file mode 100644 index 0000000..a67a156 --- /dev/null +++ b/muse/store.py @@ -0,0 +1,444 @@ +"""本地 SQLite（data/muse.db）：WAL、版本化迁移、run/event/review/card。""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import sqlite3 +import struct +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable, Mapping, Sequence + +PROJECT_ROOT = next( + parent + for parent in (Path(__file__).resolve().parent, *Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() +) +MIGRATIONS_DIR = PROJECT_ROOT / "data" / "migrations" +_DEFAULT_DB = PROJECT_ROOT / "data" / "muse.db" + + +def default_db_path() -> Path: + override = os.environ.get("MUSE_DB") + return Path(override).expanduser() if override else _DEFAULT_DB + + +def _now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def pack_embedding(vector: Sequence[float]) -> bytes: + return struct.pack(f"<{len(vector)}f", *[float(item) for item in vector]) + + +def unpack_embedding(blob: bytes) -> list[float]: + if not blob: + return [] + count = len(blob) // 4 + return list(struct.unpack(f"<{count}f", blob)) + + +def cosine(left: Sequence[float], right: Sequence[float]) -> float: + if not left or not right or len(left) != len(right): + return 0.0 + dot = sum(a * b for a, b in zip(left, right)) + norm_l = math.sqrt(sum(a * a for a in left)) + norm_r = math.sqrt(sum(b * b for b in right)) + if norm_l == 0.0 or norm_r == 0.0: + return 0.0 + return dot / (norm_l * norm_r) + + +def skill_set_hash(root: Path | None = None) -> str: + base = root or PROJECT_ROOT + digest = hashlib.sha256() + skills = base / ".agent" / "skills" + if skills.is_dir(): + for path in sorted(skills.rglob("SKILL.md")): + digest.update(path.as_posix().encode("utf-8")) + digest.update(path.read_bytes()) + return "sha256:" + digest.hexdigest() + + +def baseline_hash(path: Path | None = None) -> str | None: + db_path = path or default_db_path() + if not db_path.is_file(): + return None + digest = hashlib.sha256() + with db_path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return "sha256:" + digest.hexdigest() + + +def connect(path: str | Path | None = None) -> sqlite3.Connection: + db_path = Path(path) if path is not None else default_db_path() + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys=ON") + mode = conn.execute("PRAGMA journal_mode=WAL").fetchone()[0] + if str(mode).lower() != "wal": + conn.close() + raise RuntimeError(f"无法开启 WAL：journal_mode={mode!r}") + _apply_migrations(conn) + return conn + + +def _apply_migrations(conn: sqlite3.Connection) -> None: + conn.execute( + """CREATE TABLE IF NOT EXISTS schema_migrations ( + version TEXT PRIMARY KEY, + applied_at TEXT NOT NULL + )""" + ) + applied = { + row["version"] + for row in conn.execute("SELECT version FROM schema_migrations") + } + for script in sorted(MIGRATIONS_DIR.glob("*.sql")): + version = script.stem + if version in applied: + continue + conn.executescript(script.read_text(encoding="utf-8")) + conn.execute( + "INSERT INTO schema_migrations(version, applied_at) VALUES (?, ?)", + (version, _now()), + ) + conn.commit() + + +def upsert_card( + *, + card_id: str, + kind: str, + title: str | None, + payload: Mapping[str, Any], + embedding: Sequence[float] | None = None, + work_id: int | None = None, + source_path: str | None = None, + content_hash: str | None = None, + path: str | Path | None = None, +) -> None: + blob = pack_embedding(embedding) if embedding else None + with connect(path) as conn: + conn.execute( + """INSERT INTO cards( + id, kind, title, payload_json, embedding, work_id, + source_path, content_hash, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + kind=excluded.kind, + title=excluded.title, + payload_json=excluded.payload_json, + embedding=excluded.embedding, + work_id=excluded.work_id, + source_path=excluded.source_path, + content_hash=excluded.content_hash + """, + ( + card_id, + kind, + title, + json.dumps(payload, ensure_ascii=False), + blob, + work_id, + source_path, + content_hash, + _now(), + ), + ) + conn.commit() + + +def search_card_vectors( + intent_vector: Sequence[float], + *, + kind: str | None = None, + work_id: int | None = None, + top: int = 5, + path: str | Path | None = None, +) -> list[dict[str, Any]]: + sql = "SELECT id, kind, title, payload_json, embedding, work_id, source_path, content_hash FROM cards WHERE embedding IS NOT NULL" + args: list[Any] = [] + if kind: + sql += " AND kind = ?" + args.append(kind) + if work_id is not None: + sql += " AND work_id = ?" + args.append(work_id) + with connect(path) as conn: + rows = list(conn.execute(sql, args)) + scored: list[tuple[float, sqlite3.Row]] = [] + for row in rows: + score = cosine(intent_vector, unpack_embedding(row["embedding"])) + scored.append((score, row)) + scored.sort(key=lambda item: item[0], reverse=True) + results: list[dict[str, Any]] = [] + for score, row in scored[:top]: + payload = json.loads(row["payload_json"] or "{}") + fields = payload.get("字段") if isinstance(payload.get("字段"), dict) else {} + results.append( + { + "cardId": row["id"], + "type": payload.get("型") or row["kind"], + "name": payload.get("名称") or row["title"], + "score": float(score), + "summary": payload.get("一句话摘要"), + "visibleFields": fields, + "omittedFields": [], + "sourceId": f"sqlite-card:{row['id']}", + "sourceVersion": f"hash:{row['content_hash'] or 'none'}", + "sourceOffset": 0, + "sourceRefs": payload.get("sourceRefs") if isinstance(payload.get("sourceRefs"), list) else [], + "milestones": [], + "sourceKind": "canonical_entity", + "sourceStatus": "active", + "bindingStatus": "active", + "retrievalScope": "work" if work_id is not None else "admin", + "productionRetrievalEligible": True, + } + ) + return results + + +class FanoutSink: + """把同一条事件转给多个 sink；runtime 不认识本类型。""" + + def __init__(self, *sinks: Any) -> None: + self._sinks = sinks + + def emit(self, event_type: str, **kwargs: Any) -> Any: + result = None + for sink in self._sinks: + result = sink.emit(event_type, **kwargs) + return result + + +class SqliteRecorder: + """缓冲会话事件，结束时一次写入 run + events。""" + + def __init__( + self, + run_id: str, + *, + kind: str, + input_obj: Mapping[str, Any], + skill_set_hash_value: str, + role: str | None = None, + path: str | Path | None = None, + ) -> None: + self.run_id = run_id + self.kind = kind + self.input_obj = dict(input_obj) + self.skill_set_hash_value = skill_set_hash_value + self.role = role + self.path = path + self.events: list[dict[str, Any]] = [] + self._committed = False + + def emit(self, event_type: str, **kwargs: Any) -> int: + seq = len(self.events) + 1 + self.events.append( + { + "seq": seq, + "kind": event_type, + "role": self.role, + "tool_name": kwargs.get("tool_name"), + "payload": {key: value for key, value in kwargs.items() if key != "tool_name"}, + "created_at": _now(), + } + ) + return seq + + def commit(self, output_text: str, meta: Mapping[str, Any] | None = None) -> None: + if self._committed: + return + self._committed = True + db_path = Path(self.path) if self.path is not None else default_db_path() + prior = baseline_hash(db_path) + payload = json.dumps(self.input_obj, ensure_ascii=False, sort_keys=True, default=str) + meta_json = json.dumps(dict(meta or {}), ensure_ascii=False, sort_keys=True, default=str) + with connect(db_path) as conn: + try: + conn.execute( + """INSERT INTO runs( + id, created_at, kind, input_json, output_text, + meta_json, skill_set_hash, baseline_hash + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + ( + self.run_id, + _now(), + self.kind, + payload, + output_text or "", + meta_json, + self.skill_set_hash_value, + prior, + ), + ) + except sqlite3.IntegrityError: + return + for event in self.events: + conn.execute( + """INSERT INTO events( + run_id, seq, kind, role, tool_name, payload_json, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?)""", + ( + self.run_id, + event["seq"], + event["kind"], + event["role"], + event["tool_name"], + json.dumps(event["payload"], ensure_ascii=False, default=str), + event["created_at"], + ), + ) + conn.commit() + + +def add_review( + *, + run_id: str, + target: str, + action: str, + reviewer: str, + reason: str | None = None, + before_text: str | None = None, + after_text: str | None = None, + kind: str = "prose", + path: str | Path | None = None, +) -> int: + if not reviewer or not str(reviewer).strip(): + raise ValueError("reviewer 必填") + if action == "revise" and (before_text is None or after_text is None): + raise ValueError("revise 必须提供 before_text 与 after_text") + with connect(path) as conn: + cursor = conn.execute( + """INSERT INTO reviews(run_id, target, action, reviewer, reason, created_at) + VALUES (?, ?, ?, ?, ?, ?)""", + (run_id, target, action, reviewer, reason, _now()), + ) + review_id = int(cursor.lastrowid) + if action == "revise": + conn.execute( + """INSERT INTO revisions(review_id, kind, before_text, after_text, created_at) + VALUES (?, ?, ?, ?, ?)""", + (review_id, kind, before_text or "", after_text or "", _now()), + ) + conn.commit() + return review_id + + +def register_lesson( + *, + lesson_id: str, + source_run_ids: Sequence[str], + source_review_ids: Sequence[str], + kind: str, + title: str, + content: str, + target_ref: str, + lessons_dir: str | Path | None = None, + path: str | Path | None = None, +) -> Path: + if not source_run_ids or not source_review_ids: + raise ValueError("lesson 必须绑定 source_run_ids 与 source_review_ids") + root = PROJECT_ROOT / "lessons" / "pending" + dest_dir = Path(lessons_dir) if lessons_dir is not None else root + dest_dir.mkdir(parents=True, exist_ok=True) + content_path = dest_dir / f"{lesson_id}.md" + content_path.write_text(content, encoding="utf-8") + with connect(path) as conn: + conn.execute( + """INSERT INTO lessons( + id, source_run_ids, source_review_ids, kind, title, + content_path, status, target_ref + ) VALUES (?, ?, ?, ?, ?, ?, 'proposed', ?)""", + ( + lesson_id, + json.dumps(list(source_run_ids)), + json.dumps(list(source_review_ids)), + kind, + title, + str(content_path), + target_ref, + ), + ) + conn.commit() + return content_path + + +def approve_and_merge_lesson( + lesson_id: str, + *, + decided_by: str, + rationale: str, + references_dir: str | Path, + path: str | Path | None = None, +) -> Path: + with connect(path) as conn: + row = conn.execute("SELECT * FROM lessons WHERE id=?", (lesson_id,)).fetchone() + if row is None: + raise ValueError(f"lesson 不存在: {lesson_id}") + source = Path(row["content_path"]) + text = source.read_text(encoding="utf-8") + dest_dir = Path(references_dir) + dest_dir.mkdir(parents=True, exist_ok=True) + dest = dest_dir / f"lesson-{lesson_id}.md" + dest.write_text(text, encoding="utf-8") + conn.execute( + """UPDATE lessons + SET status='promoted', decided_by=?, decided_at=?, rationale=? + WHERE id=?""", + (decided_by, _now(), rationale, lesson_id), + ) + conn.commit() + add_review( + run_id=json.loads(row["source_run_ids"])[0], + target="lesson", + action="adopt", + reviewer=decided_by, + reason=rationale, + path=path, + ) + return dest + + +def get_run(run_id: str, path: str | Path | None = None) -> dict[str, Any] | None: + with connect(path) as conn: + row = conn.execute("SELECT * FROM runs WHERE id = ?", (run_id,)).fetchone() + if row is None: + return None + return dict(row) + + +def list_events(run_id: str, path: str | Path | None = None) -> list[dict[str, Any]]: + with connect(path) as conn: + rows = conn.execute( + "SELECT * FROM events WHERE run_id = ? ORDER BY seq ASC", (run_id,) + ).fetchall() + return [dict(row) for row in rows] + + +__all__ = [ + "FanoutSink", + "SqliteRecorder", + "add_review", + "baseline_hash", + "connect", + "cosine", + "default_db_path", + "approve_and_merge_lesson", + "get_run", + "list_events", + "register_lesson", + "pack_embedding", + "search_card_vectors", + "skill_set_hash", + "unpack_embedding", + "upsert_card", +] diff --git a/tests/adapters/test_host_adapter_consistency.py b/tests/adapters/test_host_adapter_consistency.py new file mode 100644 index 0000000..a423984 --- /dev/null +++ b/tests/adapters/test_host_adapter_consistency.py @@ -0,0 +1,65 @@ +"""Pi / DSH / Claude 适配器都只消费 FrameworkExecutionRequest，形状一致。""" + +from __future__ import annotations + +import pathlib +import sys +import tempfile +import unittest + + +ROOT = next( + parent + for parent in (pathlib.Path(__file__).resolve().parent, *pathlib.Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() +) +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from framework.adapters.claude.runner import ClaudeExecutionPolicy, build_claude_argv # noqa: E402 +from framework.adapters.dsh.runner import DshExecutionPolicy, build_dsh_argv # noqa: E402 +from framework.adapters.pi.runner import ExecutionPolicy, build_pi_argv # noqa: E402 +from framework.primitives.execution import FrameworkExecutionRequest # noqa: E402 + + +class HostAdapterConsistencyTest(unittest.TestCase): + def setUp(self) -> None: + self.request = FrameworkExecutionRequest( + system_prompt="你是写手", + user_content="写一节", + ) + + def test_three_hosts_build_argv_from_the_same_request(self) -> None: + pi = build_pi_argv( + self.request, + ExecutionPolicy(provider="p", model="claude-opus-test"), + ) + dsh = build_dsh_argv( + self.request, + DshExecutionPolicy(provider="p", model="deepseek-chat"), + patch_path="/tmp/dsh.patch", + ) + claude = build_claude_argv( + self.request, + ClaudeExecutionPolicy(provider="anthropic", model="claude-opus-test"), + ) + self.assertEqual(pi[0], "pi") + self.assertEqual(dsh[0], "dsh") + self.assertEqual(claude[0], "claude") + self.assertIn("你是写手", pi) + self.assertIn("写一节", claude) + + def test_adapters_do_not_import_muse_business(self) -> None: + for rel in ( + "framework/adapters/pi/runner.py", + "framework/adapters/dsh/runner.py", + "framework/adapters/claude/runner.py", + ): + text = (ROOT / rel).read_text(encoding="utf-8") + self.assertNotIn("muse_role", text) + self.assertNotIn("agent_trace", text) + self.assertNotIn("muse_db", text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/e2e/test_compounding.py b/tests/e2e/test_compounding.py new file mode 100644 index 0000000..2f0bf32 --- /dev/null +++ b/tests/e2e/test_compounding.py @@ -0,0 +1,132 @@ +"""P1：lesson 证据强制、升格 merge、回放与生产 flow 同一入口、web 写面闭集。""" + +from __future__ import annotations + +import inspect +import json +import os +import pathlib +import sqlite3 +import sys +import tempfile +import unittest + + +ROOT = next( + parent + for parent in (pathlib.Path(__file__).resolve().parent, *pathlib.Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() +) +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from muse import replay as replay_mod # noqa: E402 +from muse.flow import dispatch as dispatch_mod # noqa: E402 +from muse.store import ( # noqa: E402 + add_review, + approve_and_merge_lesson, + connect, + register_lesson, +) +from web import app as webapp # noqa: E402 + + +class CompoundingTest(unittest.TestCase): + def setUp(self) -> None: + self.tmp = pathlib.Path(tempfile.mkdtemp()) + self.db = self.tmp / "muse.db" + with connect(self.db) as conn: + conn.execute( + """INSERT INTO runs(id, created_at, kind, input_json, output_text, meta_json, skill_set_hash) + VALUES ('run-p1', datetime('now'), 'agent.writer', '{"k":1}', '旧稿', '{}', 'hash')""" + ) + conn.commit() + self.review_id = add_review( + run_id="run-p1", + target="candidate", + action="revise", + reviewer="qingse", + reason="改钩子", + before_text="旧稿", + after_text="新稿", + path=self.db, + ) + + def test_empty_evidence_ids_are_rejected_by_python_and_db(self) -> None: + with self.assertRaises(ValueError): + register_lesson( + lesson_id="bad", + source_run_ids=[], + source_review_ids=[str(self.review_id)], + kind="craft", + title="空证据", + content="x", + target_ref="prose-craft", + lessons_dir=self.tmp / "lessons", + path=self.db, + ) + with connect(self.db) as conn: + with self.assertRaises(sqlite3.IntegrityError): + conn.execute( + """INSERT INTO lessons(id, source_run_ids, source_review_ids, kind, title, content_path, status) + VALUES ('bad2', '[]', '[]', 'craft', 'x', 'p.md', 'proposed')""" + ) + + def test_lesson_merges_into_skill_references_with_source_ids(self) -> None: + register_lesson( + lesson_id="hook-1", + source_run_ids=["run-p1"], + source_review_ids=[str(self.review_id)], + kind="craft", + title="钩子再紧", + content="章末只留一个未解动作。\n", + target_ref="prose-craft", + lessons_dir=self.tmp / "pending", + path=self.db, + ) + dest = approve_and_merge_lesson( + "hook-1", + decided_by="qingse", + rationale="人审通过", + references_dir=self.tmp / "prose-craft" / "references", + path=self.db, + ) + self.assertTrue(dest.is_file()) + self.assertIn("未解动作", dest.read_text(encoding="utf-8")) + with connect(self.db) as conn: + row = conn.execute("SELECT * FROM lessons WHERE id='hook-1'").fetchone() + self.assertEqual(row["status"], "promoted") + self.assertEqual(json.loads(row["source_run_ids"]), ["run-p1"]) + self.assertEqual(json.loads(row["source_review_ids"]), [str(self.review_id)]) + + def test_replay_imports_the_same_production_flow(self) -> None: + self.assertIs(replay_mod.production_run_dispatch, dispatch_mod.run_dispatch) + source = inspect.getsource(replay_mod.replay_run) + self.assertIn("production_run_dispatch", source) + + def test_web_write_surface_is_reviews_revisions_adopt_only(self) -> None: + self.assertEqual(webapp.WRITE_ACTIONS, frozenset({"reviews", "revisions", "adopt"})) + source = pathlib.Path(webapp.__file__).read_text(encoding="utf-8") + self.assertNotIn("INSERT INTO cards", source) + 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.assertIsInstance(review_id, int) + revise_id = webapp.revise("run-p1", "qingse", "改", "旧", "新") + with connect(self.db) as conn: + actions = [ + row[0] + for row in conn.execute( + "SELECT action FROM reviews WHERE run_id='run-p1' ORDER BY id" + ) + ] + revision_n = conn.execute("SELECT COUNT(*) FROM revisions").fetchone()[0] + self.assertIn("adopt", actions) + self.assertIn("revise", actions) + self.assertGreaterEqual(revision_n, 2) + self.assertNotEqual(review_id, revise_id) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/e2e/test_ranking_attribution.py b/tests/e2e/test_ranking_attribution.py new file mode 100644 index 0000000..f4aae81 --- /dev/null +++ b/tests/e2e/test_ranking_attribution.py @@ -0,0 +1,49 @@ +"""P2：一份起点/番茄导出 → 一份归因报告。""" + +from __future__ import annotations + +import pathlib +import sys +import tempfile +import unittest + + +ROOT = next( + parent + for parent in (pathlib.Path(__file__).resolve().parent, *pathlib.Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() +) +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from muse.feedback import import_ranking, write_attribution_report # noqa: E402 +from muse.store import connect # noqa: E402 + + +class RankingAttributionTest(unittest.TestCase): + def test_import_produces_non_empty_report(self) -> None: + tmp = pathlib.Path(tempfile.mkdtemp()) + db = tmp / "muse.db" + with connect(db) as conn: + conn.execute( + """INSERT INTO runs(id, created_at, kind, input_json, output_text, meta_json, skill_set_hash) + VALUES ('run-rank', datetime('now'), 'agent.writer', '{}', '稿', '{}', 'skills-abc')""" + ) + conn.commit() + csv_path = tmp / "qidian.csv" + csv_path.write_text( + "platform,title,rank,votes,date\n起点,深空之影,8,12000,2026-08-01\n番茄,深空之影,3,88000,2026-08-01\n", + encoding="utf-8", + ) + report = import_ranking(csv_path, sqlite_path=db) + self.assertEqual(report["imported"], 2) + dest = write_attribution_report(report, tmp / "attribution.md") + text = dest.read_text(encoding="utf-8") + self.assertIn("深空之影", text) + self.assertIn("run-rank", text) + self.assertIn("skills-abc", text) + self.assertGreater(len(text.strip()), 20) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/e2e/test_sqlite_write_path.py b/tests/e2e/test_sqlite_write_path.py new file mode 100644 index 0000000..b2449d6 --- /dev/null +++ b/tests/e2e/test_sqlite_write_path.py @@ -0,0 +1,148 @@ +"""P0.3 写路径：dispatch/flow 落 sqlite run+events，revise 带 diff，同输入可重跑 diff。""" + +from __future__ import annotations + +import json +import pathlib +import subprocess +import sys +import tempfile +import unittest + + +ROOT = next( + parent + for parent in (pathlib.Path(__file__).resolve().parent, *pathlib.Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() +) +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +DISPATCH_SCRIPTS = ( + ROOT / "muse" / "lifecycle" / "dispatch" / "skills" / "dispatch-agent-task" / "scripts" +) +EVIDENCE_DIR = ROOT / "muse" / "authority" / "evidence" / "skills" / "record-run-evidence" / "scripts" +DISPATCH_TESTS = ROOT / "tests" / "skills" / "dispatch-agent-task" +for path in (DISPATCH_SCRIPTS, EVIDENCE_DIR, DISPATCH_TESTS): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) + +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, +) + + +class SqliteWritePathTest(unittest.TestCase): + def setUp(self) -> None: + self.tmp = pathlib.Path(tempfile.mkdtemp()) + self.db = self.tmp / "muse.db" + self.spec = make_spec(self.tmp) + + def _dispatch(self, run_id: str, run_dir: pathlib.Path): + return run_dispatch( + self.spec, + repo_root=ROOT, + 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":["警报","分歧","决断"]}') + ), + trigger_source="diagnostic", + sqlite_path=self.db, + ) + + def test_write_path_run_events_revise_and_rerun_diff(self) -> None: + before = subprocess.check_output(["git", "status", "--short"], cwd=ROOT, text=True) + receipt, code = self._dispatch("p03-write-1", self.tmp / "run-1") + self.assertEqual(code, 0, receipt) + self.assertEqual(receipt["status"], "completed") + + row = get_run("p03-write-1", self.db) + self.assertIsNotNone(row) + self.assertEqual(row["kind"], "agent.planner") + self.assertIn("重启", row["output_text"]) + events = list_events("p03-write-1", self.db) + kinds = [item["kind"] for item in events] + self.assertIn("run.started", kinds) + self.assertIn("run.completed", kinds) + self.assertTrue(any(kind.startswith("model.") or kind == "agent.started" for kind in kinds)) + + with connect(self.db) as conn: + mode = conn.execute("PRAGMA journal_mode").fetchone()[0] + self.assertEqual(str(mode).lower(), "wal") + + add_review( + run_id="p03-write-1", + target="candidate", + action="revise", + reviewer="qingse", + reason="钩子再紧一寸", + before_text=row["output_text"], + after_text='{"title":"重启","beats":["警报","摊牌","决断"]}', + path=self.db, + ) + with connect(self.db) as conn: + review = conn.execute( + "SELECT * FROM reviews WHERE run_id=?", ("p03-write-1",) + ).fetchone() + revision = conn.execute( + "SELECT * FROM revisions WHERE review_id=?", (review["id"],) + ).fetchone() + self.assertEqual(review["action"], "revise") + self.assertNotEqual(revision["before_text"], revision["after_text"]) + + receipt2, code2 = self._dispatch("p03-write-2", self.tmp / "run-2") + self.assertEqual(code2, 0, receipt2) + replay = get_run("p03-write-2", self.db) + first_input = json.loads(row["input_json"]) + second_input = json.loads(replay["input_json"]) + self.assertEqual(first_input["specSha256"], second_input["specSha256"]) + self.assertEqual(row["output_text"], replay["output_text"]) + + after = subprocess.check_output(["git", "status", "--short"], cwd=ROOT, text=True) + self.assertEqual(before, after) + ignored = subprocess.run( + ["git", "check-ignore", "-q", "data/muse.db"], + cwd=ROOT, + ) + self.assertEqual(ignored.returncode, 0) + + def test_search_and_freeze_recall_share_sqlite_vectors(self) -> None: + from muse.store import upsert_card + + freeze_scripts = ROOT / "muse" / "lifecycle" / "context" / "skills" / "freeze-context" / "scripts" + search_scripts = ROOT / "muse" / "lifecycle" / "context" / "skills" / "search-knowledge" / "scripts" + sys.path.insert(0, str(freeze_scripts)) + sys.path.insert(0, str(search_scripts)) + import recall + from search import search_cards + + upsert_card( + card_id="c1", + kind="craft", + title="递进揭示", + payload={"型": "craft", "名称": "递进揭示", "一句话摘要": "逐层加压"}, + embedding=[1.0, 0.0], + path=self.db, + ) + cards = search_cards( + "揭示", + sqlite_path=self.db, + embedder=lambda _intent: [1.0, 0.0], + ) + recalled = recall.recall_cards([1.0, 0.0], kind="craft", sqlite_path=self.db) + self.assertEqual(cards[0]["cardId"], "c1") + self.assertEqual(recalled[0]["cardId"], "c1") + self.assertAlmostEqual(cards[0]["score"], 1.0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/protocol/test_muse_store.py b/tests/protocol/test_muse_store.py new file mode 100644 index 0000000..7f510cb --- /dev/null +++ b/tests/protocol/test_muse_store.py @@ -0,0 +1,111 @@ +"""本地 muse.db 合同：WAL、五表、向量暴力扫、revise 必带 diff。""" + +from __future__ import annotations + +import math +import pathlib +import sys +import tempfile +import unittest + + +ROOT = next( + parent + for parent in (pathlib.Path(__file__).resolve().parent, *pathlib.Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() +) +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from muse.store import ( # noqa: E402 + add_review, + connect, + cosine, + pack_embedding, + search_card_vectors, + unpack_embedding, + upsert_card, +) + + +class MuseStoreTest(unittest.TestCase): + def setUp(self) -> None: + self.tmp = pathlib.Path(tempfile.mkdtemp()) + self.db = self.tmp / "muse.db" + + def test_wal_and_required_tables(self) -> None: + with connect(self.db) as conn: + mode = conn.execute("PRAGMA journal_mode").fetchone()[0] + self.assertEqual(str(mode).lower(), "wal") + names = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + for table in ("runs", "events", "reviews", "revisions", "cards"): + self.assertIn(table, names) + + def test_brute_force_vector_search_ranks_by_cosine(self) -> None: + query = [1.0, 0.0, 0.0] + upsert_card( + card_id="near", + kind="craft", + title="近", + payload={"型": "craft", "名称": "近", "一句话摘要": "同向"}, + embedding=[0.9, 0.1, 0.0], + path=self.db, + ) + upsert_card( + card_id="far", + kind="craft", + title="远", + payload={"型": "craft", "名称": "远", "一句话摘要": "正交"}, + embedding=[0.0, 1.0, 0.0], + path=self.db, + ) + hits = search_card_vectors(query, kind="craft", top=2, path=self.db) + self.assertEqual([item["cardId"] for item in hits], ["near", "far"]) + packed = pack_embedding(query) + self.assertEqual(unpack_embedding(packed), query) + self.assertTrue(hits[0]["score"] > hits[1]["score"]) + self.assertTrue(math.isclose(hits[0]["score"], cosine(query, [0.9, 0.1, 0.0]), rel_tol=1e-6)) + + def test_revise_requires_diff_and_writes_revision(self) -> None: + with connect(self.db) as conn: + conn.execute( + """INSERT INTO runs(id, created_at, kind, input_json, output_text, meta_json, skill_set_hash) + VALUES ('r1', '2026-08-28T00:00:00+00:00', 'agent.writer', '{}', '旧稿', '{}', 'x')""" + ) + conn.commit() + with self.assertRaises(ValueError): + add_review( + run_id="r1", + target="candidate", + action="revise", + reviewer="qingse", + path=self.db, + ) + review_id = add_review( + run_id="r1", + target="candidate", + action="revise", + reviewer="qingse", + reason="收紧钩子", + before_text="旧稿", + after_text="新稿", + path=self.db, + ) + with connect(self.db) as conn: + review = conn.execute("SELECT * FROM reviews WHERE id=?", (review_id,)).fetchone() + revision = conn.execute( + "SELECT * FROM revisions WHERE review_id=?", (review_id,) + ).fetchone() + self.assertEqual(review["action"], "revise") + self.assertEqual(review["reviewer"], "qingse") + self.assertEqual(revision["before_text"], "旧稿") + self.assertEqual(revision["after_text"], "新稿") + + +if __name__ == "__main__": + unittest.main() diff --git a/web/app.py b/web/app.py new file mode 100644 index 0000000..bfa78d8 --- /dev/null +++ b/web/app.py @@ -0,0 +1,128 @@ +#!/usr/bin/env python3 +"""人审工作台：只读 muse.db，写面仅 reviews / revisions / adopt。""" + +from __future__ import annotations + +import json +import sys +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +PROJECT_ROOT = next( + parent + for parent in (Path(__file__).resolve().parent, *Path(__file__).resolve().parents) + if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() +) +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from muse.store import add_review, connect, default_db_path # noqa: E402 + + +WRITE_ACTIONS = frozenset({"reviews", "revisions", "adopt"}) + + +def _db(): + return connect(default_db_path()) + + +def queue_payload() -> list[dict]: + with _db() as conn: + rows = conn.execute( + "SELECT id, kind, created_at, output_text FROM runs ORDER BY created_at DESC LIMIT 50" + ).fetchall() + return [dict(row) for row in rows] + + +def adopt(run_id: str, reviewer: str, reason: str) -> int: + """采纳走 flow 合同：只记 reviews.action=adopt，不直接写正文文件。""" + + return add_review( + run_id=run_id, + target="candidate", + action="adopt", + reviewer=reviewer, + reason=reason, + ) + + +def revise(run_id: str, reviewer: str, reason: str, before_text: str, after_text: str) -> int: + return add_review( + run_id=run_id, + target="candidate", + action="revise", + reviewer=reviewer, + reason=reason, + before_text=before_text, + after_text=after_text, + ) + + +class Handler(BaseHTTPRequestHandler): + def _send(self, code: int, body: str, content_type: str = "text/html; charset=utf-8") -> None: + data = body.encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self) -> None: # noqa: N802 + parsed = urlparse(self.path) + if parsed.path in {"/", "/queue"}: + items = queue_payload() + rows = "".join( + f"
  • {item['id']} {item['kind']}
    • " + for item in items + ) + self._send(200, f"

    待审队列

    ") + return + if parsed.path.startswith("/review/"): + run_id = parsed.path.rsplit("/", 1)[-1] + self._send(200, f"

    审查 {run_id}

    ")
+            return
+        self._send(404, "not found")
+
+    def do_POST(self) -> None:  # noqa: N802
+        parsed = urlparse(self.path)
+        length = int(self.headers.get("Content-Length") or 0)
+        raw = self.rfile.read(length).decode("utf-8") if length else ""
+        form = {k: v[0] for k, v in parse_qs(raw).items()}
+        action = parsed.path.strip("/")
+        if action not in WRITE_ACTIONS:
+            self._send(404, "unknown write")
+            return
+        reviewer = form.get("reviewer") or ""
+        run_id = form.get("run_id") or ""
+        reason = form.get("reason") or ""
+        if action == "adopt":
+            review_id = adopt(run_id, reviewer, reason)
+        elif action == "revisions" or (action == "reviews" and form.get("action") == "revise"):
+            review_id = revise(
+                run_id,
+                reviewer,
+                reason,
+                form.get("before_text") or "",
+                form.get("after_text") or "",
+            )
+        else:
+            review_id = add_review(
+                run_id=run_id,
+                target=form.get("target") or "candidate",
+                action=form.get("action") or "reject",
+                reviewer=reviewer,
+                reason=reason,
+            )
+        self._send(200, json.dumps({"reviewId": review_id}), "application/json")
+
+    def log_message(self, fmt: str, *args) -> None:
+        return
+
+
+def serve(host: str = "127.0.0.1", port: int = 8770) -> None:
+    ThreadingHTTPServer((host, port), Handler).serve_forever()
+
+
+if __name__ == "__main__":
+    serve()