ratatoskr/tests/test_provider_affect.py

"""Tests for the Tier-3 Bifrost affect provider (ratatoskr.provider.affect_store).

Contract: docs/contracts/bifrost_affect_provider.contract.md
Vertical tracer-first: basic_emit -> opacity -> lww -> replay -> conflict ->
missing_key -> #195 parity vs InMemoryAffectStore.
"""
from __future__ import annotations

import types

import pytest
from bifrost.affect import AffectIdempotencyConflict, AffectInvalidArguments

from ratatoskr.provider.affect_store import build_affect_provider_app, open_affect_store


def _ctx(sub: str = "sub-1"):
    # Mirrors bifrost's _ctx_actor: actor = jwt_sub (test ctx) or session_id.
    return types.SimpleNamespace(jwt_sub=sub)


def _snapshot(agent: str = "a1", user: str = "u1", **payload):
    base = {
        "agent_id": agent,
        "end_user_id": user,
        "pad": {"p": 0.1, "a": 0.2, "d": 0.3},
        "valence": 0.5,
        "persona_baselines": {"warmth": 0.7},
        "emitted_at": "2026-06-14T00:00:00Z",
    }
    base.update(payload)
    return base


def _row_count(store, table: str) -> int:
    return store._conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]


# --- open_affect_store ---

def test_open_advertises_capability_and_schema():
    store = open_affect_store(":memory:")
    assert store.affect_supported is True
    # both tables queryable
    store._conn.execute("SELECT * FROM affect_snapshots")
    store._conn.execute("SELECT * FROM affect_idempotency")


def test_open_sets_busy_timeout(tmp_path):
    """INV-006: every connection sets busy_timeout>=5000ms (WAL's default is 0, so a
    contended write would fail SQLITE_BUSY immediately) — prep for the two-process
    composite/standalone topology."""
    store = open_affect_store(str(tmp_path / "affect.db"))
    assert store._conn.execute("PRAGMA busy_timeout").fetchone()[0] == 5000


def test_reopen_existing_file_is_idempotent(tmp_path):
    db = str(tmp_path / "affect.db")
    open_affect_store(db)  # first open creates schema
    store = open_affect_store(db)  # reopen: CREATE TABLE IF NOT EXISTS is a no-op
    assert store.affect_supported is True
    store._conn.execute("SELECT * FROM affect_snapshots")
    store._conn.execute("SELECT * FROM affect_idempotency")


# --- emit ---

async def test_basic_emit_stores_and_round_trips():
    store = open_affect_store(":memory:")
    snap = _snapshot()
    result = await store.emit(snap, idempotency_key="k1", ctx=_ctx())
    assert result == {"stored": True}
    assert store.get("a1", "u1") == snap


async def test_opacity_arbitrary_payload_round_trips_and_addressing_invariant():
    store = open_affect_store(":memory:")
    # arbitrary extra/unknown payload fields persist + round-trip verbatim
    snap = _snapshot(weird_field={"nested": [1, 2, 3]}, mystery="x")
    assert await store.emit(snap, idempotency_key="k1", ctx=_ctx()) == {"stored": True}
    assert store.get("a1", "u1") == snap
    # two snapshots for the same key differing ONLY in payload address the SAME row
    snap2 = _snapshot(weird_field={"nested": [9]}, mystery="y", valence=0.99)
    await store.emit(snap2, idempotency_key="k2", ctx=_ctx())
    assert store.get("a1", "u1") == snap2
    assert _row_count(store, "affect_snapshots") == 1  # same row overwritten


async def test_lww_by_arrival_ignores_emitted_at():
    store = open_affect_store(":memory:")
    a = _snapshot(valence=0.1, emitted_at="2026-06-14T10:00:00Z")
    b = _snapshot(valence=0.9, emitted_at="2026-06-14T08:00:00Z")  # OLDER emitted_at
    await store.emit(a, idempotency_key="ka", ctx=_ctx())
    await store.emit(b, idempotency_key="kb", ctx=_ctx())  # distinct key -> arrival wins
    assert store.get("a1", "u1") == b  # later arrival wins despite older emitted_at


