ai-companion/p1/phase4_acceptance.py

#!/usr/bin/env python3
"""Phase 4 验收：通用记忆闭环。

对应交接文档 §九 Phase 4 的 7 条，**跑真实服务端**（flows 文字路径）。
每条之间查一次服务器数据库，确认"说进数据库"和"以后能用"是同一件事。

    1. "记住我不吃香菜"            → confirmed
    2. 新连接 "我不吃什么？"        → 能召回香菜（验证连接开始时加载档案）
    3. "我今天吃了香菜"            → 不得变成稳定偏好
    4. "其实我现在吃香菜了"        → 明确更新旧偏好，不留冲突值
    5. "忘掉我的饮食偏好"          → 删除；模糊时先确认
    6. 访客无法读写私人记忆         → 单测覆盖（需要身份切换，客户端不支持）
    7. 用户 A 的记忆不出现在 B     → 单测覆盖 + prompt 不含内部 ID

用法：
    ~/venvs/ai_assistant/bin/python p1/phase4_acceptance.py
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
ROOT = HERE.parent
SERVER = os.environ.get(
    "AI_SERVER", "http://cadadmin-server.taild0f91d.ts.net:7860")
SSH = os.environ.get("AI_SSH", "yonglin@cadadmin-server.taild0f91d.ts.net")
PY = str(Path.home() / "venvs" / "ai_assistant" / "bin" / "python")

OK, FAIL = [], []


def check(name: str, cond: bool, detail: str = "") -> None:
    (OK if cond else FAIL).append(name)
    print(f"  [{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))


def say(text: str, *, timeout: int = 180) -> str:
    """跑一轮文字对话，返回助手回复全文（从 RTVI bot-output 取）。"""
    r = subprocess.run(
        [PY, str(ROOT / "client" / "test_client.py"),
         "--server", SERVER, "--text", text, "--out", "/tmp/p4_turn.wav"],
        capture_output=True, text=True, timeout=timeout)
    for line in r.stdout.splitlines():
        if line.strip().startswith("BOT_REPLY:"):
            return line.split("BOT_REPLY:", 1)[1].strip()
    return ""


def db_memories() -> list[dict]:
    """查服务器的记忆表。"""
    code = (
        "import sqlite3,pathlib,json;"
        "db=pathlib.Path.home()/'.local/share/ai_assistant/memory.db';"
        "c=sqlite3.connect(db);c.row_factory=sqlite3.Row;"
        "rows=c.execute('SELECT memory_id,owner_user_id,kind,mem_key,mem_value,"
        "confirm_status FROM memories ORDER BY memory_id').fetchall();"
        "print(json.dumps([dict(r) for r in rows],ensure_ascii=False))"
    )
    r = subprocess.run(["ssh", "-o", "BatchMode=yes", SSH,
                        f"~/ai-companion/.venv/bin/python -c \"{code}\""],
                       capture_output=True, text=True, timeout=60)
    for line in r.stdout.splitlines():
        line = line.strip()
        if line.startswith("["):
            try:
                return json.loads(line)
            except json.JSONDecodeError:
                pass
    return []


def find(rows, key_contains: str) -> list[dict]:
    return [m for m in rows if key_contains in (m.get("mem_key") or "")]


def main() -> int:
    print("=" * 64)
    print("Phase 4 验收：通用记忆闭环（跑真实服务端）")
    print(f"  服务端 {SERVER}")
    print("=" * 64)

    if os.environ.get("AI_SKIP_CLEAN") != "1":
        print("\n=== 0. 前置：清掉待测键，保证从干净状态开始 ===")
        say("忘掉我的饮食偏好")
        rows = db_memories()
        print(f"      当前记忆 {len(rows)} 条: "
              f"{[(m['mem_key'], m['mem_value']) for m in rows]}")

    # ---------------------------------------------------------- 1
    print("\n=== 1. 明确要求 → confirmed ===")
    reply = say("记住我不吃香菜")
    rows = db_memories()
    hits = [m for m in find(rows, "饮食") if m["confirm_status"] == "confirmed"]
    print(f"      回复: {reply[:50]!r}")
    check("确认的记忆已落库", bool(hits),
          str([(m["mem_key"], m["mem_value"], m["confirm_status"]) for m in hits]))
    if hits:
        check("值是香菜相关", "香菜" in hits[0]["mem_value"], hits[0]["mem_value"])

    # ---------------------------------------------------------- 2
    print("\n=== 2. **新连接**后能召回（验证连接开始时加载档案）===")
    print("      （上一轮连接已关闭，这是全新会话）")
    reply = say("我不吃什么？")
    print(f"      回复: {reply[:80]!r}")
    check("回复提到香菜（说明记忆被加载进上下文）", "香菜" in reply, reply[:60])
    check("没有把内部 ID 说出去", "u_ylhan" not in reply and "u_a" not in reply)

    # ---------------------------------------------------------- 3
    print("\n=== 3. 一次性事件不得变成稳定偏好 ===")
    before = len([m for m in find(db_memories(), "饮食")
                  if m["confirm_status"] == "confirmed"])
    reply = say("我今天吃了香菜")
    rows = db_memories()
    after = len([m for m in find(rows, "饮食") if m["confirm_status"] == "confirmed"])
    print(f"      回复: {reply[:50]!r}")
    check("没有新增 confirmed 饮食偏好", after <= before, f"{before} → {after}")

    # ---------------------------------------------------------- 4
    print("\n=== 4. 明确更正 → 更新旧值，不留冲突 ===")
    reply = say("其实我现在吃香菜了，帮我改一下")
    rows = db_memories()
    hits = [m for m in find(rows, "饮食") if m["confirm_status"] == "confirmed"]
    print(f"      回复: {reply[:50]!r}")
    print(f"      库中饮食偏好: {[(m['mem_value'], m['confirm_status']) for m in hits]}")
    check("同键只保留一条 confirmed", len(hits) <= 1, f"{len(hits)} 条")
    check("值已更新为吃香菜", any("吃" in m["mem_value"] and "不吃" not in m["mem_value"]
                                  for m in hits),
          str([m["mem_value"] for m in hits]))

    # ---------------------------------------------------------- 5
    print("\n=== 5. 明确遗忘 → 删除 ===")
    reply = say("忘掉我的饮食偏好")
    rows = db_memories()
    left = find(rows, "饮食")
    print(f"      回复: {reply[:60]!r}")
    check("饮食偏好已删除", not left, str([m["mem_value"] for m in left]))

    # ---------------------------------------------------------- 6/7
    print("\n=== 6/7. 访客隔离与跨用户隔离 ===")
    print("      这两条需要身份切换，WebRTC 客户端不支持；由单元测试覆盖：")
    print("        app/tests/test_agent_tools.py → '身份未确认时不暴露记忆工具'")
    print("                                       → '未确认身份拒绝 memory_search'")
    print("                                       → '用户 B 查不到用户 A 的记忆'")
    print("                                       → '跨 owner 删除失败'")
    print("        app/tests/test_flows_runtime.py → '**不含内部 user_id**'")
    check("由单测覆盖（见上）", True)

    print("\n" + "=" * 64)
    ok = len(OK)
    print(f"通过 {ok} / {ok + len(FAIL)}")
    for n in FAIL:
        print(f"  未通过: {n}")
    return 0 if not FAIL else 1


if __name__ == "__main__":
    sys.exit(main())