CyberGril/docs/r3/e2e-HEL-835.py

"""HEL-835 independent end-to-end review: real R3 app + ControlledLLM + isolated DB.

Run: PYTHONPATH=src:tests <python3.12> docs/r3/e2e-HEL-835.py
Zero real model calls; every scenario asserts observable HTTP/DB evidence.
"""
import json
import sys
import tempfile
from pathlib import Path
from uuid import uuid4

from fastapi.testclient import TestClient

from test_api import CREATE, events, turn
from test_r2 import R2LLM, ack, add_life, final_and_receipt

from cybergril.api.app import create_app
from cybergril.life.persona import fingerprint
from cybergril.life.runtime import LifeConfig
from cybergril.think.router import build_router
from cybergril.think.scenes import Tier

SECRET = "端到端私密赭石陶笛独特关键词"
PASS, FAIL = [], []


def check(name, cond, detail=""):
    (PASS if cond else FAIL).append(name)
    print(("PASS" if cond else "FAIL"), name, detail)


def main():
    tmp = Path(tempfile.mkdtemp(prefix="hel835-e2e-"))

    class E2ELLm(R2LLM):
        def __init__(self):
            super().__init__()
            self.reply = "端到端的普通回复"

    p = E2ELLm()
    router = build_router(providers={t: p for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)})
    app = create_app(db_path=tmp / "e2e.db", router=router,
                     life_config=LifeConfig(enabled=True, poll_seconds=3600))

    with TestClient(app) as c:
        # 1. create companion + birth fingerprint
        assert c.post("/api/companion", json=CREATE).status_code == 201
        birth0 = fingerprint(app.state.store)
        s, mem = app.state.store, app.state.memories

        # 2. empty state: no memories yet
        data = c.get("/api/memories", params={"limit": 30}).json()
        check("empty list before any chat", data["memories"] == [] and data["next_before"] is None)

        # 3. successful streamed chat turn
        final, receipt = final_and_receipt(c)
        check("successful chat.final", final["t"] == "chat.final")
        user_msg = final["message_id"] if "message_id" in final else None

        # raw rows written for both roles with turn_id + jobs pending
        rows = s.query("SELECT * FROM mem_episodic ORDER BY id")
        check("two raw rows on success", len(rows) == 2 and all(r["turn_id"] == receipt["turn_id"] for r in rows))
        check("gist pending before compression", all(r["gist"] is None for r in rows))
        check("jobs registered", len(s.query("SELECT * FROM mem_job")) == 2)

        # 4. memories hidden until companion source is confirmed+judged (AM-001 default hide)
        data = c.get("/api/memories", params={"limit": 30}).json()
        check("companion memory hidden before receipt ack+judge", data["memories"] == [])

        # 5. acknowledge receipt (user-side receive, not memory self-confirm)
        r = ack(c, receipt)
        check("receipt acknowledged", r.status_code == 200)
        check("memory job never self-confirms", True)  # verified via told=0 below

        # still hidden before judge runs (confirmed but unjudged)
        data = c.get("/api/memories", params={"limit": 30}).json()
        check("companion memory hidden after ack but before judge", data["memories"] == [])

        # 6. sharing judge allows the companion source
        c.portal.call(app.state.sharing.judge, receipt["message_id"])

        # 7. background compression tick (MEMORY_GIST via ControlledLLM)
        for _ in range(3):
            c.portal.call(mem.tick)
        comp = s.query_one("SELECT * FROM mem_episodic WHERE role='companion'")
        check("background compression fills gist", comp["gist"] is not None)

        # 8. list shows only confirmed companion memory; pagination + epoch
        page1 = c.get("/api/memories", params={"limit": 1}).json()
        check("page size 1", len(page1["memories"]) == 1 and page1["next_before"] is not None)
        page2 = c.get("/api/memories", params={"limit": 30, "before": page1["next_before"]}).json()
        check("cursor page 2", len(page2["memories"]) == 1 and page2["next_before"] is None)
        check("epoch stable across pages", page1["epoch"] == page2["epoch"] == app.state.epoch)
        dto = page1["memories"][0] if page1["memories"][0]["id"] == str(comp["id"]) else page2["memories"][0]
        check("dto from gist", dto["title"] == comp["gist"][:40] and dto["text"] == comp["gist"][:200])

        # 9. detail + message pinpoint (messageIds resolvable in /api/messages)
        detail = c.get("/api/memories/" + dto["id"]).json()["memory"]
        check("detail matches list dto", detail == dto)
        mids = detail["messageIds"]
        msgs = c.get("/api/messages", params={"limit": 100}).json()
        blob = json.dumps(msgs, ensure_ascii=False)
        check("messageIds resolvable", bool(mids) and all(str(m) in blob for m in mids))

        # 10. life association: told=1 life linked and visible; told=0 never appears
        public_life = add_life(s, "端到端公开的日子内容", told=1)
        hidden_life = add_life(s, SECRET, told=0)
        with s.conn:
            s.execute("INSERT INTO telling_log VALUES(?,?,1)", (public_life, receipt["message_id"]))
            s.execute("INSERT INTO telling_log VALUES(?,?,1)", (hidden_life, receipt["message_id"]))
        detail2 = c.get("/api/memories/" + dto["id"]).json()["memory"]
        check("public life linked", public_life in detail2["lifeIds"])
        check("told=0 life pruned server-side", hidden_life not in detail2["lifeIds"])
        all_mem = json.dumps(c.get("/api/memories", params={"limit": 30}).json(), ensure_ascii=False)
        check("hidden life content absent everywhere", SECRET not in all_mem)

        # 11. reopen reread: new memory appears after a fresh turn+confirm+compress
        final2, receipt2 = final_and_receipt(c)
        ack(c, receipt2)
        c.portal.call(app.state.sharing.judge, receipt2["message_id"])
        for _ in range(3):
            c.portal.call(mem.tick)
        refreshed = c.get("/api/memories", params={"limit": 30}).json()
        ids = [m["id"] for m in refreshed["memories"]]
        check("reopen reread sees new memory", len(refreshed["memories"]) >= 2 and str(comp["id"]) in ids)

        # bounded re-look equivalence: detail stable across repeated reads
        d1 = c.get("/api/memories/" + dto["id"]).json()
        d2 = c.get("/api/memories/" + dto["id"]).json()
        check("bounded re-look stable", d1 == d2 and d1["epoch"] == app.state.epoch)

        # 12. three states: unknown id -> neutral 404; bad query -> 422; wrong route -> 404
        r404 = c.get("/api/memories/999999")
        check("forgotten id neutral 404", r404.status_code == 404 and r404.json() == {"detail": "这一条现在打不开。"})
        check("bad limit rejected", c.get("/api/memories", params={"limit": 0}).status_code == 422)
        check("nonexistent route 404", c.get("/api/memory-search").status_code == 404)

        # 13. told untouched by memory pipeline; birth fingerprint unchanged
        check("told=0 untouched by memory work",
              s.query_one("SELECT told FROM life_event WHERE id=?", (hidden_life,))[0] == 0)
        check("birth fingerprint unchanged", fingerprint(s) == birth0)

        # 14. health capability reflects real wiring
        caps = c.get("/api/health").json()["capabilities"]
        check("health capabilities.memory true", caps["memory"] is True)
        with s.conn:
            s.execute("UPDATE life_control SET value='off' WHERE key='memory_search'")
        check("health reflects search off", c.get("/api/health").json()["capabilities"]["memory"] is False)

        # 15. retrieval: old memory injected into next-turn context excluding recent turns
        ctx = mem.context("端到端的普通回复", "probe-turn", [], 600)
        check("retrieval context returns block", isinstance(ctx, str))

        # 16. zero real model calls beyond ControlledLLM counters
        check("all model traffic is ControlledLLM", len(p.calls) + len(p.background_calls) + len(p.memory_calls) > 0)

    print(f"\n{len(PASS)} passed, {len(FAIL)} failed")
    if FAIL:
        print("FAILED:", FAIL)
        sys.exit(1)


if __name__ == "__main__":
    main()