diff --git a/chv-guest/build.sh b/chv-guest/build.sh
index 2896c87..bee38eb 100755
--- a/chv-guest/build.sh
+++ b/chv-guest/build.sh
@@ -166,6 +166,10 @@ if [ ! -f "$SCRIPT_DIR/chv-sync.sh" ]; then
   echo "缺少 chv-sync.sh（Gitea 双向同步脚本）：$SCRIPT_DIR" >&2
   exit 1
 fi
+if [ ! -f "$SCRIPT_DIR/chv-memory.py" ]; then
+  echo "缺少 chv-memory.py（客内记忆文件脚本）：$SCRIPT_DIR" >&2
+  exit 1
+fi
 if [ "$DRY_RUN" = "0" ]; then
   if [ ! -e /dev/kvm ]; then
     echo "CHV_GUEST_BUILD needs a KVM host: /dev/kvm 不存在" >&2
@@ -258,6 +262,8 @@ VC_ARGS=(-a "$OUT/disk.raw"
   --copy-in "$REPO_DIR/host.py:/app/"
   --copy-in "$SCRIPT_DIR/chv-sync.sh:/usr/local/bin/"
   --run-command 'mv /usr/local/bin/chv-sync.sh /usr/local/bin/chv-sync && chmod +x /usr/local/bin/chv-sync'
+  --copy-in "$SCRIPT_DIR/chv-memory.py:/usr/local/bin/"
+  --run-command 'mv /usr/local/bin/chv-memory.py /usr/local/bin/chv-memory && chmod +x /usr/local/bin/chv-memory'
   --run-command 'printf "runcmd:\n - systemctl restart ssh || service ssh restart || true\n - /usr/local/bin/chv-sync boot >>/var/log/chv-sync.log 2>&1 || true\n" > /etc/cloud/cloud.cfg.d/99-chv-sync.cfg'
   --run-command 'git config --system user.name chv && git config --system user.email chv@localhost && git config --system credential.helper store && git config --system init.defaultBranch main'
   --run-command "$GT_INSTALL_CMD"
diff --git a/chv-guest/buildinfo.py b/chv-guest/buildinfo.py
index 112b16c..ae439a3 100644
--- a/chv-guest/buildinfo.py
+++ b/chv-guest/buildinfo.py
@@ -36,7 +36,7 @@ BASE_CONTENTS = [
     "python3", "python3-pip", "curl", "ca-certificates",
     "xz-utils", "fonts-wqy-microhei", "fontconfig", "codex",
     "codex-code-mode-host", "codex-resources/bwrap",
-    "/app/host.py", "gt", "chv-sync",
+    "/app/host.py", "gt", "chv-sync", "chv-memory",
 ]
 
 # 执行器默认内核 cmdline（config.py），构建产物须与它一起记录（B-12），
diff --git a/chv-guest/chv-memory.py b/chv-guest/chv-memory.py
new file mode 100644
index 0000000..ed52d5b
--- /dev/null
+++ b/chv-guest/chv-memory.py
@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""客内记忆文件操作（对齐 Muse P7，纯标准库）。
+
+记忆家目录 /home/chv/memory（workspace 之外，不同步到 Gitea），
+布局与记忆系统同款：MEMORY.md + memory/*.md。由 executor 经
+SSH 调用（与 host.py 同级，跑在 nspawn cell 之外）；绝不含凭证
+值——写操作过本地守卫，命中疑似凭证模式即拒绝。
+
+用法：
+  chv-memory.py list [--home DIR]
+  chv-memory.py read <相对路径> [--home DIR]
+  chv-memory.py append <相对路径> <行...> [--home DIR]
+  chv-memory.py forget <needle> [--home DIR]
+  chv-memory.py export <目标目录> [--home DIR]
+"""
+
+import argparse
+import json
+import os
+import re
+import shutil
+import sys
+
+DEFAULT_HOME = "/home/chv/memory"
+
+_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-----"),
+)
+
+
+def _fail(message: str) -> "NoReturn":
+    """打印错误到 stderr 并以 1 退出。"""
+    print("chv-memory: %s" % message, file=sys.stderr)
+    raise SystemExit(1)
+
+
+def _check_no_secret(text: str) -> None:
+    """文本命中凭证模式即拒（不打印命中内容）。"""
+    for pattern in _SECRET_VALUE_PATTERNS:
+        if pattern.search(text):
+            _fail("refused: text looks like a credential")
+
+
+def _resolve(home: str, rel: str) -> str:
+    """相对路径 → 家目录内绝对路径，穿越即拒。"""
+    base = os.path.abspath(home)
+    candidate = os.path.abspath(os.path.join(base, rel))
+    if candidate != base and not candidate.startswith(base + os.sep):
+        _fail("path escapes memory home: %s" % rel)
+    return candidate
+
+
+def _ensure(home: str) -> None:
+    """建家目录结构（MEMORY.md + memory/）。"""
+    os.makedirs(os.path.join(home, "memory"), exist_ok=True)
+    memory_md = os.path.join(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")
+
+
+def cmd_list(home: str) -> dict:
+    """列家目录下全部文件（相对路径，排序）。"""
+    _ensure(home)
+    found = []
+    for root, _dirs, names in os.walk(home):
+        for name in names:
+            found.append(os.path.relpath(os.path.join(root, name), home))
+    return {"files": sorted(found)}
+
+
+def cmd_read(home: str, rel: str) -> dict:
+    """读一个记忆文件。"""
+    _ensure(home)
+    path = _resolve(home, rel)
+    if not os.path.isfile(path):
+        _fail("memory file missing: %s" % rel)
+    with open(path, "r", encoding="utf-8") as handle:
+        return {"path": rel, "text": handle.read()}
+
+
+def cmd_append(home: str, rel: str, lines: list) -> dict:
+    """向一个记忆文件追加行。"""
+    _ensure(home)
+    for line in lines:
+        _check_no_secret(line)
+    path = _resolve(home, rel)
+    parent = os.path.dirname(path)
+    if parent:
+        os.makedirs(parent, exist_ok=True)
+    with open(path, "a", encoding="utf-8") as handle:
+        for line in lines:
+            handle.write(line if line.endswith("\n") else line + "\n")
+    return {"path": rel, "appended": len(lines)}
+
+
+def cmd_forget(home: str, needle: str) -> dict:
+    """删掉全部 Markdown 里含 needle 的行（大小写不敏感）。"""
+    _ensure(home)
+    removed = 0
+    touched = []
+    for root, _dirs, names in os.walk(home):
+        for name in names:
+            if not name.endswith(".md"):
+                continue
+            path = os.path.join(root, name)
+            with open(path, "r", encoding="utf-8") as handle:
+                lines = handle.readlines()
+            kept = [ln for ln in lines
+                    if needle.lower() not in ln.lower()]
+            cut = len(lines) - len(kept)
+            if cut:
+                with open(path, "w", encoding="utf-8") as handle:
+                    handle.writelines(kept)
+                removed += cut
+                touched.append(os.path.relpath(path, home))
+    return {"needle": needle, "removed": removed, "files": touched}
+
+
+def cmd_export(home: str, dest: str) -> dict:
+    """导出全部记忆文件到目标目录（覆盖式，不含审计日志）。"""
+    _ensure(home)
+    os.makedirs(dest, exist_ok=True)
+    count = 0
+    for root, _dirs, names in os.walk(home):
+        for name in names:
+            if name == "audit.jsonl":
+                continue
+            src = os.path.join(root, name)
+            rel = os.path.relpath(src, home)
+            target = os.path.join(dest, rel)
+            os.makedirs(os.path.dirname(target) or dest, exist_ok=True)
+            shutil.copyfile(src, target)
+            count += 1
+    return {"dest": dest, "files": count}
+
+
+def build_parser() -> argparse.ArgumentParser:
+    """命令行解析器。"""
+    parser = argparse.ArgumentParser(prog="chv-memory.py")
+    parser.add_argument("--home", default=DEFAULT_HOME)
+    sub = parser.add_subparsers(dest="cmd", required=True)
+    sub.add_parser("list")
+    read_p = sub.add_parser("read")
+    read_p.add_argument("path")
+    append_p = sub.add_parser("append")
+    append_p.add_argument("path")
+    append_p.add_argument("lines", nargs="+")
+    forget_p = sub.add_parser("forget")
+    forget_p.add_argument("needle")
+    export_p = sub.add_parser("export")
+    export_p.add_argument("dest")
+    return parser
+
+
+def main(argv: list | None = None) -> int:
+    """入口：按子命令分发，打印 JSON 结果。"""
+    args = build_parser().parse_args(argv)
+    if args.cmd == "list":
+        result = cmd_list(args.home)
+    elif args.cmd == "read":
+        result = cmd_read(args.home, args.path)
+    elif args.cmd == "append":
+        result = cmd_append(args.home, args.path, args.lines)
+    elif args.cmd == "forget":
+        result = cmd_forget(args.home, args.needle)
+    elif args.cmd == "export":
+        result = cmd_export(args.home, args.dest)
+    else:  # pragma: no cover - argparse required=True 已挡住
+        _fail("unknown command: %s" % args.cmd)
+    print(json.dumps(result, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/cloud/chv/app_state.py b/cloud/chv/app_state.py
new file mode 100644
index 0000000..8a94354
--- /dev/null
+++ b/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]}
diff --git a/cloud/chv/memory_files.py b/cloud/chv/memory_files.py
new file mode 100644
index 0000000..1e474e9
--- /dev/null
+++ b/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
diff --git a/cloud/tests/test_chv_muse_p7.py b/cloud/tests/test_chv_muse_p7.py
new file mode 100644
index 0000000..b799937
--- /dev/null
+++ b/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()
diff --git a/docs/chv-muse-p7-gap.md b/docs/chv-muse-p7-gap.md
new file mode 100644
index 0000000..cd38f1e
--- /dev/null
+++ b/docs/chv-muse-p7-gap.md
@@ -0,0 +1,112 @@
+# 对齐 Muse P7 差距表：状态与记忆分离
+
+依据：`~/workspace/research_notes/meta-muse-ai-architecture-20261101-0755/report.md`
+（Meta 官方 security.muse.ai / 设计文 / 新闻室 2026-09-08 为准；第三方拆解只作参考，
+逆向细节不得写成官方架构）。基线：`06c3e17`。
+
+## §1 Muse 官方要求（本轨范围）
+
+| # | 官方要求 | 来源 |
+|---|---|---|
+| 1 | Postgres 存持久应用状态，与运行时单元、凭证存储分离 | 安全文 §2.1 架构图 + §3.4 |
+| 2 | 凭证与 OAuth token 存于用户 VM 内、不在 Meta 集中基础设施；VM 是用户放入 Muse 的一切内容的系统记录 | 安全文 §3.3 |
+| 3 | 用户可检查、编辑、下载 VM 内所有文件，包括 Muse 关于用户的记忆 | 安全文 §3.3 |
+| 4 | 记忆跨对话持续存在，用户可回指以往讨论过的内容；记忆文件可直接阅读与编辑 | 设计文 + 帮助中心 §2.4 |
+| 5 | 用户可随时让 Muse“忘记”它学到的特定内容 | 新闻室 §2.4 |
+| 6 | VM 数据持续备份以便恢复 | 安全文 §3.3 |
+
+官方**没有**披露：Postgres 表结构、记忆文件内部结构、记忆向量索引细节
+（报告 §4/§5：这些来自逆向与用户观察，不得写成官方架构）。本轨的表结构、
+文件布局是本项目自定的，只对齐上面六条高层要求。
+
+## §2 现状
+
+- 应用状态散落在 `state_dir` 下的多个 JSON 文件（net.json、snapshots.json、
+  pool registry、host-state.json），无结构化存储、无统一审计、无备份恢复。
+- 记忆系统是另一个已闭合的项目（`~/workspace/memory-system`，v13 终态
+  730/730，常驻部署在 172）：Markdown 为可读源头、SQLite bank + FTS5 为派生
+  索引、bge-m3 语义检索（阈值 0.47）、claim 写库、忘记传播、备份恢复已做。
+  对外接口是真实 CLI：`node dist/cli.js <search|get|explain|forget-plan|
+  forget-confirm|backup|restore|export|import|...>`。
+- 凭证已分离（P3 authd）：真 token 只进客内守护进程，agent 只拿替身。
+  但**没有任何机制阻止应用状态里误写凭证**——状态 JSON 可含任意内容。
+- 客机记忆文件：不存在。`~/workspace` 经 chv-sync 同步到 Gitea；记忆若放
+  进 workspace 会进 git 历史，不符合“VM 是系统记录、备份恢复”的口径。
+
+## §3 改动清单（本轨）
+
+1. `cloud/chv/app_state.py`（新）：每 VM 一个 SQLite 状态库
+   `state_dir/<vm>/app-state.sqlite`（0600，WAL），表 tasks / goals /
+   sessions / audit（只追加）/ meta。写操作一律过“无凭证”守卫
+   （键名与值模式双重检查，拒绝时只报键名/模式、不含值）；备份用
+   sqlite3 在线备份 API + manifest（sha256），恢复先验 manifest、
+   先留 pre-restore 副本。
+2. `cloud/chv/memory_files.py`（新）：客机记忆文件接线。
+   - `MemoryFiles`：宿主侧镜像 `state_dir/<vm>/memory-home/`
+     （MEMORY.md + memory/*.md）读/列/写/追加，路径穿越守卫，写操作审计。
+   - `MemorySystemClient`：subprocess 调真实记忆系统 CLI
+     （search / get / explain / forget-plan / forget-confirm /
+     backup / restore / export），不重造检索引擎、不重造忘记传播；
+     CLI 未配置时本地模式（只动 Markdown 镜像，bank 忘记记为待办）。
+   - `pull_memory_home`：经 SSH + tar 拉取客机 `/home/chv/memory`
+     到宿主镜像（命令构造可测，传输走已有 SSH 通道）。
+   - `forget`：plan → confirm → 本地 Markdown 清行 → search 验证
+     查不到 → 审计，全程不含凭证值。
+3. `chv-guest/chv-memory.py`（新，纯标准库）：客内记忆文件操作
+   （list / read / append / forget / export），记忆家目录
+   `/home/chv/memory`（0700，chv 所有），在 workspace 之外——
+   记忆不同步到 Gitea。`build.sh` 拷入脚本并建目录。
+4. `docs/chv-production-readiness.md`：补 P7 一节。
+
+## §4 存储选型结论：SQLite，不是 Postgres
+
+| 维度 | Postgres（Muse 原样） | SQLite（本轨采用） |
+|---|---|---|
+| 部署成本 | 每 VM 一个 server 进程 + 端口 + 认证 + 数据目录；镜像 +150–250MB，常驻 RSS 约 50–100MB | 零 server，单个 0600 文件，WAL，空闲成本 0 |
+| 隔离 | VM 内 Postgres 才符合“VM 是系统记录”；宿主侧共享 Postgres 会把多用户状态集中到宿主，破坏每用户 VM 隔离 | 文件随 VM 走，`state_dir/<vm>/`，与 VM 同生死 |
+| 本仓既有模式 | 无 | 记忆系统本身用 SQLite 作 claim bank；cloud.db 的 AutoMemoryStore 回落 SQLite |
+| 并发 | 多写者 + 行级安全 | 单写者每 VM（进程内锁 + 文件锁守卫）；客机 cell 绝不直读库文件 |
+
+**差距（诚实写明，不冒充 Postgres）**：无多写者服务器、无网络访问、
+无行级安全、无内置复制（复制 = 备份/恢复路径 + Gitea 只管仓库内容）。
+DDL 用 SQLite 与 Postgres 共通的受限子集（TEXT/INTEGER/REAL 列、TEXT id、
+无 AUTOINCREMENT、无 SQLite 专有语法、参数化查询），仓库层只依赖连接
+工厂；将来换 Postgres = 换工厂 + 同一份 DDL + 可序列化隔离说明。
+
+## §5 诚实边界
+
+- 记忆**引擎**不在本轨：检索、bank、FTS5、语义索引、忘记传播、审计
+  全是记忆系统 v13 的已有实现，本轨只调它的 CLI。CLI 不可用时
+  降级为本地 Markdown 模式，bank 忘记明确记为待办，不谎称已忘。
+- 备份恢复演练在本机做（备份→改坏→恢复→校验条数与内容一致）；
+  真机上的定时备份是 P8（crons）那轨的活，本轨只留接口与一次手工演练。
+- “忘记”是尽力删除 + 验证查不到，不是密码学擦除：已备份的快照里
+  仍有旧内容（与记忆系统 CLI 的 backup 语义一致：快照在 home 之外，
+  忘记不 scrub 快照），差距表与用户可见文案里写清。
+- 单元内直读状态库被拒是**构造性保证**：库文件在宿主 `state_dir`，
+  不在客机磁盘镜像里，cell 物理上够不着；测试钉住“库路径不在
+  客机 rootfs 内”与“客机脚本无 state 库引用”。
+
+## §6 真机待验证清单（round-11 之后，测试机空闲后）
+
+1. lite12 镜像含 `chv-memory.py`，`/home/chv/memory` 存在且 0700。
+2. 客内 `chv-memory.py` append → 宿主 `pull_memory_home` → 宿主镜像
+   内容一致；再经 CLI search 能查到。
+3. CLI forget-confirm 后，客机与宿主两处都查不到；审计有 forget 记录。
+4. 状态库备份→删库→恢复→条数与内容一致；备份文件里无 token 值
+   （以测试 token 做扫描）。
+5. 单元内（nspawn cell）`ls /home/chv/memory` 与直读库路径均被拒；
+   经 executor 的记忆读接口可通。
+6. 备份恢复演练在真机跑一遍（含 pre-restore 副本存在）。
+
+## §7 与 P1/P2/P3/P4/P5/P6 的接口
+
+- P3 authd：状态库与记忆文件绝不存凭证值（守卫 + 否定测试）；authd
+  的 vault 不动。
+- P2 cell/broker：cell 不直读状态库；客机记忆脚本跑在 cell 外（与
+  chv-sync 同级），由 executor 经 SSH 调用。
+- P1 常驻池：常驻 VM 的状态库与记忆家目录随 VM 持久化（文件在
+  state_dir/<vm>/，VM pause/resume 不丢）；VM 销毁时状态库与记忆
+  家目录随任务盘清理策略走（删盘即删，不留孤儿——接 orphans 口径）。
+- P4/P5/P6：不碰浏览器、Sentinel、privsep；P6 的安全检查层如需审计
+  记忆读，读本轨的 audit 表（只读）。
diff --git a/docs/chv-production-readiness.md b/docs/chv-production-readiness.md
index bb4e6f2..6b29919 100644
--- a/docs/chv-production-readiness.md
+++ b/docs/chv-production-readiness.md
@@ -833,3 +833,22 @@
   UNSTABLE、未合并、未部署、真机 0 项。差：外部为主。
 - 动态安全 2.0 → 2.0，未达 10：本轨无任何动态验证，
   V-7 未跑。差：外部（KVM 宿主 + 审计结论）。
+
+## 15. 对齐 Muse P7：状态与记忆分离（feat-chv-muse-p7）
+
+- 选型：SQLite 每 VM 一库（`state_dir/<vm>/app-state.sqlite`，0600，
+  WAL），不是 Postgres。理由与差距见 docs/chv-muse-p7-gap.md §4：
+  轻量 VM 跑 Postgres server 太重；宿主侧共享 Postgres 破坏每用户
+  VM 隔离；DDL 用两方共通受限子集，换 Postgres = 换连接工厂。
+- 记忆引擎不重做：调记忆系统 v13 的真实 CLI（search/get/explain/
+  forget-plan/forget-confirm/backup/restore/export）；CLI 缺位时
+  降级本地 Markdown 模式，bank 忘记记为待办。
+- 记忆家目录客内 `/home/chv/memory`（workspace 之外，不同步到
+  Gitea），宿主镜像 `state_dir/<vm>/memory-home/`；客内脚本
+  `chv-memory.py`（纯标准库）经 SSH 调用。
+- 凭证绝不进状态库与记忆文件：写操作过无凭证守卫（键名 + 值模式），
+  备份文件无 token 值（测试钉住）；audit 表只追加。
+- “忘记”是尽力删除 + 验证查不到，不是密码学擦除：已备份快照里
+  仍有旧内容（与记忆系统 CLI 语义一致），差距表 §5 写清。
+- 真机待验证：见 docs/chv-muse-p7-gap.md §6（lite12 镜像、双向
+  拉取、双向忘记、备份恢复演练、cell 内直读被拒）。