CyberGril/scripts/verify_hel876_voice_seam.py

"""HEL-876 offline protocol check for the R5 voice-worker seam (zero model / zero embed / zero TTS / zero ASR).

Proves, against the unified-candidate API with a fake LLM provider, that a voice
worker reusing POST /api/chat gets the exact contract the text adapter already
uses: epoch resync (409), serialized turns, mid-stream cancel (client disconnect
-> row marked cancelled, next turn clean), idempotent replay of a completed turn
(no second model call), and the receive-confirmation receipt round trip.
Run: PYTHONPATH=src .venv311/bin/python scripts/verify_hel876_voice_seam.py
"""
from __future__ import annotations

import json
import sys
import tempfile
import threading
import time
import uuid
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tests"))

from fastapi.testclient import TestClient  # noqa: E402
from test_r2 import R2LLM  # noqa: E402

from cybergril.api.app import create_app  # noqa: E402
from cybergril.life.runtime import LifeConfig  # noqa: E402
from cybergril.think.router import build_router  # noqa: E402
from cybergril.think.scenes import Tier  # noqa: E402

CREATE = {"name": "协议检查", "birth": "2000-10-18", "hour": "亥时", "relation": "朋友"}
T1, T2, T3, T4 = (str(uuid.uuid4()) for _ in range(4))


def sse_frames(response) -> list[dict]:
    frames = []
    for line in response.iter_lines():
        if line.startswith("data: "):
            frames.append(json.loads(line[6:]))
    return frames


def main() -> int:
    provider = R2LLM()
    router = build_router(providers={t: provider for t in (Tier.HIGH, Tier.STANDARD, Tier.LOW)})
    tmp = Path(tempfile.mkdtemp(prefix="hel876-seam-"))
    app = create_app(db_path=tmp / "seam.db", router=router,
                     life_config=LifeConfig(enabled=True, poll_seconds=3600),
                     semantic_mode="stub")
    ok = []

    with TestClient(app) as client:
        # 1. epoch discovery (worker startup: GET /api/health)
        health = client.get("/api/health").json()
        epoch = health["epoch"]
        assert health["capabilities"]["voice"] is False, "voice must stay honestly false"
        ok.append(f"health epoch={epoch}, voice=false")

        # 2. companion exists (worker does NOT create a second her)
        client.post("/api/companion", json=CREATE)
        assert client.get("/api/companion").status_code == 200

        # 3. full turn: deltas then final with receipt
        r = client.post("/api/chat", json={"text": "你好呀", "turn_id": T1, "epoch": epoch})
        frames = sse_frames(r)
        deltas = [f["delta"] for f in frames if f["t"] == "chat.delta"]
        final = [f for f in frames if f["t"] == "chat.final"]
        assert deltas and "".join(deltas) == final[0]["utterance"]["text"]
        assert final[0]["receipt"], "final must carry the receive-confirmation receipt"
        receipt = final[0]["receipt"]
        ok.append(f"turn v-1 streamed {len(deltas)} deltas + final(receipt keys={sorted(receipt)})")
        calls_after_turn = len(provider.calls)

        # 4. replay of the completed turn: no second model call
        r2 = client.post("/api/chat", json={"text": "你好呀", "turn_id": T1, "epoch": epoch})
        frames2 = sse_frames(r2)
        assert [f["t"] for f in frames2] == ["chat.final"], frames2
        assert len(provider.calls) == calls_after_turn, "replay must not call the model"
        ok.append("replay of v-1 returns saved final only, model calls unchanged")

        # 5. receive confirmation (worker acks the receipt after TTS playback)
        ack = client.post("/api/chat/received", json={k: receipt[k] for k in
                                                       ("message_id", "turn_id", "epoch", "token", "text_hash")})
        assert ack.status_code == 200 and ack.json() == {"accepted": True}
        ok.append("receipt acknowledged via /api/chat/received")

        # 6. mid-stream interruption: a new turn (barge-in) cancels the active one
        provider.mode = "slow"  # reply hangs after first delta
        def slow_turn() -> None:
            with client.stream("POST", "/api/chat",
                               json={"text": "说一半我打断你", "turn_id": T2, "epoch": epoch}) as s:
                for _ in s.iter_lines():
                    pass

        slow = threading.Thread(target=slow_turn, daemon=True)
        slow.start()
        time.sleep(1.0)  # first delta arrives, turn v-2 is active and hanging
        r3 = client.post("/api/chat", json={"text": "还在吗", "turn_id": T3, "epoch": epoch})
        assert [f["t"] for f in sse_frames(r3)][-1] == "chat.final"
        slow.join(timeout=10)
        provider.mode = ""
        messages = client.get("/api/messages").json()["messages"]
        v2 = [m for m in messages if m["turn_id"] == T2]
        v3 = [m for m in messages if m["turn_id"] == T3]
        assert v2 and all(m["status"] == "cancelled" for m in v2), v2
        assert not [m for m in v2 if m["role"] == "companion"], "no phantom companion row for aborted turn"
        assert any(m["status"] == "sent" and m["role"] == "companion" for m in v3)
        ok.append("barge-in new turn cancels active v-2 (row=cancelled, no companion row); v-3 completes")

        # 8. epoch mismatch -> 409 with fresh epoch (worker resyncs like real.ts)
        stale = client.post("/api/chat", json={"text": "旧纪元", "turn_id": T4, "epoch": epoch + 999})
        assert stale.status_code == 409 and "epoch" in stale.json()
        ok.append("stale epoch rejected with 409 + current epoch for resync")

    print("HEL-876 voice-seam offline protocol check: ALL PASS")
    for line in ok:
        print("  ✓", line)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())