codex-app-server/chv-guest/chv-memory.py

#!/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())