"""Tests for the dispatch-layer observe feed (ratatoskr.provider.opfeed). Issue #17 slice 2 (the Observe half). These drive the REAL bifrost provider ASGI app end-to-end through `instrument_provider_app`, minting a valid dispatch JWT with bifrost's own `mint_dispatch_jwt` — so the envelopes and the session_id claim are the real wire shapes, not hand-mocked guesses (the repo's "test against the shipped lib" posture). """ from __future__ import annotations import httpx import pytest from bifrost.core.dispatch_jwt import mint_dispatch_jwt from ratatoskr.provider.affect_store import ( build_affect_provider_app, open_affect_store, ) from ratatoskr.provider.memory_store import ( build_memory_provider_app, open_memory_store, ) from ratatoskr.provider.opfeed import instrument_provider_app _KEY = "shared-secret" _DIM = 8 _CONSUMER = "ratatoskr" class _RecordingSink: """An OpSink that just records events (so a test can assert on them).""" def __init__(self) -> None: self.events: list = [] def emit(self, event) -> None: self.events.append(event) def _mint(session_id: str, scope: list[str]) -> str: return mint_dispatch_jwt( session_id=session_id, consumer_id=_CONSUMER, issuer="worldtree", scope=scope, secret_or_key=_KEY, algorithm="HS256", ) def _wrapped_memory_app(sink): store = open_memory_store(":memory:", embedding_dim=_DIM) app = build_memory_provider_app( store, heimdall_key=_KEY.encode(), consumer_id=_CONSUMER ) return instrument_provider_app(app, plane="memory", sink=sink), store def _wrapped_affect_app(sink): store = open_affect_store(":memory:") app = build_affect_provider_app( store, heimdall_key=_KEY.encode(), consumer_id=_CONSUMER ) return instrument_provider_app(app, plane="affect", sink=sink), store def _chunk(cid: str, scope: dict | None = None) -> dict: return { "id": cid, "scope": scope or {"end_user": "u1"}, "embedding": [0.1] * _DIM, "content": "x", } class _RaisingSink: def emit(self, event) -> None: raise RuntimeError("boom") async def _post(app, path: str, body: dict, jwt: str | None) -> httpx.Response: transport = httpx.ASGITransport(app=app) headers = {"Authorization": f"Bearer {jwt}"} if jwt else {} async with httpx.AsyncClient( transport=transport, base_url="http://provider" ) as client: return await client.post(path, json=body, headers=headers) class TestOpFeedMemory: async def test_search_emits_one_opevent(self) -> None: """search [tracer]: a memory search dispatched through the wrapped app emits EXACTLY ONE OpEvent — plane=memory, op=search, session_id from the JWT sub, status=ok, scope-only req/resp summaries (POST-001/002, INV-005). Empty store → 0 hits.""" sink = _RecordingSink() app, _store = _wrapped_memory_app(sink) jwt = _mint("session-abc", ["memory:read"]) body = { "operation": "search", "args": { "vector": [0.0] * _DIM, "top_k": 5, "scope_all": {"end_user": "u1"}, "scope_any": [], }, } resp = await _post(app, "/bifrost/memory-call", body, jwt) assert resp.status_code == 200 assert len(sink.events) == 1 ev = sink.events[0] assert ev.plane == "memory" assert ev.op == "search" assert ev.session_id == "session-abc" assert ev.status == "ok" assert ev.req_summary == { "scope_all": {"end_user": "u1"}, "scope_any": [], "top_k": 5, } assert ev.resp_summary["hit_count"] == 0 assert ev.turn_id is None assert ev.ts # non-empty capture timestamp async def test_upsert_many_summary(self) -> None: """upsert_many: req carries record_count + per-record scopes (no bodies); resp carries upserted + replayed.""" sink = _RecordingSink() app, _store = _wrapped_memory_app(sink) jwt = _mint("session-up", ["memory:write"]) body = { "operation": "upsert_many", "args": {"records": [_chunk("c1"), _chunk("c2", {"end_user": "u2"})]}, "idempotency_key": "k1", } resp = await _post(app, "/bifrost/memory-call", body, jwt) assert resp.status_code == 200 ev = sink.events[-1] assert ev.op == "upsert_many" assert ev.status == "ok" assert ev.req_summary == { "record_count": 2, "scopes": [{"end_user": "u1"}, {"end_user": "u2"}], } assert ev.resp_summary == {"upserted": 2, "replayed": False} async def test_get_and_delete_summaries(self) -> None: """get -> found_count; delete_many -> deleted; both req carry ids only.""" sink = _RecordingSink() app, _store = _wrapped_memory_app(sink) await _post( app, "/bifrost/memory-call", { "operation": "upsert_many", "args": {"records": [_chunk("c1")]}, "idempotency_key": "k1", }, _mint("s", ["memory:write"]), ) await _post( app, "/bifrost/memory-call", {"operation": "get", "args": {"chunk_id": "c1"}}, _mint("s", ["memory:read"]), ) get_ev = sink.events[-1] assert get_ev.op == "get" assert get_ev.req_summary == {"ids": ["c1"]} assert get_ev.resp_summary == {"found_count": 1} await _post( app, "/bifrost/memory-call", {"operation": "delete_many", "args": {"ids": ["c1"]}}, _mint("s", ["memory:write"]), ) del_ev = sink.events[-1] assert del_ev.op == "delete_many" assert del_ev.req_summary == {"ids": ["c1"]} assert del_ev.resp_summary == {"deleted": 1} async def test_error_status_records_the_bifrost_code(self) -> None: """error [adversarial]: an unknown operation -> status=error and the bifrost error `code` is recorded, never hidden (INV-007).""" sink = _RecordingSink() app, _store = _wrapped_memory_app(sink) resp = await _post( app, "/bifrost/memory-call", {"operation": "bogus", "args": {}}, _mint("s", ["memory:read"]), ) assert resp.status_code != 200 assert len(sink.events) == 1 ev = sink.events[0] assert ev.op == "bogus" assert ev.status == "error" assert ev.resp_summary == {"error": "memory.invalid_arguments"} async def test_missing_jwt_session_id_none_still_emits(self) -> None: """no_jwt [boundary]: a call with NO Authorization still emits exactly one OpEvent with session_id=None (INV-005) and status=error (auth rejected).""" sink = _RecordingSink() app, _store = _wrapped_memory_app(sink) resp = await _post( app, "/bifrost/memory-call", {"operation": "search", "args": {"vector": [0.0] * _DIM, "top_k": 1}}, None, ) assert resp.status_code != 200 assert len(sink.events) == 1 assert sink.events[0].session_id is None assert sink.events[0].status == "error" async def test_handshake_op_from_path(self) -> None: """handshake: op is derived from the PATH (handshake bodies carry no `operation` field); still exactly one OpEvent (POST-001 incl. handshake).""" sink = _RecordingSink() app, _store = _wrapped_memory_app(sink) resp = await _post( app, "/bifrost/handshake", {"bifrost_version": "99.0.0", "mcp_version": "0.4.0"}, None, ) assert resp.status_code != 200 # version-major mismatch, cleanly rejected assert len(sink.events) == 1 assert sink.events[0].op == "handshake" async def test_handshake_req_summary_reads_real_capabilities_field(self) -> None: """The handshake req-summary reads the REAL wire field `capabilities` (bifrost _protocol.py:181), not the transposed `capabilities_requested` — so caps_requested is actually populated (heid-code-review #17 catch). A bad-version handshake still emits the OpEvent carrying the requested caps from the request body.""" sink = _RecordingSink() app, _store = _wrapped_memory_app(sink) resp = await _post( app, "/bifrost/handshake", {"bifrost_version": "99.0.0", "mcp_version": "0.4.0", "capabilities": ["memory"]}, None, ) assert resp.status_code != 200 assert len(sink.events) == 1 assert sink.events[0].req_summary == {"caps_requested": ["memory"]} async def test_sink_failure_never_breaks_dispatch(self) -> None: """sink_swallow [adversarial]: a raising sink must NOT break the dispatch path — the search still returns 200 (POST-003 / INV-007).""" app, _store = _wrapped_memory_app(_RaisingSink()) resp = await _post( app, "/bifrost/memory-call", { "operation": "search", "args": {"vector": [0.0] * _DIM, "top_k": 1, "scope_all": {}}, }, _mint("s", ["memory:read"]), ) assert resp.status_code == 200 class TestOpFeedAffect: async def test_emit_op_normalized_and_plane_affect(self) -> None: """affect emit: op is the bare verb (affect.emit -> emit), plane=affect, session_id from the JWT; affect stays conduit-opaque (empty req_summary).""" sink = _RecordingSink() app, _store = _wrapped_affect_app(sink) jwt = _mint("session-aff", ["affect:write"]) body = { "operation": "affect.emit", "args": { "agent_id": "ratatoskr:sindra", "end_user_id": "u1", "pad": {"p": 0.1, "a": 0.2, "d": 0.3}, "valence": 0.5, "emitted_at": "2026-06-18T00:00:00Z", }, "idempotency_key": "k1", } resp = await _post(app, "/bifrost/affect-call", body, jwt) assert resp.status_code == 200 assert len(sink.events) == 1 ev = sink.events[0] assert ev.plane == "affect" assert ev.op == "emit" assert ev.session_id == "session-aff" assert ev.status == "ok" assert ev.req_summary == {} # conduit-opaque assert ev.resp_summary == {"stored": True} async def test_pad_read_route_emits_no_opevent(self) -> None: """INV-004 (#18 D2): the non-bifrost PAD read route is OUTSIDE _BIFROST_PATHS, so the op-feed passes it through and records NO OpEvent — observe is bifrost- only and the read path adds no plane attribution.""" import json as _json sink = _RecordingSink() app, store = _wrapped_affect_app(sink) blob = _json.dumps( { "agent_id": "ratatoskr:sindra", "end_user_id": "vuong", "pad": {"pleasure": 0.1, "arousal": 0.0, "dominance": 0.0}, "valence": [], "emitted_at": "2026-06-18T00:00:00+00:00", }, sort_keys=True, separators=(",", ":"), ) store._conn.execute( "INSERT INTO affect_snapshots (agent_id, end_user_id, snapshot_json, arrived_at) " "VALUES (?, ?, ?, ?)", ("ratatoskr:sindra", "vuong", blob, "0"), ) store._conn.commit() transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient( transport=transport, base_url="http://provider" ) as client: resp = await client.get( "/affect/state/ratatoskr:sindra", params={"end_user_id": "vuong"} ) assert resp.status_code == 200 assert sink.events == [] # op-feed recorded nothing for the non-bifrost route class TestInstrumentGuards: def test_unknown_plane_raises(self) -> None: with pytest.raises(ValueError): instrument_provider_app(object(), plane="persona", sink=_RecordingSink())