feat(#17): dispatch-layer op-feed for the provider (slice 2 — Observe)

Slice 2 of issue #17 — the OBSERVE half. New ratatoskr.provider.opfeed:

- OpEvent{ts, plane, op, session_id, status, req_summary, resp_summary,
  turn_id=None} — scope-only summaries, never record bodies / PAD content
- OpSink Protocol + JsonlOpSink (continuous append-only JSONL, INV-007)
- instrument_provider_app(app, *, plane, sink): an ASGI middleware over the
  built bifrost provider app. Buffers+replays the request, captures the
  response, reads session_id off the dispatch JWT's "sub" claim (INV-005 —
  present for ALL verbs incl. search/get/delete, which bifrost withholds from
  the store method), emits exactly one OpEvent per inbound bifrost-call incl.
  handshake + errors. Read-only over dispatch; store scope semantics untouched
  (INV-004). A sink/summary failure is swallowed + logged, never breaks serve
  (POST-003).
- Per-verb summaries: search {scope_all,scope_any,top_k}->{hit_count,hits};
  upsert_many {record_count,scopes}->{upserted,replayed}; get/get_many/
  delete_many {ids}->{found_count|deleted}; emit (affect, opaque)->{stored};
  handshake {caps_requested}->{caps_granted,ok}; error->{error: code}
- serve_memory/serve wired: opt-in via RATATOSKR_OPFEED_PATH (maybe_instrument)

Resolves the contract's open question: the dispatch JWT DOES carry session_id
(= the "sub" claim). Tests drive the REAL bifrost dispatch end-to-end with
minted JWTs. 11 new tests; full suite 453 green; ruff + mypy clean (opfeed.py).

tests/test_provider_opfeed.py

@@ -0,0 +1,284 @@
+"""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_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}
+
+
+class TestInstrumentGuards:
+    def test_unknown_plane_raises(self) -> None:
+        with pytest.raises(ValueError):
+            instrument_provider_app(object(), plane="persona", sink=_RecordingSink())