feat(chv): 对齐 Muse P7——状态与记忆分离

- 每 VM 一个 SQLite 状态库（cloud/chv/app_state.py）：
  state_dir/<vm>/app-state.sqlite（0600，WAL），表 tasks/goals/
  sessions/audit（只追加）/meta；写操作过无凭证守卫（键名 +
  值模式）；备份用 sqlite3 在线备份 + manifest（sha256），恢复
  先验 manifest、先留 pre-restore 副本；DDL 用 SQLite 与
  Postgres 共通受限子集，换 Postgres = 换连接工厂。
- 记忆文件接线（cloud/chv/memory_files.py）：宿主镜像读/列/
  写/追加（路径穿越守卫、写审计）；MemorySystemClient 调真实
  记忆系统 CLI（search/get/explain/forget-plan/forget-confirm/
  backup/restore/export），不重造引擎；forget = plan → confirm
  → 本地清行 → search 验证查不到；pull_memory_home 经 SSH +
  tar 拉取客机记忆家目录。
- 客内脚本 chv-guest/chv-memory.py（纯标准库，list/read/
  append/forget/export），家目录 /home/chv/memory（workspace
  之外，不同步到 Gitea）；build.sh 拷入 + buildinfo 登记。
- 新测试 cloud/tests/test_chv_muse_p7.py 46 项全过（含真 CLI
  write/search/forget/backup/restore 实测、假 ssh 拉取、否定
  测试）；全量 cloud 套件 1913 项 OK（skipped=8）。
- 存储选型：SQLite 不是 Postgres（见差距表 §4；轻量 VM 跑
  Postgres server 太重，宿主共享 Postgres 破坏每用户 VM
  隔离）；忘记是尽力删除 + 验证，不是密码学擦除。
- 真机待验证（未做，等测试机空闲）：见
  docs/chv-muse-p7-gap.md §6 清单

cloud/chv/app_state.py

