ratatoskr/tests/test_provider_opfeed.py

"""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())