async def test_replay_noop_same_key_same_payload():
    store = open_affect_store(":memory:")
    snap = _snapshot()
    assert await store.emit(snap, idempotency_key="k1", ctx=_ctx()) == {"stored": True}
    assert await store.emit(snap, idempotency_key="k1", ctx=_ctx()) == {"stored": True}
    assert store.get("a1", "u1") == snap
    assert _row_count(store, "affect_snapshots") == 1  # replay did not duplicate


async def test_idempotency_conflict_same_key_different_payload():
    store = open_affect_store(":memory:")
    first = _snapshot(valence=0.1)
    await store.emit(first, idempotency_key="k1", ctx=_ctx())
    with pytest.raises(AffectIdempotencyConflict):
        await store.emit(_snapshot(valence=0.2), idempotency_key="k1", ctx=_ctx())
    assert store.get("a1", "u1") == first  # prior snapshot untouched


async def test_same_key_distinct_actor_is_not_a_conflict():
    # idempotency is actor-scoped (INV-006/-008): same key, different ctx actor
    store = open_affect_store(":memory:")
    await store.emit(_snapshot(valence=0.1), idempotency_key="k1", ctx=_ctx("sub-A"))
    # different actor, same key, different payload -> NOT a conflict (distinct id)
    assert await store.emit(
        _snapshot(valence=0.2), idempotency_key="k1", ctx=_ctx("sub-B")
    ) == {"stored": True}


async def test_missing_end_user_id_raises_and_no_write():
    store = open_affect_store(":memory:")
    bad = _snapshot()
    del bad["end_user_id"]
    with pytest.raises(AffectInvalidArguments):
        await store.emit(bad, idempotency_key="k1", ctx=_ctx())
    assert _row_count(store, "affect_snapshots") == 0


async def test_missing_agent_id_raises_and_no_write():
    # PRE-001 guards BOTH addressing keys symmetrically.
    store = open_affect_store(":memory:")
    bad = _snapshot()
    del bad["agent_id"]
    with pytest.raises(AffectInvalidArguments):
        await store.emit(bad, idempotency_key="k1", ctx=_ctx())
    assert _row_count(store, "affect_snapshots") == 0


async def test_empty_idempotency_key_raises():
    store = open_affect_store(":memory:")
    with pytest.raises(AffectInvalidArguments):
        await store.emit(_snapshot(), idempotency_key="", ctx=_ctx())


# --- get ---

def test_get_absent_returns_none():
    store = open_affect_store(":memory:")
    assert store.get("nope", "nope") is None


async def test_get_after_emit_returns_equal():
    store = open_affect_store(":memory:")
    snap = _snapshot()
    await store.emit(snap, idempotency_key="k1", ctx=_ctx())
    assert store.get("a1", "u1") == snap


# --- fetch (affect.fetch wire verb — bifrost >=0.10.0, INV-010 strong-or-absent) ---

async def test_fetch_absent_returns_found_false():
    """fetch_absent: no row for the key → {"found": False} (mirrors reference)."""
    store = open_affect_store(":memory:")
    assert await store.fetch("nope", "nope") == {"found": False}


async def test_fetch_after_emit_returns_snapshot():
    """fetch_after_emit [tracer]: emit then fetch → {"found": True, "snapshot": <verbatim>}."""
    store = open_affect_store(":memory:")
    snap = _snapshot()
    await store.emit(snap, idempotency_key="k1", ctx=_ctx())
    assert await store.fetch("a1", "u1") == {"found": True, "snapshot": snap}


async def test_fetch_missing_key_raises():
    """fetch_missing_key: empty/missing addressing key → AffectInvalidArguments
    (PRE-001; symmetric across both keys, belt-and-suspenders behind the wire)."""
    store = open_affect_store(":memory:")
    with pytest.raises(AffectInvalidArguments):
        await store.fetch("", "u1")
    with pytest.raises(AffectInvalidArguments):
        await store.fetch("a1", "")


