diff --git a/pyproject.toml b/pyproject.toml index 76fee41..4435adb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "ratatoskr" -version = "0.17.8" +version = "0.17.9" description = "Worldtree Conversation API debug TUI — multi-pane observability dashboard" readme = "README.md" requires-python = ">=3.12" diff --git a/src/ratatoskr/provider/opfeed.py b/src/ratatoskr/provider/opfeed.py new file mode 100644 index 0000000..a35e442 --- /dev/null +++ b/src/ratatoskr/provider/opfeed.py @@ -0,0 +1,256 @@ +"""Dispatch-layer observe feed for the Bifrost provider (issue #17, Observe half). + +`instrument_provider_app` wraps a built provider ASGI app so every inbound +bifrost-call emits one structured `OpEvent` — correlated by `session_id` read off +the dispatch JWT — WITHOUT touching the store's scope semantics (INV-004). It is +the lens that lets ratatoskr, owning BOTH ends of the round-trip, see exactly +which memory/affect ops a given turn produced. + +The store-method stdout shim in `memory_store.py` cannot see `session_id` for +search/get/delete (bifrost withholds `ctx` from those store methods); this feed +sits at the DISPATCH/ASGI layer where the JWT — and thus `session_id` (its `sub` +claim) — is always present (INV-005). +""" +from __future__ import annotations + +import base64 +import json +import sys +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from typing import Any, Protocol + + +@dataclass(frozen=True) +class OpEvent: + """One observed bifrost-call at the dispatch layer (scope-only, never content).""" + + ts: str # ISO 8601 UTC, capture time + plane: str # "memory" | "affect" + op: str # verb: search / upsert_many / get / get_many / delete_many / emit / handshake + session_id: str | None # JWT `sub` at the dispatch layer; None only if the JWT omits it + status: str # "ok" | "error" + req_summary: dict[str, Any] # per-verb, scope-only — no record bodies + resp_summary: dict[str, Any] # per-verb counts + ids/scores — never verbatim content + turn_id: str | None = None # INV-005 reservation, literal; unused in v1 + + +class OpSink(Protocol): + """Anything that accepts an OpEvent. v1 ships JsonlOpSink; tests pass fakes.""" + + def emit(self, event: OpEvent) -> None: ... + + +class JsonlOpSink: + """Continuous append-only JSONL sink (INV-007: NOT per-session). One JSON line + per OpEvent to a text stream (default stdout).""" + + def __init__(self, stream: Any = None) -> None: + self._stream = sys.stdout if stream is None else stream + + def emit(self, event: OpEvent) -> None: + self._stream.write(json.dumps(asdict(event), separators=(",", ":")) + "\n") + self._stream.flush() + + +def maybe_instrument_from_env(app: Any, env: Any, *, plane: str) -> Any: + """Opt-in serve wiring: when `RATATOSKR_OPFEED_PATH` is set, wrap `app` with + the dispatch-layer op-feed writing JSONL to that path; otherwise return `app` + unchanged. The append stream lives for the process (a long-running server).""" + path = env.get("RATATOSKR_OPFEED_PATH") + if not path: + return app + stream = open(path, "a", encoding="utf-8") # process-lifetime append stream + return instrument_provider_app(app, plane=plane, sink=JsonlOpSink(stream)) + + +_BIFROST_PATHS = ( + "/bifrost/handshake", + "/bifrost/memory-call", + "/bifrost/affect-call", +) + + +def _b64url_decode(seg: str) -> bytes: + return base64.urlsafe_b64decode(seg + "=" * (-len(seg) % 4)) + + +def _session_id_from_auth(auth: bytes | None) -> str | None: + """Read the `sub` claim (= session_id, per bifrost DispatchContext) off the + dispatch JWT WITHOUT verifying its signature — the inner app does real + verification; we only read a claim for correlation. None if absent/malformed.""" + if not auth: + return None + try: + token = auth.decode("latin-1").strip() + if token.lower().startswith("bearer "): + token = token[7:].strip() + parts = token.split(".") + if len(parts) != 3: + return None + payload = json.loads(_b64url_decode(parts[1])) + sub = payload.get("sub") + return sub if isinstance(sub, str) else None + except Exception: + return None + + +def _op_from(path: str, req: dict[str, Any]) -> str: + """The verb: 'handshake' for the handshake path; otherwise the body's + `operation`, with the affect-plane `affect.` prefix stripped (affect.emit -> + emit) so op vocabulary stays bare per the contract.""" + if path == "/bifrost/handshake": + return "handshake" + operation = req.get("operation") or "unknown" + if path == "/bifrost/affect-call" and operation.startswith("affect."): + return operation.split(".", 1)[1] + return operation + + +def _ids_summary(args: dict[str, Any]) -> list[Any]: + """Mirror bifrost `_ids_arg`: ids | chunk_ids | [chunk_id|id].""" + ids = args.get("ids") or args.get("chunk_ids") + if ids is None: + single = args.get("chunk_id") or args.get("id") + ids = [single] if single is not None else [] + return ids + + +def _req_summary(plane: str, path: str, op: str, req: dict[str, Any]) -> dict[str, Any]: + """Scope-only request summary — NEVER record bodies / PAD content.""" + if path == "/bifrost/handshake": + return {"caps_requested": req.get("capabilities_requested")} + if plane == "affect": + return {} # affect stays conduit-opaque — no PAD content surfaced + args = req.get("args") or {} + if op == "search": + return { + "scope_all": args.get("scope_all") or {}, + "scope_any": args.get("scope_any") or [], + "top_k": args.get("top_k"), + } + if op == "upsert_many": + records = args.get("records") or [] + return { + "record_count": len(records), + "scopes": [r.get("scope") for r in records], + } + if op in ("get", "get_many", "delete_many"): + return {"ids": _ids_summary(args)} + return {} + + +def _resp_summary( + plane: str, path: str, op: str, resp: dict[str, Any], status: str +) -> dict[str, Any]: + """Per-verb counts + ids/scores — never verbatim content. On error, the + bifrost error `code` (INV-007: failures recorded, not hidden).""" + if status == "error": + return {"error": resp.get("code")} + if path == "/bifrost/handshake": + return {"ok": True, "caps_granted": resp.get("capabilities_granted")} + if plane == "affect": + return {"stored": bool(resp.get("stored"))} + if op == "search": + results = resp.get("results") or [] + return { + "hit_count": len(results), + "hits": [ + {"chunk_id": r.get("chunk_id"), "score": r.get("score")} + for r in results + ], + } + if op == "upsert_many": + return {"upserted": resp.get("upserted"), "replayed": resp.get("replayed")} + if op == "get": + return {"found_count": 1 if resp.get("record") else 0} + if op == "get_many": + return {"found_count": len(resp.get("records") or [])} + if op == "delete_many": + return {"deleted": resp.get("deleted")} + return {} + + +def _build_event( + plane: str, path: str, scope: dict[str, Any], req_body: bytes, captured: dict[str, Any] +) -> OpEvent: + headers = dict(scope.get("headers") or []) + session_id = _session_id_from_auth(headers.get(b"authorization")) + status = "ok" if 200 <= int(captured["status"]) < 300 else "error" + req = _safe_json(req_body) + resp = _safe_json(captured["body"]) + + op = _op_from(path, req) + return OpEvent( + ts=datetime.now(UTC).isoformat(), + plane=plane, + op=op, + session_id=session_id, + status=status, + req_summary=_req_summary(plane, path, op, req), + resp_summary=_resp_summary(plane, path, op, resp, status), + ) + + +def _safe_json(raw: bytes) -> dict[str, Any]: + if not raw: + return {} + try: + value = json.loads(raw) + except (json.JSONDecodeError, ValueError): + return {} + return value if isinstance(value, dict) else {} + + +def instrument_provider_app(app: Any, *, plane: str, sink: OpSink) -> Any: + """Wrap a built provider ASGI `app` so each inbound bifrost-call emits one + OpEvent to `sink`. Read-only over dispatch — store scope semantics untouched + (INV-004). A sink/summary failure never propagates into the dispatch path + (POST-003 / INV-007) — it is swallowed and logged to stderr. + """ + if plane not in ("memory", "affect"): + raise ValueError(f"plane must be 'memory' or 'affect', got {plane!r}") + + async def wrapped(scope: dict[str, Any], receive: Any, send: Any) -> None: + if scope.get("type") != "http" or scope.get("path") not in _BIFROST_PATHS: + await app(scope, receive, send) + return + + # Buffer the request body so we can BOTH summarise it AND replay it to the + # inner app (the inner app consumes `receive`; we must not steal the body). + req_body = b"" + more = True + while more: + message = await receive() + if message["type"] == "http.request": + req_body += message.get("body", b"") + more = message.get("more_body", False) + else: # http.disconnect + more = False + + replayed = False + + async def replay_receive() -> dict[str, Any]: + nonlocal replayed + if not replayed: + replayed = True + return {"type": "http.request", "body": req_body, "more_body": False} + return {"type": "http.disconnect"} + + captured: dict[str, Any] = {"status": 500, "body": b""} + + async def capture_send(message: dict[str, Any]) -> None: + if message["type"] == "http.response.start": + captured["status"] = message["status"] + elif message["type"] == "http.response.body": + captured["body"] += message.get("body", b"") + await send(message) + + await app(scope, replay_receive, capture_send) + + try: + sink.emit(_build_event(plane, scope["path"], scope, req_body, captured)) + except Exception as exc: # observe gap, never a serve break (POST-003) + sys.stderr.write(f"[opfeed] OpEvent emit failed (swallowed): {exc!r}\n") + + return wrapped diff --git a/src/ratatoskr/provider/serve.py b/src/ratatoskr/provider/serve.py index e0b1e4b..5bd3154 100644 --- a/src/ratatoskr/provider/serve.py +++ b/src/ratatoskr/provider/serve.py @@ -12,6 +12,7 @@ import os from collections.abc import Mapping from ratatoskr.provider.affect_store import build_affect_provider_app, open_affect_store +from ratatoskr.provider.opfeed import maybe_instrument_from_env def build_app_from_env(env: Mapping[str, str] | None = None): @@ -23,11 +24,13 @@ def build_app_from_env(env: Mapping[str, str] | None = None): "RATATOSKR_HEIMDALL_KEY is required to serve the affect provider" ) store = open_affect_store(env.get("RATATOSKR_AFFECT_DB", "affect.db")) - return build_affect_provider_app( + app = build_affect_provider_app( store, heimdall_key=key.encode(), consumer_id=env.get("RATATOSKR_CONSUMER_ID", "ratatoskr"), ) + # Issue #17 (Observe): opt-in dispatch-layer op-feed when RATATOSKR_OPFEED_PATH set. + return maybe_instrument_from_env(app, env, plane="affect") def main() -> None: diff --git a/src/ratatoskr/provider/serve_memory.py b/src/ratatoskr/provider/serve_memory.py index 44c6dd1..ae7eb67 100644 --- a/src/ratatoskr/provider/serve_memory.py +++ b/src/ratatoskr/provider/serve_memory.py @@ -17,6 +17,7 @@ import os from collections.abc import Mapping from ratatoskr.provider.memory_store import build_memory_provider_app, open_memory_store +from ratatoskr.provider.opfeed import maybe_instrument_from_env def build_memory_app_from_env(env: Mapping[str, str] | None = None): @@ -43,11 +44,13 @@ def build_memory_app_from_env(env: Mapping[str, str] | None = None): store = open_memory_store( env.get("RATATOSKR_MEMORY_DB", "memory.db"), embedding_dim=embedding_dim ) - return build_memory_provider_app( + app = build_memory_provider_app( store, heimdall_key=key.encode(), consumer_id=env.get("RATATOSKR_CONSUMER_ID", "ratatoskr"), ) + # Issue #17 (Observe): opt-in dispatch-layer op-feed when RATATOSKR_OPFEED_PATH set. + return maybe_instrument_from_env(app, env, plane="memory") def main() -> None: diff --git a/tests/test_provider_opfeed.py b/tests/test_provider_opfeed.py new file mode 100644 index 0000000..0f19be6 --- /dev/null +++ b/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()) diff --git a/tests/test_provider_serve.py b/tests/test_provider_serve.py index 0577b42..f094c34 100644 --- a/tests/test_provider_serve.py +++ b/tests/test_provider_serve.py @@ -21,3 +21,16 @@ def test_build_app_from_env_builds_app_with_routes(): paths = {getattr(r, "path", None) for r in app.routes} assert "/bifrost/handshake" in paths assert "/bifrost/affect-call" in paths + + +def test_opfeed_path_wraps_app(tmp_path): + # Issue #17 slice 2: RATATOSKR_OPFEED_PATH opts the dispatch op-feed in; the + # returned app is then the instrumented ASGI wrapper, not the raw Starlette. + app = build_app_from_env( + { + "RATATOSKR_HEIMDALL_KEY": "shared-secret", + "RATATOSKR_AFFECT_DB": ":memory:", + "RATATOSKR_OPFEED_PATH": str(tmp_path / "ops.jsonl"), + } + ) + assert not hasattr(app, "routes") # wrapped: a bare ASGI callable diff --git a/tests/test_provider_serve_memory.py b/tests/test_provider_serve_memory.py index 3bd53ae..db3096c 100644 --- a/tests/test_provider_serve_memory.py +++ b/tests/test_provider_serve_memory.py @@ -46,3 +46,17 @@ def test_build_memory_app_from_env_builds_app_with_routes(): paths = {getattr(r, "path", None) for r in app.routes} assert "/bifrost/handshake" in paths assert "/bifrost/memory-call" in paths + + +def test_opfeed_path_wraps_app(tmp_path): + # Issue #17 slice 2: RATATOSKR_OPFEED_PATH opts the dispatch op-feed in; the + # returned app is then the instrumented ASGI wrapper, not the raw Starlette. + app = build_memory_app_from_env( + { + "RATATOSKR_HEIMDALL_KEY": "shared-secret", + "RATATOSKR_MEMORY_DB": ":memory:", + "RATATOSKR_MEMORY_EMBEDDING_DIM": "8", + "RATATOSKR_OPFEED_PATH": str(tmp_path / "ops.jsonl"), + } + ) + assert not hasattr(app, "routes") # wrapped: a bare ASGI callable diff --git a/uv.lock b/uv.lock index 3700845..bb30919 100644 --- a/uv.lock +++ b/uv.lock @@ -1052,7 +1052,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.17.8" +version = "0.17.9" source = { editable = "." } dependencies = [ { name = "httpx" },