@@ -0,0 +1,499 @@
+"""每 VM 结构化应用状态库（对齐 Muse P7，叶子模块）。
+
+Muse 官方安全文（security.muse.ai 2026-09-08，见
+docs/chv-muse-p7-gap.md §1）：Postgres 存持久应用状态，与运行时
+单元、凭证存储分离。本轨的选型结论写在差距表 §4：轻量线用
+SQLite（每 VM 一个 0600 文件，WAL），不是 Postgres——镜像
++150–250MB 与每 VM 一个 server 进程对 2GB 轻量 VM 太重；宿主侧
+共享 Postgres 会把多用户状态集中到宿主，破坏每用户 VM 隔离。
+
+与 Postgres 的差距（诚实写明）：无多写者服务器、无网络访问、
+无行级安全、无内置复制（复制 = 备份/恢复路径）。DDL 用两方
+共通的受限子集（TEXT/INTEGER/REAL 列、TEXT id、无
+AUTOINCREMENT、无 SQLite 专有语法、参数化查询），换 Postgres =
+换连接工厂 + 同一份 DDL。
+
+分离保证：
+- 库文件只活在宿主 `state_dir/<vm>/app-state.sqlite`，不在客机
+  磁盘镜像里，nspawn cell 物理上够不着（测试钉住）。
+- 凭证绝不进库：所有写操作过无凭证守卫（键名 + 值模式双重
+  检查）；守卫拒绝时只报键名/模式，不含值。
+- audit 表只追加：本模块不提供任何删除/修改审计的方法。
+"""
+
+import hashlib
+import json
+import os
+import re
+import shutil
+import sqlite3
+import threading
+import time
+import uuid
+from collections.abc import Mapping
+
+from cloud.chv.atomic_file import atomic_write, atomic_write_text
+
+SCHEMA_VERSION = "1"
+
+# DDL：SQLite 与 Postgres 共通的受限子集。TEXT id 由调用方生成
+#（uuid hex），不用 AUTOINCREMENT；时间戳用 ISO 文本。
+DDL = """
+CREATE TABLE IF NOT EXISTS meta (
+    key TEXT PRIMARY KEY,
+    value TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS tasks (
+    id TEXT PRIMARY KEY,
+    user_id TEXT NOT NULL,
+    kind TEXT NOT NULL,
+    status TEXT NOT NULL,
+    payload_json TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_tasks_user ON tasks(user_id, status);
+CREATE TABLE IF NOT EXISTS goals (
+    id TEXT PRIMARY KEY,
+    user_id TEXT NOT NULL,
+    title TEXT NOT NULL,
+    plan_json TEXT NOT NULL,
+    status TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_goals_user ON goals(user_id, status);
+CREATE TABLE IF NOT EXISTS sessions (
+    id TEXT PRIMARY KEY,
+    user_id TEXT NOT NULL,
+    chat_id TEXT NOT NULL,
+    summary TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id, chat_id);
+CREATE TABLE IF NOT EXISTS audit (
+    id TEXT PRIMARY KEY,
+    actor TEXT NOT NULL,
+    action TEXT NOT NULL,
+    target TEXT NOT NULL,
+    detail_json TEXT NOT NULL,
+    created_at TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_audit_target ON audit(target, created_at);
+"""
+
+_BACKUP_FILENAME = "app-state.sqlite"
+_MANIFEST_FILENAME = "manifest.json"
+
+# 无凭证守卫：键名含这些子串即拒；值匹配这些模式即拒。
+_SECRET_KEY_SUBSTRINGS = ("TOKEN", "SECRET", "PASSWORD", "PRIVATE_KEY",
+                          "API_KEY", "ACCESS_KEY", "CLIENT_SECRET")
+_SECRET_VALUE_PATTERNS = (
+    re.compile(r"ghp_[A-Za-z0-9]{20,}"),
+    re.compile(r"gho_[A-Za-z0-9]{20,}"),
+    re.compile(r"github_pat_[A-Za-z0-9_]{20,}"),
+    re.compile(r"sk-[A-Za-z0-9]{20,}"),
+    re.compile(r"xox[baprs]-[A-Za-z0-9-]{10,}"),
+    re.compile(r"AKIA[0-9A-Z]{16}"),
+    re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
+)
+
+
+class SecretRejectedError(ValueError):
+    """写操作含疑似凭证：只报键名/模式，绝不含值。"""
+
+
+class AppStateError(RuntimeError):
+    """状态库操作失败（未打开、备份损坏等）。"""
+
+
+def state_db_path(state_dir: str, vm_name: str) -> str:
+    """某 VM 状态库路径：宿主 state_dir 下，不在客机磁盘镜像里。"""
+    return os.path.join(state_dir, vm_name, "app-state.sqlite")
+
+
+def _utcnow() -> str:
+    """ISO 时间戳（UTC，秒级）。"""
+    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+
+
+def _new_id() -> str:
+    """TEXT 主键：uuid hex。"""
+    return uuid.uuid4().hex
+
+
+def _scan_value(value: object, path: str) -> None:
+    """递归扫一个值：键名与字符串值过无凭证守卫。"""
+    if isinstance(value, Mapping):
+        for key, item in value.items():
+            _check_key_name("%s.%s" % (path, key))
+            _scan_value(item, "%s.%s" % (path, key))
+    elif isinstance(value, (list, tuple)):
+        for item in value:
+            _scan_value(item, path)
+    elif isinstance(value, str):
+        _check_secret_value(value, path)
+
+
+def _check_key_name(path: str) -> None:
+    """键路径含凭证子串即拒（只报路径，不含值）。"""
+    upper = path.upper()
+    for sub in _SECRET_KEY_SUBSTRINGS:
+        if sub in upper:
+            raise SecretRejectedError(
+                "secret_key_name:%s contains %s" % (path, sub))
+
+
+def _check_secret_value(value: str, path: str) -> None:
+    """字符串值命中凭证模式即拒（只报路径与模式序号，不含值）。"""
+    for index, pattern in enumerate(_SECRET_VALUE_PATTERNS):
+        if pattern.search(value):
+            raise SecretRejectedError(
+                "secret_value_pattern:%s pattern#%d" % (path, index))
+
+
+def assert_no_secrets(mapping: Mapping[str, object]) -> None:
+    """断言一个映射可入库：键名与值都不过凭证守卫即拒。"""
+    _scan_value(dict(mapping), "payload")
+
+
+def _sha256_file(path: str) -> str:
+    """文件 sha256（流式读，不整文件进内存）。"""
+    digest = hashlib.sha256()
+    with open(path, "rb") as handle:
+        while True:
+            chunk = handle.read(65536)
+            if not chunk:
+                break
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _read_manifest(snapshot_dir: str) -> dict[str, str]:
+    """读并校验备份 manifest：文件在 + sha256 对得上才返回。"""
+    manifest_path = os.path.join(snapshot_dir, _MANIFEST_FILENAME)
+    if not os.path.isfile(manifest_path):
+        raise AppStateError("backup manifest missing: %s" % manifest_path)
+    with open(manifest_path, "r", encoding="utf-8") as handle:
+        manifest = json.load(handle)
+    backup_path = os.path.join(snapshot_dir, _BACKUP_FILENAME)
+    if not os.path.isfile(backup_path):
+        raise AppStateError("backup file missing: %s" % backup_path)
+    actual = _sha256_file(backup_path)
+    if actual != manifest.get("sha256"):
+        raise AppStateError("backup sha256 mismatch: %s" % backup_path)
+    return manifest
+
+
+def verify_backup(snapshot_dir: str) -> dict[str, str]:
+    """校验一次备份（manifest + sha256），返回 manifest。"""
+    return _read_manifest(snapshot_dir)
+
+
+def restore_app_state(snapshot_dir: str, dest_path: str) -> dict[str, str]:
+    """从备份恢复：先验 manifest，成功后先留 pre-restore 副本再替换。"""
+    manifest = _read_manifest(snapshot_dir)
+    backup_path = os.path.join(snapshot_dir, _BACKUP_FILENAME)
+    if os.path.exists(dest_path):
+        stamp = time.strftime("%Y%m%dT%H%M%S", time.gmtime())
+        preserved = "%s.pre-restore-%s" % (dest_path, stamp)
+        shutil.copyfile(dest_path, preserved)
+        os.chmod(preserved, 0o600)
+    os.makedirs(os.path.dirname(dest_path) or ".", exist_ok=True)
+    shutil.copyfile(backup_path, dest_path)
+    os.chmod(dest_path, 0o600)
+    return manifest
+
+
+class AppStateStore:
+    """每 VM 应用状态库：单写者（进程内锁 + SQLite busy 超时）。
+
+    用法：store = AppStateStore(path); store.open(); ...;
+    store.close()。也可用作上下文管理器。库文件 0600，目录 0700。
+    """
+
+    def __init__(self, path: str) -> None:
+        self._path = path
+        self._conn: sqlite3.Connection | None = None
+        self._lock = threading.RLock()
+
+    def __enter__(self) -> "AppStateStore":
+        self.open()
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        self.close()
+
+    @property
+    def path(self) -> str:
+        """库文件路径。"""
+        return self._path
+
+    def open(self) -> None:
+        """打开（建库 + 建表 + 记 schema 版本），幂等。"""
+        with self._lock:
+            if self._conn is not None:
+                return
+            parent = os.path.dirname(self._path)
+            if parent:
+                os.makedirs(parent, mode=0o700, exist_ok=True)
+            conn = sqlite3.connect(self._path, timeout=30.0)
+            conn.execute("PRAGMA journal_mode=WAL")
+            conn.execute("PRAGMA synchronous=FULL")
+            conn.execute("PRAGMA busy_timeout=30000")
+            conn.executescript(DDL)
+            conn.execute(
+                "INSERT OR REPLACE INTO meta(key, value) VALUES (?, ?)",
+                ("schema_version", SCHEMA_VERSION))
+            conn.commit()
+            os.chmod(self._path, 0o600)
+            self._conn = conn
+
+    def close(self) -> None:
+        """关闭连接（幂等）。"""
+        with self._lock:
+            if self._conn is None:
+                return
+            self._conn.close()
+            self._conn = None
+
+    def _require_conn(self) -> sqlite3.Connection:
+        """取已打开的连接，未打开即报错。"""
+        if self._conn is None:
+            raise AppStateError("app state store not open: %s" % self._path)
+        return self._conn
+
+    def schema_version(self) -> str:
+        """库的 schema 版本。"""
+        with self._lock:
+            conn = self._require_conn()
+            row = conn.execute(
+                "SELECT value FROM meta WHERE key = ?",
+                ("schema_version",)).fetchone()
+            return row[0] if row else ""
+
+    def counts(self) -> dict[str, int]:
+        """各表行数（备份/恢复校验用）。"""
+        with self._lock:
+            conn = self._require_conn()
+            out: dict[str, int] = {}
+            for table in ("tasks", "goals", "sessions", "audit"):
+                row = conn.execute(
+                    "SELECT COUNT(*) FROM %s" % table).fetchone()
+                out[table] = int(row[0])
+            return out
+
+    def put_task(self, user_id: str, task_id: str, kind: str,
+                 status: str, payload: Mapping[str, object]) -> None:
+        """写/覆盖一个任务（payload 过无凭证守卫）。"""
+        assert_no_secrets(payload)
+        now = _utcnow()
+        blob = json.dumps(dict(payload), sort_keys=True)
+        with self._lock:
+            conn = self._require_conn()
+            conn.execute(
+                "INSERT INTO tasks(id, user_id, kind, status,"
+                " payload_json, created_at, updated_at)"
+                " VALUES (?, ?, ?, ?, ?, ?, ?)"
+                " ON CONFLICT(id) DO UPDATE SET"
+                " status = excluded.status,"
+                " payload_json = excluded.payload_json,"
+                " updated_at = excluded.updated_at",
+                (task_id, user_id, kind, status, blob, now, now))
+            conn.commit()
+
+    def get_task(self, task_id: str) -> dict[str, object] | None:
+        """取一个任务，不存在返回 None。"""
+        with self._lock:
+            conn = self._require_conn()
+            row = conn.execute(
+                "SELECT id, user_id, kind, status, payload_json,"
+                " created_at, updated_at FROM tasks WHERE id = ?",
+                (task_id,)).fetchone()
+            return _row_to_task(row) if row else None
+
+    def list_tasks(self, user_id: str,
+                   status: str | None = None) -> list[dict[str, object]]:
+        """列某用户的任务（可按状态过滤，按创建时间升序）。"""
+        with self._lock:
+            conn = self._require_conn()
+            if status is None:
+                rows = conn.execute(
+                    "SELECT id, user_id, kind, status, payload_json,"
+                    " created_at, updated_at FROM tasks"
+                    " WHERE user_id = ? ORDER BY created_at",
+                    (user_id,)).fetchall()
+            else:
+                rows = conn.execute(
+                    "SELECT id, user_id, kind, status, payload_json,"
+                    " created_at, updated_at FROM tasks"
+                    " WHERE user_id = ? AND status = ? ORDER BY created_at",
+                    (user_id, status)).fetchall()
+            return [_row_to_task(row) for row in rows]
+
+    def update_task_status(self, task_id: str, status: str) -> bool:
+        """只改任务状态（不碰 payload），改到返回 True。"""
+        with self._lock:
+            conn = self._require_conn()
+            cur = conn.execute(
+                "UPDATE tasks SET status = ?, updated_at = ?"
+                " WHERE id = ?",
+                (status, _utcnow(), task_id))
+            conn.commit()
+            return cur.rowcount == 1
+
+    def put_goal(self, user_id: str, goal_id: str, title: str,
+                 plan: Mapping[str, object], status: str) -> None:
+        """写/覆盖一个目标（plan 过无凭证守卫）。"""
+        assert_no_secrets(plan)
+        now = _utcnow()
+        blob = json.dumps(dict(plan), sort_keys=True)
+        with self._lock:
+            conn = self._require_conn()
+            conn.execute(
+                "INSERT INTO goals(id, user_id, title, plan_json,"
+                " status, created_at, updated_at)"
+                " VALUES (?, ?, ?, ?, ?, ?, ?)"
+                " ON CONFLICT(id) DO UPDATE SET"
+                " title = excluded.title,"
+                " plan_json = excluded.plan_json,"
+                " status = excluded.status,"
+                " updated_at = excluded.updated_at",
+                (goal_id, user_id, title, blob, status, now, now))
+            conn.commit()
+
+    def get_goal(self, goal_id: str) -> dict[str, object] | None:
+        """取一个目标，不存在返回 None。"""
+        with self._lock:
+            conn = self._require_conn()
+            row = conn.execute(
+                "SELECT id, user_id, title, plan_json, status,"
+                " created_at, updated_at FROM goals WHERE id = ?",
+                (goal_id,)).fetchone()
+            return _row_to_goal(row) if row else None
+
+    def list_goals(self, user_id: str) -> list[dict[str, object]]:
+        """列某用户的目标（按创建时间升序）。"""
+        with self._lock:
+            conn = self._require_conn()
+            rows = conn.execute(
+                "SELECT id, user_id, title, plan_json, status,"
+                " created_at, updated_at FROM goals"
+                " WHERE user_id = ? ORDER BY created_at",
+                (user_id,)).fetchall()
+            return [_row_to_goal(row) for row in rows]
+
+    def put_session(self, user_id: str, session_id: str, chat_id: str,
+                    summary: str) -> None:
+        """写/覆盖一个会话摘要（summary 过无凭证守卫）。"""
+        assert_no_secrets({"summary": summary})
+        now = _utcnow()
+        with self._lock:
+            conn = self._require_conn()
+            conn.execute(
+                "INSERT INTO sessions(id, user_id, chat_id, summary,"
+                " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)"
+                " ON CONFLICT(id) DO UPDATE SET"
+                " summary = excluded.summary,"
+                " updated_at = excluded.updated_at",
+                (session_id, user_id, chat_id, summary, now, now))
+            conn.commit()
+
+    def get_session(self, session_id: str) -> dict[str, object] | None:
+        """取一个会话，不存在返回 None。"""
+        with self._lock:
+            conn = self._require_conn()
+            row = conn.execute(
+                "SELECT id, user_id, chat_id, summary, created_at,"
+                " updated_at FROM sessions WHERE id = ?",
+                (session_id,)).fetchone()
+            if row is None:
+                return None
+            return {"id": row[0], "user_id": row[1], "chat_id": row[2],
+                    "summary": row[3], "created_at": row[4],
+                    "updated_at": row[5]}
+
+    def append_audit(self, actor: str, action: str, target: str,
+                     detail: Mapping[str, object]) -> str:
+        """追加一条审计记录（只追加，无删除/修改接口）。"""
+        assert_no_secrets(detail)
+        record_id = _new_id()
+        blob = json.dumps(dict(detail), sort_keys=True)
+        with self._lock:
+            conn = self._require_conn()
+            conn.execute(
+                "INSERT INTO audit(id, actor, action, target,"
+                " detail_json, created_at) VALUES (?, ?, ?, ?, ?, ?)",
+                (record_id, actor, action, target, blob, _utcnow()))
+            conn.commit()
+        return record_id
+
+    def list_audit(self, target: str | None = None,
+                   limit: int = 100) -> list[dict[str, object]]:
+        """列审计记录（最新在前；target 可过滤）。"""
+        with self._lock:
+            conn = self._require_conn()
+            if target is None:
+                rows = conn.execute(
+                    "SELECT id, actor, action, target, detail_json,"
+                    " created_at FROM audit ORDER BY created_at DESC"
+                    " LIMIT ?", (limit,)).fetchall()
+            else:
+                rows = conn.execute(
+                    "SELECT id, actor, action, target, detail_json,"
+                    " created_at FROM audit WHERE target = ?"
+                    " ORDER BY created_at DESC LIMIT ?",
+                    (target, limit)).fetchall()
+            return [_row_to_audit(row) for row in rows]
+
+    def backup(self, dest_dir: str) -> dict[str, str]:
+        """在线备份到目录：库文件 + manifest（sha256 + 行数 + 版本）。"""
+        with self._lock:
+            conn = self._require_conn()
+            counts = self.counts()
+            os.makedirs(dest_dir, mode=0o700, exist_ok=True)
+            backup_path = os.path.join(dest_dir, _BACKUP_FILENAME)
+            if os.path.exists(backup_path):
+                os.remove(backup_path)
+            target = sqlite3.connect(backup_path)
+            try:
+                conn.backup(target)
+            finally:
+                target.close()
+            os.chmod(backup_path, 0o600)
+            manifest = {
+                "schema_version": SCHEMA_VERSION,
+                "sha256": _sha256_file(backup_path),
+                "tasks": str(counts["tasks"]),
+                "goals": str(counts["goals"]),
+                "sessions": str(counts["sessions"]),
+                "audit": str(counts["audit"]),
+                "created_at": _utcnow(),
+            }
+            atomic_write_text(
+                os.path.join(dest_dir, _MANIFEST_FILENAME),
+                json.dumps(manifest, sort_keys=True, indent=2) + "\n")
+            return manifest
+
+
+def _row_to_task(row: tuple) -> dict[str, object]:
+    """tasks 行 → dict（payload_json 解析回 dict）。"""
+    return {"id": row[0], "user_id": row[1], "kind": row[2],
+            "status": row[3],
+            "payload": json.loads(row[4]),
+            "created_at": row[5], "updated_at": row[6]}
+
+
+def _row_to_goal(row: tuple) -> dict[str, object]:
+    """goals 行 → dict（plan_json 解析回 dict）。"""
+    return {"id": row[0], "user_id": row[1], "title": row[2],
+            "plan": json.loads(row[3]), "status": row[4],
+            "created_at": row[5], "updated_at": row[6]}
+
+
+def _row_to_audit(row: tuple) -> dict[str, object]:
+    """audit 行 → dict（detail_json 解析回 dict）。"""
+    return {"id": row[0], "actor": row[1], "action": row[2],
+            "target": row[3], "detail": json.loads(row[4]),
+            "created_at": row[5]}

