CyberGril/tests/test_semantic.py

"""R4 semantic wiring: pre-grouped extraction, fail-closed provenance gate, restarts.

All deterministic with the stub backend — zero model calls, zero embeddings.
The gate negative cases mirror the audited HEL-859–867 rejection forms ported
onto the product's own tables (main archive + PublicView share state).
"""
import json

import pytest

from cybergril.memory.semantic import (
    SemanticMemory,
    StubBackend,
    anchor_fingerprint,
    build_semantic,
)
from cybergril.memory.store import Store


@pytest.fixture
def store(tmp_path):
    s = Store(tmp_path / "sem.db")
    yield s
    s.close()


def add_turn(store, turn, user_text, companion_text=None, *, ts=1700000000000):
    with store.conn:
        store.execute("INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES(?,'user',?,?,'sent')",
                      (ts, user_text, turn))
        if companion_text is not None:
            store.execute(
                "INSERT INTO chat_message(ts,role,text,turn_id,status) VALUES(?,'companion',?,?,'sent')",
                (ts + 1, companion_text, turn))


def make_public_companion(store, turn):
    """Mark this turn's companion row public: judged+acked attempt, all-told candidates."""
    row = store.query_one("SELECT id FROM chat_message WHERE turn_id=? AND role='companion'", (turn,))
    event = store.execute(
        "INSERT INTO life_event(ts,kind,content,place,valence,salience,source,told,day_key) "
        "VALUES(?,?,?,?,?,?,'generated',1,?)",
        (1700000000000, "thought", "已讲的生活", "卧室", 0.2, 0.6, "2026-10-26"),
    ).lastrowid
    with store.conn:
        store.execute(
            "INSERT INTO sharing_attempt(message_id,turn_id,epoch,token,text_hash,expires_at,candidates,"
            "state,selected,ack_at) VALUES(?,?,1,'t','h',9223372036854775806,?,'judged',?,1700000001000)",
            (row["id"], turn, json.dumps([event]), json.dumps([event])),
        )
    return row["id"]


@pytest.fixture
def sem(store):
    return SemanticMemory(store, StubBackend())


async def test_only_public_rows_enter_batches(store, sem):
    add_turn(store, "t1", "我说了蓝色", "我听到了蓝色")  # companion row not shared yet
    sem.enqueue_uncovered()
    job = store.query_one("SELECT * FROM sem_batch")
    # user rows are the user's own speech and always public; the unshared
    # companion row stays out of every outbound batch.
    anchor = json.loads(job["anchor_json"])
    assert [r[1] for r in anchor["turn_roles"]] == ["user"]
    rows = sem.public_rows_for_turn("t1")
    assert {r["role"] for r in rows} == {"user"}
    assert anchor["input_sha256"] == anchor_fingerprint(rows)
    # once shared, the companion row gets its own deterministic batch
    make_public_companion(store, "t1")
    sem.enqueue_uncovered()
    jobs = store.query("SELECT * FROM sem_batch ORDER BY id")
    assert len(jobs) == 2
    assert {r[1] for r in json.loads(jobs[1]["anchor_json"])["turn_roles"]} == {"companion"}
    # one archive row enters at most one batch, ever
    sem.enqueue_uncovered()
    assert store.query_one("SELECT COUNT(*) n FROM sem_batch")["n"] == 2


async def test_gate_rejects_mismatched_fingerprint_and_missing_rows(store, sem):
    add_turn(store, "t2", "钥匙放在抽屉里", "好，我记住抽屉了")
    make_public_companion(store, "t2")
    sem.enqueue_uncovered()
    await sem.tick()
    assert store.query_one("SELECT state FROM sem_batch")["state"] == "done"
    hits = await sem.backend.search("钥匙")
    assert hits
    # archive row rewritten after the fact -> fingerprint no longer matches
    with store.conn:
        store.execute("UPDATE chat_message SET text='被改写过的原话' WHERE turn_id='t2' AND role='user'")
    chain = sem.chain_resolve(hits[0])
    assert not chain["resolved"] and chain["reject_reason"] == "source_mismatch"
    block = await sem.context("钥匙放在哪里", [], 600)
    assert block == ""


async def test_gate_rejects_fabricated_and_foreign_hits(store, sem):
    add_turn(store, "t3", "我最爱海雾的颜色", "海边的感觉很好")
    make_public_companion(store, "t3")
    sem.enqueue_uncovered()
    await sem.tick()
    good = (await sem.backend.search("海雾"))[0]
    assert sem.chain_resolve(good)["resolved"]
    # a hit whose payload anchor is missing entirely
    assert sem.chain_resolve({"id": "stub-999", "memory": "任意文本"})["reject_reason"] == "no_payload"
    # a real entry but text swapped in flight (hit_text_mismatch form)
    forged = dict(good)
    forged["memory"] = good["memory"] + "捏造的尾巴"
    chain = sem.chain_resolve(forged)
    assert chain["reject_reason"] == "hit_text_mismatch"


async def test_gate_rejects_unshared_companion_row(store, sem):
    add_turn(store, "t4", "今天面试结束了", "辛苦了，等你好消息")
    make_public_companion(store, "t4")
    sem.enqueue_uncovered()
    await sem.tick()
    # per-row anchors (HEL-885): the companion's own entry anchors only her row
    hits = await sem.backend.search("辛苦")
    assert sem.chain_resolve(hits[0])["resolved"]
    # share state revoked at read time (attempt no longer acked) -> fail closed
    with store.conn:
        store.execute("UPDATE sharing_attempt SET ack_at=NULL WHERE turn_id='t4'")
    chain = sem.chain_resolve(hits[0])
    assert chain["reject_reason"] == "companion_unshared"
    block = await sem.context("辛苦了等什么", [], 600)
    assert "辛苦" not in block
    # her row's rejection never touches the user row's own entry
    user_hit = (await sem.backend.search("面试"))[0]
    assert sem.chain_resolve(user_hit)["resolved"]


async def test_context_respects_budget_and_excludes_turns(store, sem):
    add_turn(store, "t5", "窗台的多肉冒了新芽", "新芽很可爱")
    make_public_companion(store, "t5")
    sem.enqueue_uncovered()
    await sem.tick()
    block = await sem.context("多肉", [], 600)
    assert block.startswith("她确实记得的更早的事")
    assert "多肉" in block
    # the current turn's own rows are never recalled as "earlier" memory
    assert await sem.context("多肉", ["t5"], 600) == ""
    assert await sem.context("多肉", [], 10) == ""


async def test_pending_batch_restart_recovery_never_resends(store, sem):
    add_turn(store, "t6", "备忘：周五看牙医", "记住了")
    make_public_companion(store, "t6")
    sem.enqueue_uncovered()
    with store.conn:
        store.execute("UPDATE sem_batch SET state='pending',lease_until=1,owner='dead'")
    sem.recover()
    job = store.query_one("SELECT state,reason FROM sem_batch")
    assert job["state"] == "failed" and job["reason"] == "lease_expired"
    # failed batches retry within attempts bound; coverage rows prevent duplicates
    await sem.tick()
    assert store.query_one("SELECT state FROM sem_batch")["state"] == "done"
    assert store.query_one("SELECT COUNT(*) n FROM sem_covered")["n"] == 2


def test_build_semantic_modes(store):
    assert build_semantic(store, mode="off") is None
    stub = build_semantic(store, mode="stub")
    assert stub.backend.backend == "stub"
    # mem0 without an approved embed cap fails closed to disabled
    assert build_semantic(store, mode="mem0", config={}) is None