"""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_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 # --- 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 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 )