cloud/chv/memory_files.py

@@ -0,0 +1,343 @@
+"""客机记忆文件接线（对齐 Muse P7，叶子模块）。
+
+Muse 官方要求（security.muse.ai / 设计文 2026-09-08，见
+docs/chv-muse-p7-gap.md §1）：记忆跨对话持续存在；记忆文件可
+直接阅读与编辑；用户可随时让 Muse“忘记”指定内容；VM 数据持续
+备份以便恢复；VM 是用户放入 Muse 的一切内容的系统记录。
+
+本模块只做接线，不重造记忆引擎：
+- 记忆引擎 = 记忆系统 v13（~/workspace/memory-system），经真实
+  CLI 调用（`node dist/cli.js search|get|explain|forget-plan|
+  forget-confirm|backup|restore|export`）。CLI 不可用时降级为本地
+  Markdown 模式，bank 忘记明确记为待办，不谎称已忘。
+- 记忆文件布局与记忆系统同款：家目录下 MEMORY.md + memory/*.md；
+  宿主侧镜像在 `state_dir/<vm>/memory-home/`，客内家目录在
+  `/home/chv/memory`（workspace 之外，不同步到 Gitea）。
+- “忘记”是尽力删除 + 验证查不到，不是密码学擦除：已备份的快照
+  里仍有旧内容（与记忆系统 CLI 的 backup 语义一致），差距表 §5
+  写清。
+"""
+
+import json
+import os
+import subprocess
+import threading
+from collections.abc import Mapping
+
+from cloud.chv.app_state import assert_no_secrets
+
+GUEST_MEMORY_HOME = "/home/chv/memory"
+_MEMORY_MD = "MEMORY.md"
+
+
+def memory_home_path(state_dir: str, vm_name: str) -> str:
+    """某 VM 记忆家目录镜像路径：宿主 state_dir 下，不在客机磁盘里。"""
+    return os.path.join(state_dir, vm_name, "memory-home")
+_MEMORY_SUBDIR = "memory"
+_AUDIT_FILENAME = "audit.jsonl"
+_CLI_TIMEOUT_SECONDS = 120
+
+
+class MemoryError(RuntimeError):
+    """记忆文件操作失败（路径穿越、CLI 失败等）。"""
+
+
+def _is_within(base: str, path: str) -> bool:
+    """path 是否在家目录内（防路径穿越）。"""
+    base_abs = os.path.abspath(base)
+    path_abs = os.path.abspath(path)
+    return path_abs == base_abs or path_abs.startswith(base_abs + os.sep)
+
+
+def _remove_lines_containing(path: str, needle: str) -> int:
+    """删掉文件里含 needle 的行（大小写不敏感），返回删掉的行数。"""
+    if not os.path.isfile(path):
+        return 0
+    with open(path, "r", encoding="utf-8") as handle:
+        lines = handle.readlines()
+    kept = [line for line in lines if needle.lower() not in line.lower()]
+    removed = len(lines) - len(kept)
+    if removed:
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.writelines(kept)
+    return removed
+
+
+def _audit_line(actor: str, action: str, target: str,
+                detail: Mapping[str, object]) -> str:
+    """一条审计 JSON 行（detail 过无凭证守卫）。"""
+    assert_no_secrets(detail)
+    record = {"actor": actor, "action": action, "target": target,
+              "detail": dict(detail)}
+    return json.dumps(record, sort_keys=True) + "\n"
+
+
+class MemoryFiles:
+    """宿主侧记忆家目录镜像：读/列/写/追加，写操作记审计。"""
+
+    def __init__(self, home: str) -> None:
+        self._home = home
+        self._lock = threading.RLock()
+
+    @property
+    def home(self) -> str:
+        """镜像家目录路径。"""
+        return self._home
+
+    def ensure(self) -> None:
+        """建家目录结构（MEMORY.md + memory/，0700）。"""
+        os.makedirs(os.path.join(self._home, _MEMORY_SUBDIR),
+                    mode=0o700, exist_ok=True)
+        memory_md = os.path.join(self._home, _MEMORY_MD)
+        if not os.path.exists(memory_md):
+            with open(memory_md, "w", encoding="utf-8") as handle:
+                handle.write("# MEMORY.md\n")
+            os.chmod(memory_md, 0o600)
+        os.chmod(self._home, 0o700)
+
+    def resolve(self, rel: str) -> str:
+        """相对路径 → 家目录内绝对路径，穿越即拒。"""
+        candidate = os.path.abspath(os.path.join(self._home, rel))
+        if not _is_within(self._home, candidate):
+            raise MemoryError("path escapes memory home: %s" % rel)
+        return candidate
+
+    def list_files(self) -> list[str]:
+        """列家目录下全部文件（相对路径，排序）。"""
+        found: list[str] = []
+        for root, _dirs, names in os.walk(self._home):
+            for name in names:
+                full = os.path.join(root, name)
+                found.append(os.path.relpath(full, self._home))
+        return sorted(found)
+
+    def read_file(self, rel: str) -> str:
+        """读一个记忆文件（相对路径），不存在即报错。"""
+        path = self.resolve(rel)
+        if not os.path.isfile(path):
+            raise MemoryError("memory file missing: %s" % rel)
+        with open(path, "r", encoding="utf-8") as handle:
+            return handle.read()
+
+    def write_file(self, rel: str, text: str, actor: str = "host") -> None:
+        """写一个记忆文件（相对路径，0600），记审计。"""
+        assert_no_secrets({"text": text})
+        path = self.resolve(rel)
+        parent = os.path.dirname(path)
+        if parent:
+            os.makedirs(parent, mode=0o700, exist_ok=True)
+        with self._lock:
+            with open(path, "w", encoding="utf-8") as handle:
+                handle.write(text)
+            os.chmod(path, 0o600)
+            self._append_audit(actor, "write", rel, {"bytes": len(text)})
+
+    def append_lines(self, rel: str, lines: list[str],
+                     actor: str = "host") -> None:
+        """向一个记忆文件追加行（0600），记审计。"""
+        assert_no_secrets({"lines": lines})
+        path = self.resolve(rel)
+        parent = os.path.dirname(path)
+        if parent:
+            os.makedirs(parent, mode=0o700, exist_ok=True)
+        with self._lock:
+            with open(path, "a", encoding="utf-8") as handle:
+                for line in lines:
+                    handle.write(line if line.endswith("\n") else line + "\n")
+            os.chmod(path, 0o600)
+            self._append_audit(actor, "append", rel,
+                               {"lines": len(lines)})
+
+    def remove_lines(self, rel: str, needle: str,
+                     actor: str = "host") -> int:
+        """删掉一个记忆文件里含 needle 的行，返回行数，记审计。"""
+        path = self.resolve(rel)
+        with self._lock:
+            removed = _remove_lines_containing(path, needle)
+            if removed:
+                self._append_audit(actor, "forget-lines", rel,
+                                   {"needle": needle, "removed": removed})
+            return removed
+
+    def _append_audit(self, actor: str, action: str, target: str,
+                      detail: Mapping[str, object]) -> None:
+        """审计追加到家目录 memory/audit.jsonl（O_APPEND）。"""
+        audit_path = os.path.join(self._home, _MEMORY_SUBDIR,
+                                  _AUDIT_FILENAME)
+        with open(audit_path, "a", encoding="utf-8") as handle:
+            handle.write(_audit_line(actor, action, target, detail))
+
+
+class MemorySystemClient:
+    """记忆系统 CLI 客户端（subprocess 调真实 CLI，不重造引擎）。
+
+    cli_path 指向 `memory-system/dist/cli.js`，node_path 指向 node
+    可执行文件。所有调用都有超时；非零退出即 MemoryError（stderr
+    截断 500 字符，绝不含凭证值——CLI 输出本就不含凭证）。
+    """
+
+    def __init__(self, cli_path: str, node_path: str = "node") -> None:
+        self._cli_path = cli_path
+        self._node_path = node_path
+
+    @property
+    def cli_path(self) -> str:
+        """CLI 脚本路径。"""
+        return self._cli_path
+
+    def available(self) -> bool:
+        """CLI 脚本与 node 是否都存在。"""
+        import shutil
+        if not os.path.isfile(self._cli_path):
+            return False
+        return shutil.which(self._node_path) is not None
+
+    def _run(self, args: list[str],
+             timeout: int = _CLI_TIMEOUT_SECONDS) -> dict[str, object]:
+        """跑一次 CLI，stdout 解析为 JSON dict。"""
+        cmd = [self._node_path, self._cli_path] + args
+        try:
+            proc = subprocess.run(cmd, capture_output=True, text=True,
+                                  timeout=timeout)
+        except subprocess.TimeoutExpired as exc:
+            raise MemoryError("memory cli timeout: %s" % " ".join(args[:2]))
+        if proc.returncode != 0:
+            raise MemoryError("memory cli failed (%d): %s" % (
+                proc.returncode, proc.stderr.strip()[:500]))
+        try:
+            parsed = json.loads(proc.stdout)
+        except json.JSONDecodeError as exc:
+            raise MemoryError("memory cli bad json: %s" % exc)
+        if not isinstance(parsed, dict):
+            raise MemoryError("memory cli bad json: not an object")
+        return parsed
+
+    def search(self, home: str, *queries: str) -> dict[str, object]:
+        """词汇检索（CLI search），返回 {"hits": [...]}。"""
+        return self._run(["search", home] + list(queries))
+
+    def search_semantic(self, home: str, query: str) -> dict[str, object]:
+        """混合语义 + 词汇检索（无 embedder 时退化为纯词汇）。"""
+        return self._run(["search-semantic", home, query])
+
+    def get(self, home: str, citation: str) -> dict[str, object]:
+        """按引用取一条记忆（MEMORY.md#L7 或 memory/...#L7-L9）。"""
+        return self._run(["get", home, citation])
+
+    def explain(self, home: str, claim: str) -> dict[str, object]:
+        """解释一条 claim 的来源与替代关系。"""
+        return self._run(["explain", home, claim])
+
+    def forget_plan(self, home: str, query: str) -> dict[str, object]:
+        """忘记计划：列出 query 命中的 claims，不删任何东西。"""
+        return self._run(["forget-plan", home, query])
+
+    def forget_confirm(self, home: str, query: str,
+                       actor: str = "cli") -> dict[str, object]:
+        """确认忘记：删命中的 claims，记审计，返回结果。"""
+        return self._run(["forget-confirm", home, query,
+                          "--actor", actor])
+
+    def backup(self, home: str,
+               out: str | None = None) -> dict[str, object]:
+        """整 home 快照备份（文件级 + manifest sha256）。"""
+        args = ["backup", home]
+        if out is not None:
+            args += ["--out", out]
+        return self._run(args)
+
+    def restore(self, home: str, snapshot: str,
+                actor: str = "cli") -> dict[str, object]:
+        """从快照恢复（先验 manifest，失败则 home 不动）。"""
+        return self._run(["restore", home, snapshot,
+                          "--actor", actor])
+
+    def export(self, home: str, out_dir: str) -> dict[str, object]:
+        """导出全部记忆为 Markdown 目录。"""
+        return self._run(["export", home, out_dir])
+
+
+def build_pull_command(ssh_cmd: list[str], remote_home: str) -> list[str]:
+    """构造拉取命令：ssh … tar -czf - -C <父目录> <家目录名>。"""
+    parent = os.path.dirname(remote_home.rstrip("/"))
+    base = os.path.basename(remote_home.rstrip("/"))
+    return list(ssh_cmd) + ["tar -czf - -C %s %s" % (parent, base)]
+
+
+def pull_memory_home(ssh_cmd: list[str], remote_home: str,
+                     local_home: str,
+                     timeout: int = _CLI_TIMEOUT_SECONDS) -> dict[str, int]:
+    """经 SSH + tar 把客机记忆家目录拉到宿主镜像（覆盖式）。
+
+    ssh_cmd 是已构造好的 ssh 基础命令（端口/密钥/用户已定，见
+    cloud.chv.ssh_transport.build_ssh_command）。返回拉取到的
+    文件数；ssh 或 tar 失败即 MemoryError。
+    """
+    os.makedirs(local_home, mode=0o700, exist_ok=True)
+    pull_cmd = build_pull_command(ssh_cmd, remote_home)
+    fetch = subprocess.Popen(pull_cmd, stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE)
+    try:
+        extract = subprocess.run(
+            ["tar", "-xzf", "-", "-C", local_home],
+            stdin=fetch.stdout, capture_output=True, text=True,
+            timeout=timeout)
+    finally:
+        if fetch.stdout is not None:
+            fetch.stdout.close()
+    fetch_rc = fetch.wait()
+    err = ""
+    if fetch.stderr is not None:
+        err = fetch.stderr.read().decode("utf-8", "replace")[:500]
+        fetch.stderr.close()
+    if fetch_rc != 0:
+        raise MemoryError("memory pull ssh failed (%d): %s" % (fetch_rc, err))
+    if extract.returncode != 0:
+        raise MemoryError("memory pull tar failed: %s"
+                          % extract.stderr.strip()[:500])
+    count = 0
+    for _root, _dirs, names in os.walk(local_home):
+        count += len(names)
+    return {"files": count}
+
+
+def forget(memory_files: MemoryFiles,
+           client: MemorySystemClient | None, query: str,
+           actor: str = "host") -> dict[str, object]:
+    """忘记指定内容：bank 忘记 + 本地 Markdown 清行 + 验证查不到。
+
+    client 在位时走真实 CLI（plan → confirm → 验证 search 为空）；
+    client 缺位时只做本地清行，bank 忘记记为待办（不谎称已忘）。
+    返回 {"mode", "claims", "lines_removed", "verified"}。
+    """
+    assert_no_secrets({"query": query})
+    if client is not None and client.available():
+        plan = client.forget_plan(memory_files.home, query)
+        claims = plan.get("claims")
+        claim_list = claims if isinstance(claims, list) else []
+        client.forget_confirm(memory_files.home, query, actor=actor)
+        removed = _forget_local_lines(memory_files, query, actor)
+        result = client.search(memory_files.home, query)
+        hits = result.get("hits")
+        hit_list = hits if isinstance(hits, list) else []
+        if hit_list:
+            raise MemoryError(
+                "forget verify failed: %d hits remain" % len(hit_list))
+        return {"mode": "cli", "claims": len(claim_list),
+                "lines_removed": removed, "verified": True}
+    removed = _forget_local_lines(memory_files, query, actor)
+    memory_files._append_audit(actor, "forget-pending", query,
+                               {"reason": "cli-unavailable",
+                                "lines_removed": removed})
+    return {"mode": "local-only", "claims": 0,
+            "lines_removed": removed, "verified": False}
+
+
+def _forget_local_lines(memory_files: MemoryFiles, query: str,
+                        actor: str) -> int:
+    """从镜像家目录全部 Markdown 里清掉含 query 的行。"""
+    removed = 0
+    for rel in memory_files.list_files():
+        if rel.endswith(".md"):
+            removed += memory_files.remove_lines(rel, query, actor=actor)
+    return removed