# --- build_affect_provider_app ---

def test_build_app_exposes_handshake_and_affect_routes():
    store = open_affect_store(":memory:")
    app = build_affect_provider_app(store, heimdall_key=b"secret-key")
    routes = {getattr(r, "path", None): r for r in app.routes}
    assert "/bifrost/handshake" in routes
    assert "/bifrost/affect-call" in routes
    assert "POST" in routes["/bifrost/affect-call"].methods  # POST-001: the verb, not just the path
    # POST-002: bifrost routes remain REACHABLE (not merely registered) after the read
    # route is composed in via add_route — drive one without a JWT → routed (auth-
    # rejected), never 404.
    r = TestClient(app).post("/bifrost/affect-call", json={"operation": "affect.emit"})
    assert r.status_code != 404


def test_build_app_rejects_non_advertising_store():
    store = open_affect_store(":memory:")
    store.affect_supported = False
    with pytest.raises(ValueError):
        build_affect_provider_app(store, heimdall_key=b"k")


def test_build_app_rejects_empty_key():
    store = open_affect_store(":memory:")
    with pytest.raises(ValueError):
        build_affect_provider_app(store, heimdall_key=b"")


# --- #195 conformance: parity vs the reference store through the real engine ---

def _dispatch_ctx(*scopes: str, session_id: str = "actor-1"):
    return types.SimpleNamespace(
        scope=list(scopes), session_id=session_id, jwt_sub=session_id
    )


def _env(snap: dict, key: str = "sess-1:1:affect") -> dict:
    return {
        "operation": "affect.emit",
        "args": snap,
        "idempotency_key": key,
        "idempotency_class": "short-retry",
    }


def _ref_shaped_snapshot(*, pleasure: float = 0.5, emitted_at: str = "2026-06-14T12:00:00Z"):
    # Mirror the reference test's snapshot shape so the envelope validates.
    return {
        "agent_id": "agent-1",
        "end_user_id": "user-1",
        "pad": {"pleasure": pleasure, "arousal": 0.2, "dominance": -0.1},
        "valence": [{"entity_id": "e1", "regard": 0.7, "familiarity": 0.3}],
        "emitted_at": emitted_at,
    }


async def test_parity_vs_reference_store_through_dispatch():
    from bifrost.affect import dispatch_affect_call
    from bifrost.consumer.testing import InMemoryAffectStore

    ref = InMemoryAffectStore()
    mine = open_affect_store(":memory:")
    ctx = _dispatch_ctx("affect:write")
    snap = _ref_shaped_snapshot()

    # happy persist: wire bodies must agree
    assert await dispatch_affect_call(_env(snap), ctx, ref) == await dispatch_affect_call(
        _env(snap), ctx, mine
    )
    # replay (same key + same payload): both no-op {stored: true}
    assert await dispatch_affect_call(_env(snap), ctx, ref) == await dispatch_affect_call(
        _env(snap), ctx, mine
    )
    # conflict (same key + different payload): both map to the same error envelope
    other = _ref_shaped_snapshot(pleasure=0.99)
    assert await dispatch_affect_call(_env(other), ctx, ref) == await dispatch_affect_call(
        _env(other), ctx, mine
    )


def _fetch_env(agent_id: str = "agent-1", end_user_id: str = "user-1") -> dict:
    return {"operation": "affect.fetch", "args": {"agent_id": agent_id, "end_user_id": end_user_id}}


