diff --git a/cloud/chv/app_state.py b/cloud/chv/app_state.py index 8a94354..37da664 100644 --- a/cloud/chv/app_state.py +++ b/cloud/chv/app_state.py @@ -17,7 +17,10 @@ AUTOINCREMENT、无 SQLite 专有语法、参数化查询），换 Postgres = - 库文件只活在宿主 `state_dir//app-state.sqlite`，不在客机 磁盘镜像里，nspawn cell 物理上够不着（测试钉住）。 - 凭证绝不进库：所有写操作过无凭证守卫（键名 + 值模式双重 - 检查）；守卫拒绝时只报键名/模式，不含值。 + 检查）；守卫拒绝时只报键名/模式，不含值。唯一例外是 + round-12 新增的 ``kv_put``（P8 登记文档含用户 prompt， + prompt 里合法出现密钥样文本是调用方的事，旧 JSON 后端 + 同样不过守卫；见 kv_put 的 docstring）。 - audit 表只追加：本模块不提供任何删除/修改审计的方法。 """ @@ -81,6 +84,29 @@ CREATE TABLE IF NOT EXISTS audit ( created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_audit_target ON audit(target, created_at); +-- P12 统一审计轨迹（round-12 迁移目标）：与 JSONL 版同一记录 +-- 形状（seq/hash/prev_hash 链、project_id 隔离），seq 由写入方 +-- 显式给（链是全局的，不按项目分链），不靠自增。 +CREATE TABLE IF NOT EXISTS audit_trail ( + seq INTEGER PRIMARY KEY, + project_id TEXT NOT NULL, + actor TEXT NOT NULL, + category TEXT NOT NULL, + action TEXT NOT NULL, + object_type TEXT NOT NULL, + object_id TEXT NOT NULL, + result TEXT NOT NULL, + reason TEXT NOT NULL, + task_id TEXT NOT NULL, + approval_id TEXT NOT NULL, + detail_json TEXT NOT NULL, + ts REAL NOT NULL, + prev_hash TEXT NOT NULL, + hash TEXT NOT NULL, + legacy INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_audit_trail_project + ON audit_trail(project_id, ts, seq); """ _BACKUP_FILENAME = "app-state.sqlite" @@ -113,6 +139,17 @@ def state_db_path(state_dir: str, vm_name: str) -> str: return os.path.join(state_dir, vm_name, "app-state.sqlite") +def host_state_db_path(state_dir: str) -> str: + """宿主级状态库路径（round-12）：与 per-VM 库同库不同文件。 + + P8 的三处登记与 P12 的审计轨迹是宿主级状态（横跨多个 VM、 + 按 project_id 隔离），不属于任何一个 VM，所以不进 + ``state_db_path(state_dir, vm_name)`` 的 per-VM 库，另落在 + state_dir 根下的 app-state.sqlite。同一个 AppStateStore 类、 + 同一套备份/恢复与 0600 权限口径，只是路径不同。""" + return os.path.join(state_dir, "app-state.sqlite") + + def _utcnow() -> str: """ISO 时间戳（UTC，秒级）。""" return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) @@ -273,12 +310,116 @@ class AppStateStore: ("schema_version",)).fetchone() return row[0] if row else "" + def kv_put(self, namespace: str, key: str, value: str) -> None: + """写一条命名空间 KV（round-12：P8 三处登记的迁移后端）。 + + 键存为 ``kv::``，值原样存 TEXT（调用方 + 自己 JSON 序列化）。命名空间与键不得含 ``:``（分隔符）， + 不得为空。 + + 注意：本方法**不过无凭证守卫**——P8 登记文档里含用户 + 写的子代理 prompt，prompt 里合法出现密钥样文本是调用 + 方的事（旧 JSON 文件后端同样不过守卫）；守卫仍覆盖 + put_task/put_goal/put_session/append_audit 这些写操作。 + KV 的键是代码里写死的固定名，不含用户输入。""" + if not namespace or ":" in namespace \ + or not isinstance(namespace, str): + raise ValueError("kv namespace must be a non-empty string" + " without ':'") + if not key or ":" in key or not isinstance(key, str): + raise ValueError("kv key must be a non-empty string" + " without ':'") + if not isinstance(value, str): + raise ValueError("kv value must be str") + with self._lock: + conn = self._require_conn() + conn.execute( + "INSERT OR REPLACE INTO meta(key, value) VALUES (?, ?)", + ("kv:%s:%s" % (namespace, key), value)) + conn.commit() + + def kv_get(self, namespace: str, key: str) -> str | None: + """读一条命名空间 KV；不存在回 None。""" + with self._lock: + conn = self._require_conn() + row = conn.execute( + "SELECT value FROM meta WHERE key = ?", + ("kv:%s:%s" % (namespace, key),)).fetchone() + return row[0] if row else None + + def kv_delete(self, namespace: str, key: str) -> None: + """删一条命名空间 KV（不存在也静默成功）。""" + with self._lock: + conn = self._require_conn() + conn.execute("DELETE FROM meta WHERE key = ?", + ("kv:%s:%s" % (namespace, key),)) + conn.commit() + + def insert_audit_trail_row(self, record: dict) -> None: + """插一条 P12 审计轨迹行（round-12）：原样存入，不重算 + 链——seq/hash/prev_hash 由调用方（StoreAuditTrail）算好， + 迁移时也原样搬旧行。seq 重复即抛 sqlite3.IntegrityError。""" + detail = record.get("detail") + if not isinstance(detail, dict): + detail = {} + with self._lock: + conn = self._require_conn() + conn.execute( + "INSERT INTO audit_trail(seq, project_id, actor," + " category, action, object_type, object_id, result," + " reason, task_id, approval_id, detail_json, ts," + " prev_hash, hash, legacy) VALUES (?, ?, ?, ?, ?, ?," + " ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (int(record["seq"]), str(record.get("project_id", "")), + str(record.get("actor", "")), + str(record.get("category", "")), + str(record.get("action", "")), + str(record.get("object_type", "")), + str(record.get("object_id", "")), + str(record.get("result", "ok")), + str(record.get("reason", "")), + str(record.get("task_id", "")), + str(record.get("approval_id", "")), + json.dumps(detail, sort_keys=True, + ensure_ascii=False), + float(record.get("ts", 0.0)), + str(record.get("prev_hash", "")), + str(record.get("hash", "")), + 1 if record.get("legacy") else 0)) + conn.commit() + + def list_audit_trail_rows(self) -> list[dict]: + """全部审计轨迹行，按 seq 升序（链顺序），与 JSONL 版 + 同款记录形状；legacy 只在为真时出现。""" + with self._lock: + conn = self._require_conn() + rows = conn.execute( + "SELECT seq, project_id, actor, category, action," + " object_type, object_id, result, reason, task_id," + " approval_id, detail_json, ts, prev_hash, hash," + " legacy FROM audit_trail ORDER BY seq ASC").fetchall() + out = [] + for row in rows: + rec = {"seq": row[0], "ts": row[12], + "project_id": row[1], "actor": row[2], + "category": row[3], "action": row[4], + "object_type": row[5], "object_id": row[6], + "result": row[7], "reason": row[8], + "task_id": row[9], "approval_id": row[10], + "detail": json.loads(row[11]), + "prev_hash": row[13], "hash": row[14]} + if row[15]: + rec["legacy"] = True + out.append(rec) + return out + def counts(self) -> dict[str, int]: """各表行数（备份/恢复校验用）。""" with self._lock: conn = self._require_conn() out: dict[str, int] = {} - for table in ("tasks", "goals", "sessions", "audit"): + for table in ("tasks", "goals", "sessions", "audit", + "audit_trail"): row = conn.execute( "SELECT COUNT(*) FROM %s" % table).fetchone() out[table] = int(row[0]) @@ -469,6 +610,7 @@ class AppStateStore: "goals": str(counts["goals"]), "sessions": str(counts["sessions"]), "audit": str(counts["audit"]), + "audit_trail": str(counts["audit_trail"]), "created_at": _utcnow(), } atomic_write_text( diff --git a/cloud/chv/audit_trail.py b/cloud/chv/audit_trail.py index 6513b69..bad0799 100644 --- a/cloud/chv/audit_trail.py +++ b/cloud/chv/audit_trail.py @@ -34,9 +34,13 @@ VM 内所有文件。 - project_id 强制：``list_records()``/``export_json()`` 必须传 project_id， 只返回该项目记录；跨项目读取返回空，不抛错（与现有 ``list_audit`` 行为一致，不泄露“存在别的项目记录”）。 -- 持久化：先落独立 JSONL 文件（0600），待 P7 状态库就绪后迁移； - ``from_legacy_row()`` 把现有 ``audit_log`` 行转成统一记录 - （标 ``legacy: true``，本身无哈希链，append 时重新入链）。 +- 持久化：round-12 已迁移到 P7 状态库（``StoreAuditTrail``， + 宿主级 AppStateStore 的 ``audit_trail`` 表，同一记录形状与 + 哈希链）；``AuditTrail``（JSONL 0600）保留，用于读旧文件与 + 迁移源，``migrate_jsonl_to_store()`` 原样搬链（坏链拒绝、 + 分叉拒绝、重跑幂等）；``from_legacy_row()`` 把现有 + ``audit_log`` 行转成统一记录（标 ``legacy: true``，本身无 + 哈希链，append 时重新入链）。 """ import hashlib @@ -310,6 +314,168 @@ class AuditTrail: return [line.strip() for line in f if line.strip()] +class StoreAuditTrail: + """AppStateStore 后端的统一审计轨迹（round-12 迁移）。 + + 与 ``AuditTrail``（JSONL）同一记录形状、同一哈希链算法、 + 同一过滤/导出语义；持久化改走宿主级 AppStateStore 的 + ``audit_trail`` 表（``state_dir/app-state.sqlite``）。 + JSONL 版保留，用于读旧文件与迁移源；新写一律走本类。 + + 写失败不静默丢（与 JSONL 版同款：append 抛错由调用方 + 决定是否 best-effort）；库打不开时 __init__ 即抛错，不 + 回空链——审计链丢了不能假装没事（与 P8 登记的 fail-safe + 不同，见差距表）。 + """ + + def __init__(self, state_dir: str) -> None: + if not state_dir or not isinstance(state_dir, str): + raise ValueError("state_dir required") + self.state_dir = state_dir + self._last_hash = GENESIS_HASH + self._seq = 0 + self._load_tail() + + def _open_store(self): + """打开宿主级状态库（每次调用新建连接，用完即关）。""" + from cloud.chv.app_state import AppStateStore, \ + host_state_db_path + store = AppStateStore(host_state_db_path(self.state_dir)) + store.open() + return store + + def _load_tail(self) -> None: + """启动时读库尾，恢复末尾 seq 与链哈希。""" + with self._open_store() as store: + rows = store.list_audit_trail_rows() + if not rows: + return + last = rows[-1] + self._seq = int(last["seq"]) + self._last_hash = str(last["hash"]) + + def append(self, project_id: str, actor: str, category: str, + action: str, object_type: str = "", + object_id: str = "", result: str = "ok", + reason: str = "", task_id: str = "", + approval_id: str = "", + detail: dict | None = None, + ts: float | None = None) -> dict: + """追加一条审计记录，返回该记录（含 seq 与链哈希）。""" + if not project_id: + raise ValueError("project_id 不能为空") + if category not in AUDIT_CATEGORIES: + raise ValueError("未知审计分类：%s" % category) + seq = self._seq + 1 + record = { + "seq": seq, + "ts": ts if ts is not None else time.time(), + "project_id": project_id, + "actor": actor, + "category": category, + "action": action, + "object_type": object_type, + "object_id": object_id, + "result": result, + "reason": reason, + "task_id": task_id, + "approval_id": approval_id, + "detail": sanitize_detail(detail), + "prev_hash": self._last_hash, + } + chain_hash = compute_hash(self._last_hash, record) + record["hash"] = chain_hash + with self._open_store() as store: + store.insert_audit_trail_row(record) + self._seq = seq + self._last_hash = chain_hash + return dict(record) + + def list_records(self, project_id: str, category: str = "", + action: str = "", task_id: str = "", + since: float = 0.0, until: float = 0.0, + limit: int = 100, offset: int = 0) -> list: + """按时间倒序返回该项目的审计记录（副本，可分页过滤）。 + project_id 为空或不匹配一律返回空列表。limit <= 0 + 表示不限制条数。""" + if not project_id: + return [] + with self._open_store() as store: + rows = store.list_audit_trail_rows() + kept = [r for r in rows if r.get("project_id") == project_id] + kept = _filter_rows(kept, category, action, task_id, + since, until) + kept.sort(key=lambda r: (r.get("ts", 0.0), r.get("seq", 0)), + reverse=True) + if offset: + kept = kept[offset:] + if limit > 0: + kept = kept[:limit] + return [dict(r) for r in kept] + + def verify(self) -> tuple[bool, int]: + """校验整条哈希链：返回 (是否通过, 首个坏行下标)。 + 通过时下标为 -1。""" + with self._open_store() as store: + rows = store.list_audit_trail_rows() + expected_prev = GENESIS_HASH + expected_seq = 1 + for index, record in enumerate(rows): + if not _chain_ok(record, expected_prev, expected_seq): + return False, index + expected_prev = record["hash"] + expected_seq += 1 + return True, -1 + + def export_json(self, project_id: str, dest: str) -> int: + """导出该项目全部审计记录为 JSON 数组，返回导出条数。""" + rows = self.list_records(project_id, limit=0) + rows.sort(key=lambda r: (r.get("ts", 0.0), r.get("seq", 0))) + with open(dest, "w", encoding="utf-8") as f: + json.dump(rows, f, ensure_ascii=False, indent=2) + os.chmod(dest, 0o600) + return len(rows) + + +def migrate_jsonl_to_store(path: str, state_dir: str) -> int: + """把 JSONL 轨迹原样搬进 StoreAuditTrail，返回新插入条数。 + + 先校验 JSONL 整条链（坏链即 ValueError，拒绝迁移，不静 + 默丢行）；行按原文 seq/hash/prev_hash 插入，不重算链； + 库里已有且内容相同的 seq 跳过（幂等，可重跑）；已有但 + 内容不同的 seq 即 ValueError（分叉，fail-closed）。 + 文件不存在回 0。""" + if not os.path.exists(path): + return 0 + trail = AuditTrail(path) + ok, bad = trail.verify() + if not ok: + raise ValueError("JSONL 审计链校验失败，拒绝迁移：行 %d" + % bad) + rows = trail._read_all() + from cloud.chv.app_state import AppStateStore, \ + host_state_db_path + store = AppStateStore(host_state_db_path(state_dir)) + store.open() + try: + existing = {int(r["seq"]): r + for r in store.list_audit_trail_rows()} + inserted = 0 + for row in sorted(rows, key=lambda r: int(r.get("seq", 0))): + seq = int(row["seq"]) + if seq in existing: + if existing[seq] != row: + raise ValueError( + "审计 seq %d 在库里已有不同内容，拒绝迁移" + % seq) + continue + store.insert_audit_trail_row(row) + inserted += 1 + return inserted + finally: + store.close() + + def _filter_rows(rows: list, category: str, action: str, task_id: str, since: float, until: float) -> list: """list() 的过滤：分类/动作/任务/时间窗口，全空即不过滤。""" diff --git a/cloud/chv/browser_tasks.py b/cloud/chv/browser_tasks.py index 55951f3..61bc908 100644 --- a/cloud/chv/browser_tasks.py +++ b/cloud/chv/browser_tasks.py @@ -12,8 +12,12 @@ transport 接口：真实 transport 走 P2 的 broker 协议 复制 P2 的文件）。本层只管生命周期、单实例单任务与状态隔离， transport 注入（测试用 fake）。 -存储：state_dir/browser-tasks.json（原子写 + 损坏 quarantine， -orphans 同款口径）；P7 状态存储落地后迁移。 +存储：宿主级 AppStateStore（round-12 已迁移，见 +docs/chv-muse-p8-gap.md §6）：state_dir/app-state.sqlite 的 +``p8-browser-tasks`` 命名空间，一行存整份登记；旧 +browser-tasks.json 首次加载时自动导入并改名 ``.migrated``； +行损坏/库打不开时回空结构（fail-safe，orphans 同款口径）， +下次成功保存即覆盖自愈。 """ from __future__ import annotations @@ -24,12 +28,14 @@ import os import time import uuid -from cloud.chv.atomic_file import atomic_write_text +from cloud.chv.app_state import AppStateStore, host_state_db_path from cloud.chv.compat215 import resolve_inside _LOGGER = logging.getLogger("cloud.cloud_hypervisor") _REGISTRY_FILENAME = "browser-tasks.json" +_KV_NAMESPACE = "p8-browser-tasks" +_KV_KEY = "registry" class BrowserTransport: @@ -51,8 +57,15 @@ class BrowserTransport: raise NotImplementedError +def _open_store(state_dir: str) -> AppStateStore: + """打开宿主级状态库（每次调用新建连接，用完即关）。""" + store = AppStateStore(host_state_db_path(state_dir)) + store.open() + return store + + def _registry_path(state_dir: str) -> str: - """实例登记文件路径。""" + """旧版实例登记文件路径（只用于一次性迁移导入）。""" return os.path.join(state_dir, _REGISTRY_FILENAME) @@ -66,7 +79,35 @@ def _quarantine(path: str) -> None: def _load_registry(state_dir: str) -> dict: - """读登记：无文件回空结构；损坏则 quarantine 后回空结构。""" + """读登记：KV 行缺失时走一次性旧文件导入；行损坏/库打不 + 开时回空结构（fail-safe，orphans 同款口径；下次成功保存 + 即覆盖自愈）。""" + try: + with _open_store(state_dir) as store: + raw = store.kv_get(_KV_NAMESPACE, _KV_KEY) + except Exception: + _LOGGER.warning("chv_browser_tasks_store_unreadable " + "state_dir=%s", state_dir) + return {"instances": {}} + if raw is None: + return _import_legacy(state_dir) + try: + data = json.loads(raw) + except ValueError: + _LOGGER.warning("chv_browser_tasks_corrupt state_dir=%s", + state_dir) + return {"instances": {}} + if not isinstance(data, dict) \ + or not isinstance(data.get("instances"), dict): + return {"instances": {}} + return data + + +def _import_legacy(state_dir: str) -> dict: + """一次性导入旧 browser-tasks.json：合法即进 KV 并把文件 + 改名 ``.migrated``；无文件回空；损坏则 quarantine 后回空 + （旧口径）。导入失败时返回读到的数据、文件保留，下次加 + 载重试。""" path = _registry_path(state_dir) try: with open(path, "r") as fh: @@ -80,15 +121,26 @@ def _load_registry(state_dir: str) -> dict: or not isinstance(data.get("instances"), dict): _quarantine(path) return {"instances": {}} + try: + with _open_store(state_dir) as store: + store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data)) + except Exception: + _LOGGER.warning("chv_browser_tasks_import_failed " + "state_dir=%s", state_dir) + return data + try: + os.replace(path, path + ".migrated") + except OSError: + pass return data def _save_registry(state_dir: str, data: dict) -> None: - """原子落盘登记；写失败只记日志。""" + """落盘登记到 KV 行；写失败只记日志。""" try: - os.makedirs(state_dir, exist_ok=True) - atomic_write_text(_registry_path(state_dir), json.dumps(data)) - except OSError: + with _open_store(state_dir) as store: + store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data)) + except Exception: _LOGGER.warning("chv_browser_tasks_save_failed " "state_dir=%s", state_dir) diff --git a/cloud/chv/presentation.py b/cloud/chv/presentation.py index d2fa8cb..0f1c84f 100644 --- a/cloud/chv/presentation.py +++ b/cloud/chv/presentation.py @@ -12,11 +12,13 @@ - cancelled / 未知终态 -> 只记日志（未知终态 fail-closed， 不通知）。 -账本：state_dir/notification-ledger.json（原子写 + 损坏 -quarantine，orphans 同款口径）；同一任务同一原因在节流窗口 -（3600 秒，本项目自定）内只通知一次。P7 状态存储落地后迁移， -本模块 docstring 与差距表已标注；账本丢了只影响去重（可能 -重复通知一次），不影响任务执行本身。 +账本：宿主级 AppStateStore（round-12 已迁移，见 +docs/chv-muse-p8-gap.md §6）：state_dir/app-state.sqlite 的 +``p8-ledger`` 命名空间，一行存整份账本；旧 +notification-ledger.json 首次加载时自动导入并改名 +``.migrated``；行损坏/库打不开时回空账本（去重信息丢了只会 +导致一次重复通知，不影响任务本身），下次成功保存即覆盖自愈。 +同一任务同一原因在节流窗口（3600 秒，本项目自定）内只通知一次。 """ from __future__ import annotations @@ -26,13 +28,15 @@ import logging import os import time -from cloud.chv.atomic_file import atomic_write_text +from cloud.chv.app_state import AppStateStore, host_state_db_path _LOGGER = logging.getLogger("cloud.cloud_hypervisor") # 同一任务同一原因的通知节流窗口（秒，本项目自定）。 NOTIFY_THROTTLE_SECONDS = 3600 _LEDGER_FILENAME = "notification-ledger.json" +_KV_NAMESPACE = "p8-ledger" +_KV_KEY = "registry" _MAX_LEDGER_ENTRIES = 500 @@ -71,8 +75,15 @@ def _evaluate_completed(result: dict) -> dict: return {"present": False, "reason": "routine_completed"} +def _open_store(state_dir: str) -> AppStateStore: + """打开宿主级状态库（每次调用新建连接，用完即关）。""" + store = AppStateStore(host_state_db_path(state_dir)) + store.open() + return store + + def _ledger_path(state_dir: str) -> str: - """通知账本文件路径。""" + """旧版通知账本文件路径（只用于一次性迁移导入）。""" return os.path.join(state_dir, _LEDGER_FILENAME) @@ -85,8 +96,34 @@ def _quarantine(path: str) -> None: def _load_ledger(state_dir: str) -> dict: - """读账本：无文件回空结构；损坏则 quarantine 后回空结构 - （去重信息丢了只会导致一次重复通知，不影响任务本身）。""" + """读账本：KV 行缺失时走一次性旧文件导入；行损坏/库打不 + 开时回空结构（去重信息丢了只会导致一次重复通知，不影响 + 任务本身；下次成功保存即覆盖自愈）。""" + try: + with _open_store(state_dir) as store: + raw = store.kv_get(_KV_NAMESPACE, _KV_KEY) + except Exception: + _LOGGER.warning("chv_ledger_store_unreadable state_dir=%s", + state_dir) + return {"notified": {}} + if raw is None: + return _import_legacy(state_dir) + try: + data = json.loads(raw) + except ValueError: + _LOGGER.warning("chv_ledger_corrupt state_dir=%s", state_dir) + return {"notified": {}} + if not isinstance(data, dict) \ + or not isinstance(data.get("notified"), dict): + return {"notified": {}} + return data + + +def _import_legacy(state_dir: str) -> dict: + """一次性导入旧 notification-ledger.json：合法即进 KV 并 + 把文件改名 ``.migrated``；无文件回空；损坏则 quarantine + 后回空（旧口径）。导入失败时返回读到的数据、文件保留， + 下次加载重试。""" path = _ledger_path(state_dir) try: with open(path, "r") as fh: @@ -100,19 +137,30 @@ def _load_ledger(state_dir: str) -> dict: or not isinstance(data.get("notified"), dict): _quarantine(path) return {"notified": {}} + try: + with _open_store(state_dir) as store: + store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data)) + except Exception: + _LOGGER.warning("chv_ledger_import_failed state_dir=%s", + state_dir) + return data + try: + os.replace(path, path + ".migrated") + except OSError: + pass return data def _save_ledger(state_dir: str, data: dict) -> None: - """原子落盘账本（截断到上限）；写失败只记日志。""" + """落盘账本到 KV 行（截断到上限）；写失败只记日志。""" notified = data.get("notified", {}) if len(notified) > _MAX_LEDGER_ENTRIES: items = sorted(notified.items(), key=lambda kv: kv[1]) data["notified"] = dict(items[-_MAX_LEDGER_ENTRIES:]) try: - os.makedirs(state_dir, exist_ok=True) - atomic_write_text(_ledger_path(state_dir), json.dumps(data)) - except OSError: + with _open_store(state_dir) as store: + store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data)) + except Exception: _LOGGER.warning("chv_ledger_save_failed state_dir=%s", state_dir) @@ -120,8 +168,8 @@ class NotificationLedger: """通知去重节流账本（每任务每原因节流）。 should_notify(task_id, reason) 在窗口内未通知过时回 True - 并记下这次通知；窗口内已通知过回 False。账本先走原子 - JSON 文件，待迁移到 P7 状态存储（见模块 docstring）。 + 并记下这次通知；窗口内已通知过回 False。账本落在宿主级 + AppStateStore（round-12 已迁移，见模块 docstring）。 """ def __init__(self, state_dir: str) -> None: diff --git a/cloud/chv/swarm.py b/cloud/chv/swarm.py index 0732d90..f800216 100644 --- a/cloud/chv/swarm.py +++ b/cloud/chv/swarm.py @@ -10,9 +10,12 @@ VM 需要足够资源处理并发子代理与 crons。资源上限是本项目 理自己的 VM/单元里（P1 常驻池 + P2 单元是底座），本层只做登 记、派发与审计，不执行工具代码。 -存储：state_dir/swarms.json（原子写 + 损坏 quarantine，orphans -同款口径）；P7 状态存储落地后迁移，本模块 docstring 与差距表 -已标注。 +存储：宿主级 AppStateStore（round-12 已迁移，见 +docs/chv-muse-p8-gap.md §6）：state_dir/app-state.sqlite 的 +``p8-swarm`` 命名空间，一行存整份登记文档（原子写由 SQLite +事务保证）；旧 swarms.json 首次加载时自动导入并改名 +``.migrated``；KV 行损坏/库打不开时回空结构（fail-safe， +orphans 同款口径），下次成功保存即覆盖自愈。 """ from __future__ import annotations @@ -24,7 +27,7 @@ import time import uuid from collections.abc import Callable -from cloud.chv.atomic_file import atomic_write_text +from cloud.chv.app_state import AppStateStore, host_state_db_path from cloud.chv.compat215 import resolve_inside _LOGGER = logging.getLogger("cloud.cloud_hypervisor") @@ -38,10 +41,19 @@ MAX_AUDIT_ENTRIES = 200 MAX_INVOCATION_ENTRIES = 200 AGENT_STATUSES = ("pending", "running", "done", "failed") _REGISTRY_FILENAME = "swarms.json" +_KV_NAMESPACE = "p8-swarm" +_KV_KEY = "registry" + + +def _open_store(state_dir: str) -> AppStateStore: + """打开宿主级状态库（每次调用新建连接，用完即关）。""" + store = AppStateStore(host_state_db_path(state_dir)) + store.open() + return store def _registry_path(state_dir: str) -> str: - """群登记文件路径。""" + """旧版群登记文件路径（只用于一次性迁移导入）。""" return os.path.join(state_dir, _REGISTRY_FILENAME) @@ -61,8 +73,38 @@ def _empty_registry() -> dict: def _load_registry(state_dir: str) -> dict: - """读登记：无文件回空结构；损坏则 quarantine 后回空结构 - （损坏丢状态是 orphans 同款 fail-safe 口径，见差距表）。""" + """读登记：KV 行缺失时走一次性旧文件导入；行损坏/库打不 + 开时回空结构（损坏丢状态是 orphans 同款 fail-safe 口径， + 见差距表；下次成功保存即覆盖自愈）。""" + try: + with _open_store(state_dir) as store: + raw = store.kv_get(_KV_NAMESPACE, _KV_KEY) + except Exception: + _LOGGER.warning("chv_swarm_registry_store_unreadable " + "state_dir=%s", state_dir) + return _empty_registry() + if raw is None: + return _import_legacy(state_dir) + try: + data = json.loads(raw) + except ValueError: + _LOGGER.warning("chv_swarm_registry_corrupt state_dir=%s", + state_dir) + return _empty_registry() + if not isinstance(data, dict): + return _empty_registry() + data.setdefault("swarms", {}) + data.setdefault("tools", {}) + data.setdefault("invocations", []) + data.setdefault("audit", []) + return data + + +def _import_legacy(state_dir: str) -> dict: + """一次性导入旧 swarms.json：合法即进 KV 并把文件改名 + ``.migrated``；无文件回空；损坏则 quarantine 后回空（旧 + 口径）。导入失败（库写不进）时返回读到的数据、文件保留， + 下次加载重试。""" path = _registry_path(state_dir) try: with open(path, "r") as fh: @@ -72,27 +114,41 @@ def _load_registry(state_dir: str) -> dict: except (OSError, ValueError): _quarantine(path) return _empty_registry() - if not isinstance(data, dict): + if not isinstance(data, dict) \ + or not isinstance(data.get("swarms", {}), dict) \ + or not isinstance(data.get("tools", {}), dict) \ + or not isinstance(data.get("invocations", []), list) \ + or not isinstance(data.get("audit", []), list): _quarantine(path) return _empty_registry() data.setdefault("swarms", {}) data.setdefault("tools", {}) data.setdefault("invocations", []) data.setdefault("audit", []) + try: + with _open_store(state_dir) as store: + store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data)) + except Exception: + _LOGGER.warning("chv_swarm_registry_import_failed " + "state_dir=%s", state_dir) + return data + try: + os.replace(path, path + ".migrated") + except OSError: + pass return data def _save_registry(state_dir: str, data: dict) -> None: - """原子落盘登记（审计与调用记录各截断到上限）；写失败 - 只记日志，不炸调用方。""" + """落盘登记到 KV 行（审计与调用记录各截断到上限）；写失 + 败只记日志，不炸调用方。""" data["audit"] = data.get("audit", [])[-MAX_AUDIT_ENTRIES:] data["invocations"] = data.get("invocations", [])[ -MAX_INVOCATION_ENTRIES:] try: - os.makedirs(state_dir, exist_ok=True) - atomic_write_text(_registry_path(state_dir), - json.dumps(data)) - except OSError: + with _open_store(state_dir) as store: + store.kv_put(_KV_NAMESPACE, _KV_KEY, json.dumps(data)) + except Exception: _LOGGER.warning("chv_swarm_registry_save_failed " "state_dir=%s", state_dir) diff --git a/cloud/tests/test_chv_muse_p8.py b/cloud/tests/test_chv_muse_p8.py index c50f869..47cdbd2 100644 --- a/cloud/tests/test_chv_muse_p8.py +++ b/cloud/tests/test_chv_muse_p8.py @@ -547,15 +547,37 @@ class PresentationTestCase(unittest.TestCase): with self.assertRaises(ValueError): self.ledger.should_notify("t1", "", now=T0) - def test_ledger_corrupt_file_recovers(self): + def test_ledger_corrupt_row_recovers(self): + # round-12 迁移后：损坏的 KV 行回空账本（旧版是损坏的 + # JSON 文件 quarantine 后回空；语义同款，fail-safe）。 self.assertTrue(self.ledger.should_notify("t1", "failed", now=T0)) + from cloud.chv.app_state import AppStateStore, \ + host_state_db_path + with AppStateStore(host_state_db_path(self.state_dir)) as store: + store.kv_put("p8-ledger", "registry", "{not json") + self.assertTrue(self.ledger.should_notify("t2", "failed", + now=T0)) + + def test_ledger_legacy_json_imported_once(self): + # 旧版 JSON 文件首次加载时导入 KV 并改名 .migrated； + # 之后再写坏 JSON 也不影响 KV 里的账本。 + with open(os.path.join(self.state_dir, + "notification-ledger.json"), + "w") as fh: + fh.write('{"notified": {"t0:failed": 1000}}') + self.assertFalse(self.ledger.should_notify( + "t0", "failed", now=1000 + 10)) + self.assertTrue(os.path.exists(os.path.join( + self.state_dir, "notification-ledger.json.migrated"))) + self.assertFalse(os.path.exists(os.path.join( + self.state_dir, "notification-ledger.json"))) with open(os.path.join(self.state_dir, "notification-ledger.json"), "w") as fh: fh.write("{not json") - self.assertTrue(self.ledger.should_notify("t2", "failed", - now=T0)) + self.assertFalse(self.ledger.should_notify( + "t0", "failed", now=1000 + 20)) def test_present_or_log_combined(self): out = present_or_log({"status": "failed"}, self.ledger, diff --git a/cloud/tests/test_chv_muse_round12.py b/cloud/tests/test_chv_muse_round12.py new file mode 100644 index 0000000..b6709f3 --- /dev/null +++ b/cloud/tests/test_chv_muse_round12.py @@ -0,0 +1,528 @@ +"""对齐 Muse round-12 集成测试（P4～P12 合入后的跨轨接线）。 + +只测本轨新增的接线，不复述各轨自有测试： +- AppStateStore KV API + audit_trail 表（迁移地基）。 +- P8 三处登记 + P12 审计 JSONL 到 AppStateStore 的迁移 + （语义不变：去重、哈希链、project_id 隔离；旧文件一次 + 性导入；损坏回空 fail-safe）。 +- P6 worker → P3 helper/authd → git 的本地协议链（真 + PrivsepClient + 真 worker server + 真 git status；runuser + 与 HTTP 拉取需真机，见测试 docstring）。 +- P12 审批卡 / 权限登记对真实 P3/P5 代码的 dict 契约 + （不再用替身）。 +""" + +import importlib.util +import json +import os +import socket +import sqlite3 +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +from cloud.chv.app_state import AppStateStore, host_state_db_path +from cloud.chv.audit_trail import ( + AuditTrail, + StoreAuditTrail, + migrate_jsonl_to_store, +) +from cloud.chv.swarm import ( + create_swarm, + get_swarm, +) +from cloud.chv.presentation import NotificationLedger +from cloud.chv.browser_tasks import ( + BrowserTransport, + create_instance, + get_instance, +) + +TOKEN = "fake-token-123" +T0 = 1700000000.0 + + +def _load_guest_module(filename, module_name): + """按文件加载 chv-guest 下的模块（不走包名，与 P3 测试同款）。""" + path = REPO_ROOT / "chv-guest" / filename + spec = importlib.util.spec_from_file_location(module_name, path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +class KvApiTest(unittest.TestCase): + """AppStateStore KV API（P8 迁移的后端）。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.store = AppStateStore( + os.path.join(self.dir, "app-state.sqlite")) + self.store.open() + + def tearDown(self): + self.store.close() + self.tmp.cleanup() + + def test_put_get_delete_roundtrip(self): + self.assertIsNone(self.store.kv_get("p8-swarm", "registry")) + self.store.kv_put("p8-swarm", "registry", '{"a": 1}') + self.assertEqual(self.store.kv_get("p8-swarm", "registry"), + '{"a": 1}') + self.store.kv_put("p8-swarm", "registry", '{"a": 2}') + self.assertEqual(self.store.kv_get("p8-swarm", "registry"), + '{"a": 2}') + self.store.kv_delete("p8-swarm", "registry") + self.assertIsNone(self.store.kv_get("p8-swarm", "registry")) + self.store.kv_delete("p8-swarm", "registry") + + def test_namespaces_isolated(self): + self.store.kv_put("p8-swarm", "registry", "swarm-doc") + self.store.kv_put("p8-ledger", "registry", "ledger-doc") + self.assertEqual(self.store.kv_get("p8-swarm", "registry"), + "swarm-doc") + self.assertEqual(self.store.kv_get("p8-ledger", "registry"), + "ledger-doc") + + def test_bad_namespace_or_key_rejected(self): + for ns, key in (("", "k"), ("a:b", "k"), ("ns", ""), + ("ns", "a:b")): + with self.assertRaises(ValueError): + self.store.kv_put(ns, key, "v") + with self.assertRaises(ValueError): + self.store.kv_put("ns", "k", 123) + + def test_host_path_differs_from_vm_path(self): + from cloud.chv.app_state import state_db_path + host = host_state_db_path(self.dir) + vm = state_db_path(self.dir, "chv-abc123") + self.assertNotEqual(host, vm) + self.assertTrue(host.endswith("app-state.sqlite")) + self.assertEqual(os.path.dirname(host), self.dir) + + def test_counts_includes_audit_trail(self): + counts = self.store.counts() + self.assertIn("audit_trail", counts) + self.assertEqual(counts["audit_trail"], 0) + + +class StoreAuditTrailTest(unittest.TestCase): + """StoreAuditTrail：与 JSONL 版同款链与隔离，走 SQLite。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + + def tearDown(self): + self.tmp.cleanup() + + def test_append_chain_resumes_across_instances(self): + first = StoreAuditTrail(self.dir) + r1 = first.append("p1", "worker", "task_dispatch", "run", + object_id="t1", ts=T0) + r2 = first.append("p1", "worker", "vm_lifecycle", "wake", + object_id="vm1", ts=T0 + 1) + self.assertEqual(r1["seq"], 1) + self.assertEqual(r2["seq"], 2) + self.assertEqual(r2["prev_hash"], r1["hash"]) + second = StoreAuditTrail(self.dir) + r3 = second.append("p1", "worker", "task_dispatch", "done", + object_id="t1", ts=T0 + 2) + self.assertEqual(r3["seq"], 3) + self.assertEqual(r3["prev_hash"], r2["hash"]) + ok, bad = second.verify() + self.assertTrue(ok) + self.assertEqual(bad, -1) + + def test_project_isolation_and_filters(self): + trail = StoreAuditTrail(self.dir) + trail.append("p1", "worker", "task_dispatch", "run", + object_id="t1", ts=T0) + trail.append("p2", "worker", "task_dispatch", "run", + object_id="t2", ts=T0) + trail.append("p1", "worker", "quota", "denied", + object_id="t1", ts=T0 + 1) + self.assertEqual(len(trail.list_records("p1", limit=0)), 2) + self.assertEqual(len(trail.list_records("p2", limit=0)), 1) + self.assertEqual(trail.list_records("p3", limit=0), []) + self.assertEqual(trail.list_records("", limit=0), []) + self.assertEqual( + len(trail.list_records("p1", category="quota", + limit=0)), 1) + rows = trail.list_records("p1", limit=0) + self.assertGreater(rows[0]["ts"], rows[1]["ts"]) + + def test_tamper_detected_with_row_index(self): + trail = StoreAuditTrail(self.dir) + trail.append("p1", "worker", "task_dispatch", "run", + object_id="t1", ts=T0) + trail.append("p1", "worker", "task_dispatch", "done", + object_id="t1", ts=T0 + 1) + db = host_state_db_path(self.dir) + conn = sqlite3.connect(db) + conn.execute("UPDATE audit_trail SET action = ? WHERE seq = 2", + ("forged",)) + conn.commit() + conn.close() + ok, bad = StoreAuditTrail(self.dir).verify() + self.assertFalse(ok) + self.assertEqual(bad, 1) + + def test_export_json(self): + trail = StoreAuditTrail(self.dir) + trail.append("p1", "worker", "task_dispatch", "run", + object_id="t1", ts=T0) + trail.append("p1", "worker", "task_dispatch", "done", + object_id="t1", ts=T0 + 1) + dest = os.path.join(self.dir, "export.json") + self.assertEqual(trail.export_json("p1", dest), 2) + self.assertEqual(os.stat(dest).st_mode & 0o777, 0o600) + with open(dest) as fh: + rows = json.load(fh) + self.assertEqual([r["seq"] for r in rows], [1, 2]) + + def test_secret_detail_scrubbed(self): + trail = StoreAuditTrail(self.dir) + rec = trail.append("p1", "worker", "task_dispatch", "run", + detail={"token": TOKEN, "note": "ok"}, + ts=T0) + self.assertNotIn("token", rec["detail"]) + self.assertEqual(rec["detail"]["note"], "ok") + + +class MigrateJsonlTest(unittest.TestCase): + """migrate_jsonl_to_store：原样搬链。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.jsonl = os.path.join(self.dir, "audit.jsonl") + + def tearDown(self): + self.tmp.cleanup() + + def _seed_jsonl(self): + trail = AuditTrail(self.jsonl) + trail.append("p1", "worker", "task_dispatch", "run", + object_id="t1", ts=T0) + trail.append("p2", "worker", "quota", "denied", + object_id="t2", ts=T0 + 1) + trail.append("p1", "worker", "task_dispatch", "done", + object_id="t1", ts=T0 + 2) + return trail + + def test_migrate_then_verify_and_list(self): + self._seed_jsonl() + self.assertEqual(migrate_jsonl_to_store(self.jsonl, self.dir), + 3) + store_trail = StoreAuditTrail(self.dir) + ok, bad = store_trail.verify() + self.assertTrue(ok) + self.assertEqual(bad, -1) + self.assertEqual(len(store_trail.list_records("p1", + limit=0)), 2) + self.assertEqual(len(store_trail.list_records("p2", + limit=0)), 1) + rows = store_trail.list_records("p1", limit=0) + self.assertEqual([r["seq"] for r in rows], [3, 1]) + + def test_migrate_is_idempotent(self): + self._seed_jsonl() + self.assertEqual(migrate_jsonl_to_store(self.jsonl, self.dir), + 3) + self.assertEqual(migrate_jsonl_to_store(self.jsonl, self.dir), + 0) + ok, _ = StoreAuditTrail(self.dir).verify() + self.assertTrue(ok) + + def test_missing_file_returns_zero(self): + self.assertEqual( + migrate_jsonl_to_store(os.path.join(self.dir, "nope"), + self.dir), 0) + + def test_broken_chain_refused(self): + self._seed_jsonl() + with open(self.jsonl, "a") as fh: + fh.write("{broken json\n") + with self.assertRaises(ValueError): + migrate_jsonl_to_store(self.jsonl, self.dir) + self.assertEqual( + len(StoreAuditTrail(self.dir).list_records("p1", + limit=0)), 0) + + def test_divergent_seq_refused(self): + self._seed_jsonl() + migrate_jsonl_to_store(self.jsonl, self.dir) + other = os.path.join(self.dir, "other.jsonl") + trail = AuditTrail(other) + trail.append("p1", "worker", "task_dispatch", "run", + object_id="tX", ts=T0) + with self.assertRaises(ValueError): + migrate_jsonl_to_store(other, self.dir) + + +class P8MigrationTest(unittest.TestCase): + """P8 三处登记走 KV：无 JSON 文件、旧文件导入、损坏恢复。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + + def tearDown(self): + self.tmp.cleanup() + + def test_swarm_uses_kv_no_json_file(self): + swarm = create_swarm(self.dir, "p1", "parent1", now=T0) + self.assertFalse(os.path.exists( + os.path.join(self.dir, "swarms.json"))) + self.assertTrue(os.path.exists(host_state_db_path(self.dir))) + with AppStateStore(host_state_db_path(self.dir)) as store: + raw = store.kv_get("p8-swarm", "registry") + self.assertIsNotNone(raw) + self.assertIn(swarm["id"], json.loads(raw)["swarms"]) + self.assertEqual(get_swarm(self.dir, swarm["id"])["id"], + swarm["id"]) + + def test_swarm_legacy_json_imported_once(self): + legacy = {"swarms": {"s1": {"id": "s1", "project_id": "p1", + "parent_task_id": "parent1", + "agents": {}, "status": "active", + "created_at": T0}}, + "tools": {}, "invocations": [], "audit": []} + with open(os.path.join(self.dir, "swarms.json"), "w") as fh: + json.dump(legacy, fh) + self.assertEqual(get_swarm(self.dir, "s1")["id"], "s1") + self.assertTrue(os.path.exists( + os.path.join(self.dir, "swarms.json.migrated"))) + self.assertFalse(os.path.exists( + os.path.join(self.dir, "swarms.json"))) + + def test_swarm_corrupt_row_recovers(self): + create_swarm(self.dir, "p1", "parent1", now=T0) + with AppStateStore(host_state_db_path(self.dir)) as store: + store.kv_put("p8-swarm", "registry", "{not json") + swarm = create_swarm(self.dir, "p1", "parent2", now=T0) + self.assertEqual(swarm["parent_task_id"], "parent2") + + def test_ledger_uses_kv_no_json_file(self): + ledger = NotificationLedger(self.dir) + self.assertTrue(ledger.should_notify("t1", "failed", now=T0)) + self.assertFalse(os.path.exists(os.path.join( + self.dir, "notification-ledger.json"))) + with AppStateStore(host_state_db_path(self.dir)) as store: + raw = store.kv_get("p8-ledger", "registry") + self.assertIn("t1:failed", json.loads(raw)["notified"]) + + def test_browser_tasks_legacy_import_and_corrupt_row(self): + legacy = {"instances": { + "old-task": {"instance_id": "i1", "task_id": "old-task", + "status": "destroyed", + "shots_dir": "browser-tasks/i1/shots", + "created_at": T0}}} + with open(os.path.join(self.dir, "browser-tasks.json"), + "w") as fh: + json.dump(legacy, fh) + self.assertEqual( + get_instance(self.dir, "old-task")["instance_id"], "i1") + self.assertTrue(os.path.exists(os.path.join( + self.dir, "browser-tasks.json.migrated"))) + + class _FakeTransport(BrowserTransport): + def spawn(self, instance): + return None + + def close(self, instance): + return None + + def ping(self, instance): + return True + + with AppStateStore(host_state_db_path(self.dir)) as store: + store.kv_put("p8-browser-tasks", "registry", "{not json") + inst = create_instance(self.dir, "new-task", + _FakeTransport(), now=T0) + self.assertEqual(inst["status"], "active") + self.assertFalse(os.path.exists(os.path.join( + self.dir, "browser-tasks.json"))) + + +class WorkerProtocolTest(unittest.TestCase): + """P6 worker → P3 helper/authd → git 本地协议链。 + + 真 PrivsepClient + 真 chv-gitea-worker server（Unix socket + + SO_PEERCRED）+ 真 git status；run_git 探针调 worker 自身 + 的 ``_git_base_argv()``（含 P3 helper 与 useHttpPath）组 + argv，只去掉本机不存在的 ``runuser -u chv`` 包装——runuser + 与 HTTP 拉取（需真实 gitea.tose.top + Sentinel 放行）留 + 给真机轨，本测试不假装已覆盖。 + """ + + HELPER = REPO_ROOT / "chv-guest" / "git-credential-chv-authd" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.workspace = os.path.join(self.dir, "workspace") + os.makedirs(self.workspace) + self.repo = os.path.join(self.workspace, "testrepo") + subprocess.run(["git", "init", "-q", "-b", "main", self.repo], + check=True) + subprocess.run(["git", "-C", self.repo, "config", + "user.email", "t@example.invalid"], check=True) + subprocess.run(["git", "-C", self.repo, "config", + "user.name", "Test"], check=True) + with open(os.path.join(self.repo, "README.md"), "w") as fh: + fh.write("hello\n") + subprocess.run(["git", "-C", self.repo, "add", "."], + check=True) + subprocess.run(["git", "-C", self.repo, "commit", "-q", + "-m", "init"], check=True) + self.socket_path = os.path.join(self.dir, "worker.sock") + self.log_path = os.path.join(self.dir, "worker.log") + self.mod = _load_guest_module("chv-gitea-worker.py", + "chv_gitea_worker_round12") + self.mod.HELPER_PATH = str(self.HELPER) + self.argv_seen = [] + self.env_seen = [] + + def run_git(action, repo): + argv = ["runuser", "-u", self.mod.SYNC_USER, "--", + "env", "HOME=/home/chv"] + \ + self.mod._git_base_argv() + if action == "status": + argv += ["-C", repo, "status", "-sb"] + elif action == "pull": + argv += ["-C", repo, "pull", "--ff-only"] + elif action == "push": + argv += ["-C", repo, "push"] + else: + return 2, "", "未知 git 操作：%s" % action + self.argv_seen.append(list(argv)) + env = {"PATH": os.environ.get("PATH", ""), + "HOME": self.dir, + "CHV_AUTHD_SOCKET": ""} + self.env_seen.append(dict(env)) + runnable = [a for a in argv + if a not in ("runuser", "-u", "chv", "--")] + runnable[1] = "HOME=%s" % self.dir + proc = subprocess.run(runnable, capture_output=True, + text=True, timeout=60, + env={**env, "HOME": self.dir}) + return proc.returncode, proc.stdout, proc.stderr + + self.server = self.mod.WorkerServer( + self.socket_path, self.workspace, self.log_path, + allowed_uids=(os.getuid(),), run_git=run_git, + helper_path=str(self.HELPER)) + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.tmp.cleanup() + + def test_status_through_real_client_and_server(self): + from cloud.chv.privsep import PrivsepClient, WorkerRequest + client = PrivsepClient(socket_path=self.socket_path, + connector="gitea", + workspace=self.workspace) + ping = client.ping() + self.assertTrue(ping.ok) + resp = client.call(WorkerRequest( + "status", "gitea", self.repo, "req-1")) + self.assertTrue(resp.ok, resp.stderr) + self.assertEqual(resp.returncode, 0) + self.assertIn("## main", resp.stdout) + self.assertEqual(resp.request_id, "req-1") + # worker 的 git argv 走 P3 helper + useHttpPath，不含 token： + self.assertEqual(len(self.argv_seen), 1) + argv = self.argv_seen[0] + self.assertIn("credential.helper=%s" % self.HELPER, argv) + self.assertIn("credential.useHttpPath=true", argv) + self.assertNotIn(TOKEN, " ".join(argv)) + self.assertNotIn(TOKEN, json.dumps(self.env_seen)) + # worker 用的 helper 就是 P3 的 helper 文件本身： + self.assertTrue(os.path.samefile( + self.server.helper_path, str(self.HELPER))) + # 审计日志不含 token（本测试的 token 只存在于断言字面量， + # 不进任何被测输入——worker 从未见过它）： + with open(self.log_path) as fh: + self.assertNotIn(TOKEN, fh.read()) + + def test_helper_file_is_the_p3_helper(self): + # worker 硬编码的 helper 路径与 P3 测试用的 helper 是同一 + # 个文件（接线不是替身）。 + from cloud.tests.test_chv_muse_p3 import HelperE2ETest + self.assertTrue(os.path.samefile( + str(self.HELPER), str(HelperE2ETest.HELPER))) + + +class P12DictContractTest(unittest.TestCase): + """P12 接线用真实 P3/P5 代码，不用替身。""" + + def test_approval_card_from_real_p3_queue(self): + from cloud.chv.sentinel import ApprovalQueue, EgressRequest + from cloud.chv.approval_card import ( + build_approval_card, + card_matches_channel, + ) + from cloud.chv.approval_channel import request_approval + queue = ApprovalQueue(clock=lambda: T0) + req = EgressRequest("gitea", "POST", "https", + "gitea.tose.top", 443, + "/api/v1/repos/admin/x/issues") + pending_id = queue.record(req, ["needs_approval"]) + record = queue.list_pending()[0] + self.assertEqual(record["pending_id"], pending_id) + card = build_approval_card(record) + self.assertEqual(card["pending_id"], pending_id) + self.assertEqual(card["target"]["host"], "gitea.tose.top") + self.assertEqual(card["target"]["method"], "POST") + self.assertEqual(card["risk"], "medium") + self.assertIn("needs_approval", card["reasons"]) + channel = request_approval(record, secret="test-secret", + now=T0) + self.assertTrue( + card_matches_channel(card, channel.payload())) + + def test_permission_registry_from_real_p5_describe(self): + from cloud.chv.capabilities import CapabilityIssuer + from cloud.chv.activity import PermissionRegistry + issuer = CapabilityIssuer(clock=lambda: T0) + cap = issuer.issue( + "one-shot", "gitea", "gitea.tose.top", ["GET"], + [("GET", "/api/v1/*", "allow")], "sync", now=T0) + described = cap.describe() + reg_path = os.path.join( + tempfile.mkdtemp(), "permissions.json") + reg = PermissionRegistry(reg_path) + grant = reg.record_grant("p1", described, ts=T0) + self.assertEqual(grant["capability_id"], + described["capability_id"]) + self.assertEqual(grant["connector"], "gitea") + self.assertEqual(grant["host"], "gitea.tose.top") + self.assertEqual(grant["use_case"], "sync") + self.assertEqual(grant["tier"], "one-shot") + active = reg.list_active("p1", now=T0 + 10) + self.assertEqual(len(active), 1) + self.assertEqual(active[0]["capability_id"], + described["capability_id"]) + self.assertTrue(reg.note_use("p1", + described["capability_id"])) + self.assertTrue(reg.revoke("p1", + described["capability_id"])) + self.assertEqual(reg.list_active("p1", now=T0 + 10), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/chv-muse-p12-gap.md b/docs/chv-muse-p12-gap.md index c9ec492..9063866 100644 --- a/docs/chv-muse-p12-gap.md +++ b/docs/chv-muse-p12-gap.md @@ -76,14 +76,20 @@ 字段；任务简报允许二选一。 - 审计持久化先用独立 JSONL 文件（`audit_trail.jsonl`，0600）， 待 P7 状态库就绪后迁移，迁移接口是 `from_legacy_row()` 同款 - 转换（本文件行 → P7 记录）。 + 转换（本文件行 → P7 记录）。**迁移已完成（round-12）**： + `StoreAuditTrail` 走宿主级 AppStateStore 的 `audit_trail` 表， + 与 JSONL 版同一记录形状与哈希链；`migrate_jsonl_to_store()` + 原样搬链（坏链拒绝、分叉拒绝、重跑幂等）；JSONL 版保留读旧 + 文件，新写一律走 Store 版。测试钉在 + `cloud/tests/test_chv_muse_round12.py`。 - project_id 强制：`list()`/`export()` 必须传 project_id，只返 回该项目记录；跨项目读取返回空，不抛错（与现有 `list_audit` 行为一致，不泄露“存在别的项目记录”这一事实）。 ## 6. 真机待验证清单（未做，排队在 round-11 之后） -1. 审计 JSONL 在真实 VM 生命周期（wake/pause/resume/shutdown） +1. 审计轨迹（round-12 起走 StoreAuditTrail / SQLite，不再是 + JSONL）在真实 VM 生命周期（wake/pause/resume/shutdown） 下的追加与 `verify()` 表现。 2. 活动摘要与真实任务派发/完成事件的对应（正在做是否及时变 更、计划状态流转是否与 worker 终态一致）。 @@ -91,5 +97,5 @@ （签发→使用→过期→撤销全程）。 4. 审批卡数据与 P5 对话外通道在真机上的端到端一致（卡上点的 接受/拒绝与通道答复同一 pending_id）。 -5. 审计文件落盘权限（0600）与多进程追加（worker + host.py） - 下的完整性。 +5. 审计库落盘权限（0600）与多进程追加（worker + host.py） + 下的完整性（SQLite WAL + 30 秒 busy 超时）。 diff --git a/docs/chv-muse-p6-gap.md b/docs/chv-muse-p6-gap.md index 165ee3b..c493f42 100644 --- a/docs/chv-muse-p6-gap.md +++ b/docs/chv-muse-p6-gap.md @@ -129,6 +129,12 @@ Base64、零宽字符）、藏在图片/附件/二进制里的指令、需要理 authd socket 路径、Sentinel 评估发生在 authd 放凭证前）。本轨 测试用替身实现同一协议；与 P3 代码的真正接通发生在合入时， 须由集成轨跑通 worker→helper→authd→git 的全链路。 + **已接通（round-12，本地部分）**：`cloud/tests/ + test_chv_muse_round12.py`（WorkerProtocolTest）用真 + PrivsepClient + 真 worker server + 真 git status 跑通，worker + 的 git argv 含真实 P3 helper 与 useHttpPath、argv/env/审计 + 均不含 token；runuser 包装与 HTTP 拉取（需真实 gitea.tose.top + + Sentinel 放行）留给真机轨。 2. **worker 跑在客内 root**：worker 守护进程以 root 身份跑在单元外 （与 P3 authd 同款选择），git 子进程再以 chv 身份跑。root worker 本身不受单元隔离保护——它的隔离来自 systemd 沙箱 + socket 白名单 diff --git a/docs/chv-muse-p8-gap.md b/docs/chv-muse-p8-gap.md index d2997dc..65098af 100644 --- a/docs/chv-muse-p8-gap.md +++ b/docs/chv-muse-p8-gap.md @@ -50,16 +50,19 @@ VM 观察，未见于官方页面原文，本轨不把它当官方术语用； | `cloud/chv/browser_tasks.py`（新） | 每浏览器任务一个实例：建/绑/毁生命周期、单实例单任务（复用即错、毁后复用即错）、每实例独立状态目录；transport 接口注入（测试用 fake） | 本轨 | | `cloud/automations.py`（改） | 只加重试/停用所需的列与抢单入口（`max_retries`、`retry_backoff_seconds`、`consecutive_failures`、`next_retry_at`、`paused` 状态）；host.py run-due **不动**（host 侧唯一执行器，见该模块 docstring） | 本轨 | -**P7 状态落点（更新 2026-11-07 06:39）：** P7 已完成并提交（分支 -`feat-chv-muse-p7`，提交 `29fa0c8`：每 VM 一个 SQLite 状态库 -AppStateStore，0600 WAL，tasks/goals/sessions/只追加 audit/meta）。 -本轨仍先用 `state_dir` 下的原子 JSON 文件（复用 -`cloud/chv/atomic_file.py`），模块 docstring 与本表明确标注“待迁移到 -P7 状态存储”；不重复造一套状态库。集成轨（round-12）把 -`swarms.json` / `notification-ledger.json` / `browser-tasks.json` -迁到 AppStateStore，接口（load/save/原子写）已按同一模式写好，迁移 -只是换后端。crons 与 swarm 的状态复用 `automations` 表与 swarm 登记 -文件（同款原子写），同样标注待迁移。 +**P7 状态落点（更新 2026-11-07 06:39；迁移已完成，round-12 更新）：** +P7 已完成并提交（分支 `feat-chv-muse-p7`，提交 `29fa0c8`：每 VM +一个 SQLite 状态库 AppStateStore，0600 WAL， +tasks/goals/sessions/只追加 audit/meta）。**round-12 已把三处登记 +迁到 AppStateStore**（分支 `feat-chv-muse-round12`）：宿主级库 +`state_dir/app-state.sqlite`（`host_state_db_path`，与 per-VM 库 +同库不同文件），命名空间 `p8-swarm` / `p8-ledger` / +`p8-browser-tasks`，一行存整份登记文档；旧 JSON 文件首次加载时 +自动导入并改名 `.migrated`；KV 行损坏/库打不开时回空结构 +（fail-safe，下次成功保存即覆盖自愈）。测试钉在 +`cloud/tests/test_chv_muse_round12.py`（P8MigrationTest）与 +`test_chv_muse_p8.py`（损坏行恢复 + 旧文件导入）。crons 的状态 +仍在 `automations` 表，未动。 **客机侧跟进（本轨只写规约、不改 P2 的文件）：** 每任务浏览器实例的客机 进程拉起/销毁在 P2 的 `chv-guest/chv-broker.py`（`browser` 操作固定转发到 @@ -67,6 +70,17 @@ P7 状态存储”；不重复造一套状态库。集成轨（round-12）把 socket），`close` 后杀掉进程。P2 的文件不在本分支基线里，本轨不复制、不 分叉，规约写进本表 §6，留给 round-12 集成时落在 P2 分支上。 +**客机侧跟进（round-12 决定：未做，留给真机轨）：** round-12 核实后未做， +原因三条：① `chv-browser.py` 的请求协议里没有任务 id 字段（只有 +open/snapshot/screenshot/close/ping），broker 分不清请求属于哪个浏览器任 +务，改 broker 之前须先改协议 + 调 broker 的调用方；② 进程拉起/销毁涉 +及以 chv-browser 身份起进程、socket 就绪等待、close 后杀进程与僵尸回收、 +broker 重启后的残留清理，在没有真机验证的情况下盲改，坏了就是每任务浏览 +器全断；③ 本地只能用假浏览器进程测生命周期，真 Playwright + 真 +chromium-headless-shell 的起停只能在真机上验。留给真机轨：先在真机上改 +协议（请求加 task_id）→ broker 按任务拉起/杀进程 → 连续两任务端到端 +（§6 第 4 项）。**不许写成已做。** + ## 4. 自定数值清单（全部是本项目自定，非 Meta 官方值） | 数值 | 本项目自定值 | 说明 | @@ -104,6 +118,7 @@ VM/单元里（P1 常驻池 + P2 单元是底座）。“子代理构建自己 3. 回访通知端到端：后台任务完成后有新内容才通知、无新内容不通知、同一原因 1 小时内不重复通知。 4. 每任务浏览器实例端到端：连续两个浏览器任务各起全新实例、第二个看不到 -第一个的截图与快照状态（需 §3 客机侧跟进先落地）。 -5. 账本迁移：P7 存储已落地（AppStateStore，`feat-chv-muse-p7` -提交 `29fa0c8`），集成轨把通知账本从 JSON 文件迁过去、去重语义不变。 +第一个的截图与快照状态（需 §3 客机侧跟进先落地；round-12 未做，见 §3）。 +5. 账本迁移：**已完成（round-12）**——通知账本与另两处登记已迁到宿主级 +AppStateStore（见 §3 更新），去重语义不变，测试钉住；真机上只需确认 +升级后旧 JSON 文件一次导入、不丢去重状态。 diff --git a/docs/chv-production-readiness.md b/docs/chv-production-readiness.md index 079dbc8..e56e71e 100644 --- a/docs/chv-production-readiness.md +++ b/docs/chv-production-readiness.md @@ -1040,3 +1040,46 @@ auditd 审计、与 P3 authd/Sentinel 的真接线）。 - 真机待验证：真实 token 回传下的月度累计与 402、真实多任务下的并发门、 真实 worktree 下的存储门、提醒只发一次、账本并发写不断链。 +## 20. 对齐 Muse round-12 集成：P4～P12 合入（分支 feat-chv-muse-round12） + +- 合入顺序与提交：P4 c24cc0b → P5 d1c9a8e → P6 b0b9f5e → P7 320cd36 → + P8 240949e → P9 90c42f6 → P10 b55b49b（acf98f8 的 cherry-pick，只含 + P10 自有 diff，基线已含的 P3/P5 未重复合）→ P11 77743e7 → P12 + 466cf7e。基线 round-11 HEAD 29d3a5a。 +- 冲突处理（逐项）：① executor.py：P9 的 `_inference_gate` 插到 P1 + 分发器 `_run_with_vm` 之前，gate 调用为分发器第一条语句（deny/ask + 在常驻 resume 与冷启动 wake 之前 fail-closed）；P9 自己的旧版本 + `_run_with_vm` 丢弃——其本体与 P1 `_run_cold` 逐字节相同（P1 的 + verbatim 搬运），合入时一度误删 `_run_cold` 体致 dispatch 回 None， + 已从 HEAD 原样恢复并由测试钉住；sentinel.py 取 P9 超集版。 + ② build.sh/buildinfo.py：按 hunk 合，authd 三件套 + P2 三件套 + + P4 三文件 + P7 chv-memory 全在 contents 里，GT 哈希仍为已批准的 + 49ffe090…。③ docs §15 撞号：P4→§16、P6→§17、P7→§18、P11→§19， + 顺序顺延，无覆盖。④ P4 弄坏的 P2 测试改走假 CDP broker，断言不变。 +- AppStateStore 迁移（本轨做完）：P8 三处登记（swarms.json / + notification-ledger.json / browser-tasks.json）迁到宿主级库 + `state_dir/app-state.sqlite`（`host_state_db_path`，与 per-VM 库 + 同库不同文件），命名空间 `p8-swarm` / `p8-ledger` / + `p8-browser-tasks`，一行存整份文档；旧 JSON 首次加载自动导入并改名 + `.migrated`；KV 行损坏回空 fail-safe。P12 审计 JSONL 迁到 + `audit_trail` 表（`StoreAuditTrail`，同一记录形状与哈希链）， + `migrate_jsonl_to_store()` 原样搬链（坏链拒绝、分叉拒绝、重跑幂 + 等）；JSONL 版保留读旧文件，新写一律走 Store 版。`kv_put` 不过 + 无凭证守卫（登记文档含用户 prompt，见 app_state.py 注释）。 +- 接线验证（本轨，本地）：P6 worker→P3 helper/authd→git 跑通真 + PrivsepClient + 真 worker server + 真 git status（argv 含真实 + helper 与 useHttpPath，argv/env/审计不含 token；runuser 与 HTTP + 拉取留给真机）；P12 审批卡吃真实 P3 `ApprovalQueue.record()`、 + 权限登记吃真实 P5 `ScopedCapability.describe()`，均已用真实代码 + 跑通，不再用替身。 +- 未决与留给真机轨：① P11 冲突原样保留（档位门只做准入、USD 路径 + 不动，等用户裁决，见 `docs/chv-muse-p11-gap.md` §5）。② P8 客机 + 侧跟进未做（`chv-browser.py` 请求协议无任务 id 字段、进程拉起/ + 销毁须真机验证，见 `docs/chv-muse-p8-gap.md` §3）。③ 真机验证排 + 在 round-11 之后：P1 常驻/balloon/vsock、P2 §5 九项否定测试、 + P3 §3.6 五项、P4/P5/P6/P7/P8/P9/P10/P11/P12 各自差距表 §6 清单、 + 以及本轨新增的迁移后真机表现（旧 JSON 一次导入、Store 审计链 + 在 VM 生命周期下的追加）。 +- 测试：新增 `cloud/tests/test_chv_muse_round12.py` 24 项全过； + 全量 cloud 套件见合入提交说明（每次合入后跑相关测试，最终全量 + 一次跑通才算完）。