cloud/tests/test_chv_muse_p7.py

@@ -0,0 +1,502 @@
+"""对齐 Muse P7 测试：状态与记忆分离。
+
+只测本轨交付物（cloud/chv/app_state.py、memory_files.py、
+chv-guest/chv-memory.py、build.sh 干跑、buildinfo.py 内容项）。
+真实记忆系统 CLI 只在本地临时 home 上跑（init/write/search/
+forget/backup/restore 全是真调用）；其余用假 ssh/假 CLI。
+绝不连真实网络、不起真实 VM、不改生产行为。
+"""
+import importlib.util
+import io
+import json
+import os
+import shutil
+import sqlite3
+import subprocess
+import sys
+import tarfile
+import tempfile
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+WORKSPACE = Path(__file__).resolve().parents[3]
+MEMSYS_CLI = WORKSPACE / "memory-system" / "dist" / "cli.js"
+
+sys.path.insert(0, str(ROOT))
+
+from cloud.chv import app_state
+from cloud.chv import memory_files
+from cloud.chv.app_state import (AppStateStore, SecretRejectedError,
+                                 restore_app_state, state_db_path,
+                                 verify_backup)
+from cloud.chv.memory_files import (GUEST_MEMORY_HOME, MemoryFiles,
+                                    MemorySystemClient,
+                                    build_pull_command, forget,
+                                    memory_home_path, pull_memory_home)
+
+
+def _load_guest_script(name):
+    """把 chv-guest 下的独立脚本当模块加载（无包结构）。"""
+    path = ROOT / "chv-guest" / name
+    spec = importlib.util.spec_from_file_location(
+        "chv_guest_p7_" + name[:-3], str(path))
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+CHV_MEMORY = _load_guest_script("chv-memory.py")
+BUILDINFO = _load_guest_script("buildinfo.py")
+
+
+def _has_cli():
+    """真实记忆系统 CLI 是否在位。"""
+    return os.path.isfile(str(MEMSYS_CLI)) and shutil.which("node")
+
+
+class SecretGuardTest(unittest.TestCase):
+    def test_key_name_token_rejected(self):
+        with self.assertRaises(SecretRejectedError) as ctx:
+            app_state.assert_no_secrets({"GITEA_TOKEN": "abc"})
+        self.assertNotIn("abc", str(ctx.exception))
+
+    def test_key_name_password_rejected(self):
+        with self.assertRaises(SecretRejectedError):
+            app_state.assert_no_secrets({"nested": {"db_password": "x"}})
+
+    def test_value_pattern_ghp_rejected(self):
+        with self.assertRaises(SecretRejectedError) as ctx:
+            app_state.assert_no_secrets(
+                {"note": "see ghp_" + "a" * 24 + " here"})
+        self.assertNotIn("ghp_", str(ctx.exception))
+
+    def test_plain_text_passes(self):
+        app_state.assert_no_secrets(
+            {"title": "买菜", "note": "记得买冰美式豆子"})
+
+
+class AppStateStoreTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.path = os.path.join(self.tmp.name, "vm1", "app-state.sqlite")
+        self.store = AppStateStore(self.path)
+        self.store.open()
+
+    def tearDown(self):
+        self.store.close()
+        self.tmp.cleanup()
+
+    def test_open_sets_schema_and_modes(self):
+        self.assertEqual(self.store.schema_version(), "1")
+        self.assertEqual(oct(os.stat(self.path).st_mode & 0o777), "0o600")
+        parent = os.path.dirname(self.path)
+        self.assertEqual(oct(os.stat(parent).st_mode & 0o777), "0o700")
+
+    def test_state_db_path_helper(self):
+        self.assertEqual(state_db_path("/s", "vm1"),
+                         "/s/vm1/app-state.sqlite")
+
+    def test_task_roundtrip(self):
+        self.store.put_task("u1", "t1", "code", "running",
+                            {"repo": "admin/x", "step": "build"})
+        got = self.store.get_task("t1")
+        self.assertEqual(got["status"], "running")
+        self.assertEqual(got["payload"]["repo"], "admin/x")
+        self.assertIsNone(self.store.get_task("nope"))
+        listed = self.store.list_tasks("u1")
+        self.assertEqual([t["id"] for t in listed], ["t1"])
+        self.assertEqual(self.store.list_tasks("u1", "done"), [])
+
+    def test_update_task_status(self):
+        self.store.put_task("u1", "t1", "code", "running", {})
+        self.assertTrue(self.store.update_task_status("t1", "done"))
+        self.assertEqual(self.store.get_task("t1")["status"], "done")
+        self.assertFalse(self.store.update_task_status("nope", "done"))
+
+    def test_goal_and_session_roundtrip(self):
+        self.store.put_goal("u1", "g1", "学会做饭",
+                            {"steps": ["买菜", "开火"]}, "active")
+        goal = self.store.get_goal("g1")
+        self.assertEqual(goal["title"], "学会做饭")
+        self.assertEqual(goal["plan"]["steps"][0], "买菜")
+        self.assertEqual(len(self.store.list_goals("u1")), 1)
+        self.store.put_session("u1", "s1", "chat-9", "聊了晚饭")
+        session = self.store.get_session("s1")
+        self.assertEqual(session["chat_id"], "chat-9")
+        self.assertIsNone(self.store.get_session("nope"))
+
+    def test_task_with_secret_key_refused_and_not_stored(self):
+        with self.assertRaises(SecretRejectedError):
+            self.store.put_task("u1", "t2", "code", "running",
+                                {"GITEA_TOKEN": "abc"})
+        self.assertIsNone(self.store.get_task("t2"))
+
+    def test_task_with_secret_value_refused(self):
+        with self.assertRaises(SecretRejectedError):
+            self.store.put_task("u1", "t2", "code", "running",
+                                {"note": "key sk-" + "b" * 24})
+
+    def test_audit_append_only(self):
+        rid = self.store.append_audit("host", "task-start", "t1",
+                                      {"kind": "code"})
+        self.assertTrue(rid)
+        rows = self.store.list_audit(target="t1")
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]["action"], "task-start")
+        source = Path(app_state.__file__).read_text()
+        self.assertNotIn("DELETE FROM audit", source)
+        self.assertNotIn("UPDATE audit", source)
+
+    def test_audit_detail_secret_refused(self):
+        with self.assertRaises(SecretRejectedError):
+            self.store.append_audit("host", "x", "t1",
+                                    {"token": "ghp_" + "c" * 24})
+
+    def test_counts(self):
+        self.store.put_task("u1", "t1", "code", "running", {})
+        self.store.put_goal("u1", "g1", "标题", {}, "active")
+        self.assertEqual(self.store.counts()["tasks"], 1)
+        self.assertEqual(self.store.counts()["goals"], 1)
+        self.assertEqual(self.store.counts()["audit"], 0)
+
+    def test_ddl_has_no_sqlite_only_syntax(self):
+        self.assertNotIn("AUTOINCREMENT", app_state.DDL)
+        self.assertNotIn("PRAGMA", app_state.DDL)
+        for keyword in ("CREATE TABLE IF NOT EXISTS meta",
+                        "CREATE TABLE IF NOT EXISTS tasks",
+                        "CREATE TABLE IF NOT EXISTS goals",
+                        "CREATE TABLE IF NOT EXISTS sessions",
+                        "CREATE TABLE IF NOT EXISTS audit"):
+            self.assertIn(keyword, app_state.DDL)
+
+    def test_backup_restore_roundtrip(self):
+        self.store.put_task("u1", "t1", "code", "done", {"n": "1"})
+        self.store.append_audit("host", "done", "t1", {})
+        snap = os.path.join(self.tmp.name, "snap")
+        manifest = self.store.backup(snap)
+        self.assertEqual(manifest["tasks"], "1")
+        self.assertEqual(verify_backup(snap)["sha256"],
+                         manifest["sha256"])
+        self.store.close()
+        os.remove(self.path)
+        restore_app_state(snap, self.path)
+        self.store.open()
+        self.assertEqual(self.store.get_task("t1")["status"], "done")
+        self.assertEqual(self.store.counts()["audit"], 1)
+
+    def test_restore_keeps_pre_restore_copy(self):
+        self.store.put_task("u1", "t1", "code", "running", {})
+        snap = os.path.join(self.tmp.name, "snap")
+        self.store.backup(snap)
+        self.store.close()
+        restore_app_state(snap, self.path)
+        preserved = [p for p in os.listdir(os.path.dirname(self.path))
+                     if p.startswith("app-state.sqlite.pre-restore-")]
+        self.assertEqual(len(preserved), 1)
+
+    def test_restore_corrupt_backup_refused(self):
+        self.store.put_task("u1", "t1", "code", "running", {})
+        snap = os.path.join(self.tmp.name, "snap")
+        self.store.backup(snap)
+        self.store.close()
+        with open(os.path.join(snap, "app-state.sqlite"),
+                  "ab") as handle:
+            handle.write(b"corrupt")
+        with self.assertRaises(app_state.AppStateError):
+            restore_app_state(snap, self.path)
+        self.store.open()
+        self.assertEqual(self.store.get_task("t1")["status"], "running")
+
+    def test_backup_has_no_token_bytes(self):
+        sentinel = "ghp_" + "z" * 24
+        with self.assertRaises(SecretRejectedError):
+            self.store.put_task("u1", "t9", "code", "running",
+                                {"note": sentinel})
+        snap = os.path.join(self.tmp.name, "snap")
+        self.store.backup(snap)
+        blob = Path(os.path.join(snap, "app-state.sqlite")).read_bytes()
+        self.assertNotIn(b"ghp_", blob)
+        manifest = Path(os.path.join(snap, "manifest.json")).read_bytes()
+        self.assertNotIn(b"ghp_", manifest)
+
+    def test_store_not_open_raises(self):
+        closed = AppStateStore(os.path.join(self.tmp.name, "x.db"))
+        with self.assertRaises(app_state.AppStateError):
+            closed.get_task("t1")
+
+
+class SeparationTest(unittest.TestCase):
+    def test_store_path_outside_guest_disk(self):
+        db = state_db_path("/state", "vm1")
+        disk = "/state/vm1/disk.raw"
+        self.assertNotEqual(db, disk)
+        self.assertFalse(db.startswith(disk + "/"))
+
+    def test_memory_home_path_outside_guest_disk(self):
+        home = memory_home_path("/state", "vm1")
+        self.assertEqual(home, "/state/vm1/memory-home")
+        self.assertNotEqual(home, "/state/vm1/disk.raw")
+
+    def test_guest_script_has_no_state_db_reference(self):
+        source = (ROOT / "chv-guest" / "chv-memory.py").read_text()
+        self.assertNotIn("app-state", source)
+        self.assertNotIn("app_state", source)
+
+    def test_app_state_has_no_guest_paths(self):
+        source = Path(app_state.__file__).read_text()
+        self.assertNotIn("/home/chv", source)
+        self.assertNotIn("GUEST_MEMORY_HOME", source)
+
+    def test_guest_list_cannot_see_store_file(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            home = os.path.join(tmp, "memory")
+            store = os.path.join(tmp, "app-state.sqlite")
+            Path(store).write_text("fake-db")
+            CHV_MEMORY._ensure(home)
+            listed = CHV_MEMORY.cmd_list(home)["files"]
+            self.assertNotIn("app-state.sqlite", listed)
+            self.assertNotIn("../app-state.sqlite", listed)
+
+class MemoryFilesTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.home = os.path.join(self.tmp.name, "memory-home")
+        self.files = MemoryFiles(self.home)
+        self.files.ensure()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_ensure_layout(self):
+        self.assertTrue(os.path.isfile(
+            os.path.join(self.home, "MEMORY.md")))
+        self.assertTrue(os.path.isdir(
+            os.path.join(self.home, "memory")))
+        self.assertEqual(oct(os.stat(self.home).st_mode & 0o777), "0o700")
+
+    def test_memory_home_path_helper(self):
+        self.assertEqual(memory_home_path("/s", "vm1"),
+                         "/s/vm1/memory-home")
+
+    def test_write_read_list_append(self):
+        self.files.write_file("MEMORY.md", "# 记忆\n", actor="host")
+        self.files.append_lines("memory/2026-11-07.md",
+                                ["- 喜欢冰美式", "- 怕辣"], actor="host")
+        self.assertIn("MEMORY.md", self.files.list_files())
+        self.assertIn("memory/2026-11-07.md", self.files.list_files())
+        text = self.files.read_file("memory/2026-11-07.md")
+        self.assertIn("冰美式", text)
+        audit = Path(os.path.join(
+            self.home, "memory", "audit.jsonl")).read_text()
+        self.assertIn('"action": "write"', audit)
+        self.assertIn('"action": "append"', audit)
+
+    def test_read_missing_raises(self):
+        with self.assertRaises(memory_files.MemoryError):
+            self.files.read_file("memory/nope.md")
+
+    def test_path_traversal_rejected(self):
+        for bad in ("../escape.md", "../../etc/passwd",
+                    "/etc/passwd", "memory/../../escape.md"):
+            with self.assertRaises(memory_files.MemoryError, msg=bad):
+                self.files.read_file(bad)
+            with self.assertRaises(memory_files.MemoryError, msg=bad):
+                self.files.write_file(bad, "x")
+
+    def test_write_secret_refused(self):
+        with self.assertRaises(SecretRejectedError):
+            self.files.write_file("MEMORY.md", "token ghp_" + "d" * 24)
+
+    def test_remove_lines(self):
+        self.files.write_file("MEMORY.md",
+                              "# 记忆\n- 喜欢冰美式\n- 怕辣\n")
+        removed = self.files.remove_lines("MEMORY.md", "冰美式")
+        self.assertEqual(removed, 1)
+        self.assertNotIn("冰美式", self.files.read_file("MEMORY.md"))
+        self.assertIn("怕辣", self.files.read_file("MEMORY.md"))
+
+
+class ForgetLocalOnlyTest(unittest.TestCase):
+    def test_forget_local_only_marks_pending(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            files = MemoryFiles(os.path.join(tmp, "home"))
+            files.ensure()
+            files.write_file("MEMORY.md", "# 记忆\n- 喜欢冰美式\n")
+            result = forget(files, None, "冰美式", actor="host")
+            self.assertEqual(result["mode"], "local-only")
+            self.assertEqual(result["lines_removed"], 1)
+            self.assertFalse(result["verified"])
+            self.assertNotIn("冰美式", files.read_file("MEMORY.md"))
+            audit = Path(os.path.join(
+                tmp, "home", "memory", "audit.jsonl")).read_text()
+            self.assertIn("forget-pending", audit)
+
+
+@unittest.skipUnless(_has_cli(), "memory-system CLI not available")
+class RealCliForgetTest(unittest.TestCase):
+    def test_real_cli_write_search_forget_verify(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            home = os.path.join(tmp, "home")
+            client = MemorySystemClient(str(MEMSYS_CLI))
+            self.assertTrue(client.available())
+            client._run(["init", home])
+            written = client._run([
+                "write", home, "--kind", "fact",
+                "--text", "用户喜欢喝冰美式",
+                "--chat", "c1", "--message", "m1",
+                "--speaker", "user", "--quote", "我喜欢喝冰美式"])
+            claim_id = written.get("id")
+            self.assertTrue(str(claim_id).startswith("claim:"))
+            found = client.search(home, "冰美式")
+            hits = found.get("hits")
+            self.assertTrue(isinstance(hits, list) and len(hits) == 1)
+            files = MemoryFiles(home)
+            files.write_file("MEMORY.md",
+                             "# 记忆\n- 用户喜欢喝冰美式\n")
+            result = forget(files, client, "冰美式", actor="host")
+            self.assertEqual(result["mode"], "cli")
+            self.assertEqual(result["claims"], 1)
+            self.assertTrue(result["verified"])
+            after = client.search(home, "冰美式")
+            self.assertEqual(after.get("hits"), [])
+            self.assertNotIn("冰美式", files.read_file("MEMORY.md"))
+
+    def test_real_cli_backup_restore(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            home = os.path.join(tmp, "home")
+            client = MemorySystemClient(str(MEMSYS_CLI))
+            client._run(["init", home])
+            client._run(["write", home, "--kind", "fact",
+                         "--text", "用户住在上海", "--chat", "c1",
+                         "--message", "m1", "--speaker", "user",
+                         "--quote", "我住在上海"])
+            snap = client.backup(home)
+            snapshot = snap.get("snapshot") or snap.get("path") or ""
+            self.assertTrue(snapshot)
+            victim = os.path.join(home, "MEMORY.md")
+            with open(victim, "w", encoding="utf-8") as handle:
+                handle.write("# 已破坏\n")
+            client.restore(home, str(snapshot))
+            # 恢复后 MEMORY.md 回到快照内容（不再是破坏后的文本）；
+            # claim 本身活在 bank + 每日笔记里，search 能查到才是
+            # 恢复成功的证据。
+            self.assertNotIn("已破坏", Path(victim).read_text())
+            found = client.search(home, "上海")
+            self.assertTrue(len(found.get("hits", [])) >= 1)
+
+    def test_client_failure_raises(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            fake = os.path.join(tmp, "fake-cli.js")
+            Path(fake).write_text("#!/usr/bin/env node\n"
+                                  "process.exit(1);\n")
+            client = MemorySystemClient(fake)
+            with self.assertRaises(memory_files.MemoryError):
+                client.search(os.path.join(tmp, "home"), "x")
+
+
+class PullMemoryHomeTest(unittest.TestCase):
+    def test_build_pull_command(self):
+        cmd = build_pull_command(
+            ["ssh", "-p", "2222", "chv@127.0.0.1"], "/home/chv/memory")
+        self.assertEqual(cmd, ["ssh", "-p", "2222", "chv@127.0.0.1",
+                               "tar -czf - -C /home/chv memory"])
+
+    def test_pull_with_fake_ssh(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            remote_parent = os.path.join(tmp, "remote")
+            remote_home = os.path.join(remote_parent, "memory")
+            os.makedirs(os.path.join(remote_home, "memory"))
+            Path(os.path.join(remote_home, "MEMORY.md")).write_text(
+                "# 记忆\n")
+            Path(os.path.join(remote_home, "memory", "2026-11-07.md")
+                 ).write_text("- 喜欢冰美式\n")
+            fake_ssh = os.path.join(tmp, "fake-ssh")
+            Path(fake_ssh).write_text(
+                "#!/bin/sh\n"
+                "tar -czf - -C %s memory\n" % remote_parent)
+            os.chmod(fake_ssh, 0o755)
+            local = os.path.join(tmp, "local")
+            result = pull_memory_home([fake_ssh], GUEST_MEMORY_HOME,
+                                      local)
+            self.assertEqual(result["files"], 2)
+            self.assertIn("冰美式", Path(os.path.join(
+                local, "memory", "memory", "2026-11-07.md")).read_text())
+
+    def test_pull_ssh_failure_raises(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            fake_ssh = os.path.join(tmp, "fake-ssh")
+            Path(fake_ssh).write_text("#!/bin/sh\nexit 3\n")
+            os.chmod(fake_ssh, 0o755)
+            with self.assertRaises(memory_files.MemoryError):
+                pull_memory_home([fake_ssh], GUEST_MEMORY_HOME,
+                                 os.path.join(tmp, "local"))
+
+
+class GuestScriptTest(unittest.TestCase):
+    def test_list_read_append_forget_export(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            home = os.path.join(tmp, "memory")
+            self.assertEqual(
+                CHV_MEMORY.cmd_list(home)["files"], ["MEMORY.md"])
+            CHV_MEMORY.cmd_append(home, "MEMORY.md",
+                                  ["- 喜欢冰美式", "- 怕辣"])
+            read = CHV_MEMORY.cmd_read(home, "MEMORY.md")
+            self.assertIn("冰美式", read["text"])
+            forgot = CHV_MEMORY.cmd_forget(home, "冰美式")
+            self.assertEqual(forgot["removed"], 1)
+            self.assertNotIn("冰美式",
+                             CHV_MEMORY.cmd_read(home, "MEMORY.md")["text"])
+            dest = os.path.join(tmp, "export")
+            exported = CHV_MEMORY.cmd_export(home, dest)
+            self.assertEqual(exported["files"], 1)
+            self.assertTrue(os.path.isfile(
+                os.path.join(dest, "MEMORY.md")))
+
+    def test_main_list_prints_json(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            home = os.path.join(tmp, "memory")
+            buf = io.StringIO()
+            with redirect_stdout(buf):
+                rc = CHV_MEMORY.main(["--home", home, "list"])
+            self.assertEqual(rc, 0)
+            self.assertEqual(json.loads(buf.getvalue())["files"],
+                             ["MEMORY.md"])
+
+    def test_read_traversal_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            home = os.path.join(tmp, "memory")
+            with self.assertRaises(SystemExit):
+                CHV_MEMORY.cmd_read(home, "../escape.md")
+
+    def test_append_secret_refused(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            home = os.path.join(tmp, "memory")
+            with self.assertRaises(SystemExit):
+                CHV_MEMORY.cmd_append(home, "MEMORY.md",
+                                      ["token ghp_" + "e" * 24])
+
+    def test_export_excludes_audit_log(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            home = os.path.join(tmp, "memory")
+            CHV_MEMORY._ensure(home)
+            Path(os.path.join(home, "audit.jsonl")).write_text("{}\n")
+            dest = os.path.join(tmp, "export")
+            CHV_MEMORY.cmd_export(home, dest)
+            self.assertFalse(os.path.exists(
+                os.path.join(dest, "audit.jsonl")))
+
+
+class BuildIntegrationTest(unittest.TestCase):
+    def test_build_sh_copies_chv_memory(self):
+        text = (ROOT / "chv-guest" / "build.sh").read_text()
+        self.assertIn("chv-memory.py", text)
+        self.assertIn("/usr/local/bin/chv-memory", text)
+
+    def test_buildinfo_lists_chv_memory(self):
+        self.assertIn("chv-memory", BUILDINFO.BASE_CONTENTS)
+
+
+if __name__ == "__main__":
+    unittest.main()