async def test_parity_vs_reference_fetch_through_dispatch():
    """#195 parity for affect.fetch: cold (not-found) + warm (found) read envelopes
    yield identical (status, body) through the real engine against the reference store
    and ours. Conforms to bifrost's InMemoryAffectStore.fetch ({found, snapshot})."""
    from bifrost.affect import dispatch_affect_call
    from bifrost.consumer.testing import InMemoryAffectStore

    ref = InMemoryAffectStore()
    mine = open_affect_store(":memory:")
    write_ctx = _dispatch_ctx("affect:write")
    read_ctx = _dispatch_ctx("affect:read")

    # cold fetch (nothing persisted): both -> {found: false}
    assert await dispatch_affect_call(_fetch_env(), read_ctx, ref) == await dispatch_affect_call(
        _fetch_env(), read_ctx, mine
    )

    # seed both via emit, then fetch -> both {found: true, snapshot: <verbatim>}
    snap = _ref_shaped_snapshot()
    await dispatch_affect_call(_env(snap), write_ctx, ref)
    await dispatch_affect_call(_env(snap), write_ctx, mine)
    assert await dispatch_affect_call(_fetch_env(), read_ctx, ref) == await dispatch_affect_call(
        _fetch_env(), read_ctx, mine
    )


# --- PAD read route (issue #18 Deliverable 2) ---
# Non-bifrost GET /affect/state/{agent_id}?end_user_id=… → store.get snapshot.

import json as _json

from starlette.testclient import TestClient


def _affect_snapshot(agent: str = "ratatoskr:sindra", user: str = "vuong") -> dict:
    # The real affect.emit shape (verified live): pad + per-entity valence + emitted_at.
    return {
        "agent_id": agent,
        "end_user_id": user,
        "pad": {"pleasure": 0.1459, "arousal": 0.0796, "dominance": -0.0071},
        "valence": [
            {
                "entity_id": "ratatoskr",
                "entity_type": "human",
                "familiarity": 0.5886,
                "interaction_count": 8,
                "regard": 0.15,
            }
        ],
        "emitted_at": "2026-06-18T15:58:12+00:00",
    }


def _seed(store, snap: dict) -> None:
    blob = _json.dumps(snap, sort_keys=True, separators=(",", ":"))
    store._conn.execute(
        "INSERT INTO affect_snapshots (agent_id, end_user_id, snapshot_json, arrived_at) "
        "VALUES (?, ?, ?, ?)",
        (snap["agent_id"], snap["end_user_id"], blob, "0"),
    )
    store._conn.commit()


def test_affect_state_route_returns_seeded_snapshot():
    """tracer: seeded (agent, user) → 200 with the snapshot verbatim. Colon-id in the
    path exercises INV-008 at the provider hop."""
    store = open_affect_store(":memory:")
    snap = _affect_snapshot()
    _seed(store, snap)
    client = TestClient(build_affect_provider_app(store, heimdall_key=b"k"))
    r = client.get("/affect/state/ratatoskr:sindra", params={"end_user_id": "vuong"})
    assert r.status_code == 200
    assert r.json() == snap


def test_affect_state_route_absent_returns_404_no_snapshot():
    """INV-003: no emit yet for (agent, user) → explicit 404 no_affect_snapshot,
    NEVER a zeroed pad that reads as real data."""
    store = open_affect_store(":memory:")
    client = TestClient(build_affect_provider_app(store, heimdall_key=b"k"))
    r = client.get("/affect/state/ratatoskr:ghost", params={"end_user_id": "nobody"})
    assert r.status_code == 404
    body = r.json()
    assert body["error_code"] == "no_affect_snapshot"
    assert "pad" not in body  # no fabricated PAD


def test_affect_state_route_missing_end_user_id_returns_400():
    """PRE-001: absent end_user_id query → 400 missing_end_user_id (not a silent
    no-snapshot lookup against a None partition)."""
    store = open_affect_store(":memory:")
    _seed(store, _affect_snapshot())
    client = TestClient(build_affect_provider_app(store, heimdall_key=b"k"))
    r = client.get("/affect/state/ratatoskr:sindra")  # no end_user_id
    assert r.status_code == 400
    assert r.json()["error_code"] == "missing_end_user_id"