"""router.decisions.list / router.feedback.submit RPC handlers. The list surface is a read/observe view over the V017 ``router_decisions`` table; the feedback surface is dormant plumbing (deferred F7 follow-up). All fixture data is synthetic. """ from __future__ import annotations from datetime import UTC, datetime, timedelta from pathlib import Path import pytest from opensquilla.engine.steps.router_decision_record import ( get_decision_writer, set_decision_writer, ) from opensquilla.gateway.auth import Principal from opensquilla.gateway.config import GatewayConfig from opensquilla.gateway.protocol import ERROR_INVALID_REQUEST, ERROR_UNAUTHORIZED from opensquilla.gateway.rpc import get_dispatcher, validate_classification from opensquilla.gateway.rpc.registry import RpcContext from opensquilla.gateway.rpc_router import ( _bounded_limit, _handle_router_decisions_list, _handle_router_feedback_submit, ) from opensquilla.gateway.scopes import METHOD_SCOPES, READ_SCOPE, WRITE_SCOPE from opensquilla.persistence.migrator import apply_pending from opensquilla.persistence.router_decision_writer import ( RouterDecisionWriter, open_router_decision_writer, ) MIGRATIONS_DIR = Path(__file__).resolve().parents[1].parent / "migrations" EXPECTED_WIRE_KEYS = { "decisionId", "sessionKey", "turnIndex", "tsMs", "classifier", "proposedTier", "confidence", "probs", "flags", "finalTier", "provider", "model", "thinkingLevel", "source", "trail", "baselineModel", "savingsPct", "executedKind", "ensembleProfile", "fallbackHops", } def _base_record(**overrides) -> dict: record = { "decision_id": "d" * 32, "session_key": "agent:main:webchat:s1", "turn_index": 0, "ts_ms": 1_000_000, "classifier": "v4_phase3", "proposed_tier": "c1", "confidence": 0.91, "probs": [0.05, 0.91, 0.03, 0.01], "flags": ["code", "multi_step"], "final_tier": "c2", "provider": "openrouter", "model": "deepseek/deepseek-chat", "thinking_level": "medium", "source": "v4_phase3", "trail": [ {"stage": "classify", "tier": "c1", "route_class": "R1"}, {"stage": "final", "tier": "c2", "route_class": "R2"}, ], "baseline_model": "anthropic/claude-sonnet", "savings_pct": 42.5, "executed_kind": "single", "ensemble_profile": None, "fallback_hops": 0, } record.update(overrides) return record @pytest.fixture def writer(tmp_path: Path): """Real migrated DB + registered process-wide writer, torn down after.""" db = str(tmp_path / "sessions.sqlite") apply_pending(db, MIGRATIONS_DIR) w = open_router_decision_writer(db) previous = get_decision_writer() set_decision_writer(w) try: yield w finally: set_decision_writer(previous) w.close() def _read_only_principal() -> Principal: return Principal( role="operator", scopes=frozenset({READ_SCOPE}), is_owner=False, authenticated=True, ) # --------------------------------------------------------------------------- # router.decisions.list # --------------------------------------------------------------------------- async def test_decisions_list_empty_without_writer() -> None: previous = get_decision_writer() set_decision_writer(None) try: payload = await _handle_router_decisions_list({}, RpcContext(conn_id="test")) finally: set_decision_writer(previous) assert payload == {"decisions": []} async def test_decisions_list_returns_camelcase_envelope( writer: RouterDecisionWriter, ) -> None: assert writer.record_decision(_base_record()) is True payload = await _handle_router_decisions_list({}, RpcContext(conn_id="test")) assert set(payload) == {"decisions"} (decision,) = payload["decisions"] assert set(decision) == EXPECTED_WIRE_KEYS assert decision["decisionId"] == "d" * 32 assert decision["sessionKey"] == "agent:main:webchat:s1" assert decision["turnIndex"] == 0 assert decision["tsMs"] == 1_000_000 assert decision["classifier"] == "v4_phase3" assert decision["proposedTier"] == "c1" assert decision["confidence"] == 0.91 assert decision["finalTier"] == "c2" assert decision["provider"] == "openrouter" assert decision["model"] == "deepseek/deepseek-chat" assert decision["thinkingLevel"] == "medium" assert decision["source"] == "v4_phase3" assert decision["baselineModel"] == "anthropic/claude-sonnet" assert decision["executedKind"] == "single" assert decision["ensembleProfile"] is None assert decision["fallbackHops"] == 0 # JSON columns come back as structured JSON, not serialized strings. assert decision["probs"] == [0.05, 0.91, 0.03, 0.01] assert decision["flags"] == ["code", "multi_step"] assert decision["trail"][0] == {"stage": "classify", "tier": "c1", "route_class": "R1"} async def test_decisions_list_savings_pct_verbatim_passthrough( writer: RouterDecisionWriter, ) -> None: """C2: savingsPct surfaces the stored column value untouched.""" writer.record_decision(_base_record(decision_id="s1", savings_pct=42.5)) writer.record_decision( _base_record(decision_id="s2", ts_ms=1_000_001, savings_pct=None) ) payload = await _handle_router_decisions_list({}, RpcContext(conn_id="test")) by_id = {d["decisionId"]: d for d in payload["decisions"]} assert by_id["s1"]["savingsPct"] == 42.5 assert by_id["s2"]["savingsPct"] is None async def test_decisions_list_orders_newest_first_and_pages_with_before_ts( writer: RouterDecisionWriter, ) -> None: for index in range(3): writer.record_decision( _base_record(decision_id=f"p{index}", ts_ms=1_000 * (index + 1)) ) ctx = RpcContext(conn_id="test") payload = await _handle_router_decisions_list({}, ctx) assert [d["tsMs"] for d in payload["decisions"]] == [3_000, 2_000, 1_000] oldest_seen = payload["decisions"][0]["tsMs"] # page after the newest row page = await _handle_router_decisions_list({"beforeTs": oldest_seen}, ctx) assert [d["tsMs"] for d in page["decisions"]] == [2_000, 1_000] async def test_decisions_list_filters_by_session_key( writer: RouterDecisionWriter, ) -> None: writer.record_decision(_base_record(decision_id="a1", session_key="agent:a")) writer.record_decision(_base_record(decision_id="b1", session_key="agent:b")) payload = await _handle_router_decisions_list( {"sessionKey": "agent:a"}, RpcContext(conn_id="test") ) assert [d["decisionId"] for d in payload["decisions"]] == ["a1"] assert all(d["sessionKey"] == "agent:a" for d in payload["decisions"]) async def test_decisions_list_respects_limit(writer: RouterDecisionWriter) -> None: for index in range(3): writer.record_decision( _base_record(decision_id=f"l{index}", ts_ms=1_000 * (index + 1)) ) payload = await _handle_router_decisions_list( {"limit": 1}, RpcContext(conn_id="test") ) assert [d["decisionId"] for d in payload["decisions"]] == ["l2"] def test_decisions_list_limit_is_clamped() -> None: assert _bounded_limit(None) == 50 assert _bounded_limit(-1) == 50 assert _bounded_limit("oops") == 50 assert _bounded_limit("5000") == 200 assert _bounded_limit(5000) == 200 assert _bounded_limit("12") == 12 async def test_decisions_list_allows_read_only_dispatch( writer: RouterDecisionWriter, ) -> None: writer.record_decision(_base_record()) ctx = RpcContext(conn_id="test", principal=_read_only_principal()) res = await get_dispatcher().dispatch("r1", "router.decisions.list", {}, ctx) assert res.error is None, res.error assert len(res.payload["decisions"]) == 1 # --------------------------------------------------------------------------- # router.feedback.submit (live F7 intake) # --------------------------------------------------------------------------- async def test_feedback_submit_records_to_sidecar( writer: RouterDecisionWriter, tmp_path: Path, monkeypatch ) -> None: """A rating resolves through V017 and lands in the per-agent sidecar.""" monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(tmp_path)) writer.record_decision(_base_record()) before = writer.list_decisions() payload = await _handle_router_feedback_submit( {"decisionId": "d" * 32, "rating": "down"}, RpcContext(conn_id="test"), ) assert payload == {"accepted": True, "recorded": "down"} # The decision table itself is never mutated by feedback. assert writer.list_decisions() == before from opensquilla.squilla_router.self_learning.feedback import load_feedback_map fb = load_feedback_map("main", home=tmp_path) assert fb["d" * 32].rating == "down" assert fb["d" * 32].executed_kind == "single" async def test_feedback_submit_uses_configured_retention_days( writer: RouterDecisionWriter, tmp_path: Path, monkeypatch ) -> None: monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(tmp_path)) writer.record_decision(_base_record()) from opensquilla.squilla_router.self_learning.feedback import ( load_feedback_map, write_feedback, ) write_feedback( "main", decision_id="old-rating", session_key="agent:main:webchat:s1", turn_index=1, rating="up", now=datetime.now(UTC) - timedelta(days=40), retention_days=60, ) cfg = GatewayConfig( squilla_router={"self_learning": {"retention_days": 60}} ) payload = await _handle_router_feedback_submit( {"decisionId": "d" * 32, "rating": "down"}, RpcContext(conn_id="test", config=cfg), ) assert payload == {"accepted": True, "recorded": "down"} assert "old-rating" in load_feedback_map("main", home=tmp_path) async def test_feedback_submit_unknown_decision_is_soft_failure( writer: RouterDecisionWriter, tmp_path: Path, monkeypatch ) -> None: monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(tmp_path)) payload = await _handle_router_feedback_submit( {"decisionId": "f" * 32, "rating": "up"}, RpcContext(conn_id="test"), ) assert payload == {"accepted": False, "reason": "decision_not_found"} async def test_feedback_submit_last_write_wins_and_neutral_revokes( writer: RouterDecisionWriter, tmp_path: Path, monkeypatch ) -> None: monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(tmp_path)) writer.record_decision(_base_record()) from opensquilla.squilla_router.self_learning.feedback import load_feedback_map ctx = RpcContext(conn_id="test") await _handle_router_feedback_submit({"decisionId": "d" * 32, "rating": "down"}, ctx) await _handle_router_feedback_submit({"decisionId": "d" * 32, "rating": "up"}, ctx) fb = load_feedback_map("main", home=tmp_path) assert fb["d" * 32].rating == "up" # revision wins await _handle_router_feedback_submit( {"decisionId": "d" * 32, "rating": "neutral"}, ctx ) assert load_feedback_map("main", home=tmp_path) == {} # revoked async def test_feedback_submit_preserves_ensemble_kind( writer: RouterDecisionWriter, tmp_path: Path, monkeypatch ) -> None: """executed_kind rides from V017 into the sidecar for downstream gating.""" monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(tmp_path)) writer.record_decision(_base_record(decision_id="e" * 32, executed_kind="ensemble")) await _handle_router_feedback_submit( {"decisionId": "e" * 32, "rating": "down"}, RpcContext(conn_id="test"), ) from opensquilla.squilla_router.self_learning.feedback import load_feedback_map fb = load_feedback_map("main", home=tmp_path) assert fb["e" * 32].executed_kind == "ensemble" async def test_feedback_submit_rejects_free_text_decision_id() -> None: ctx = RpcContext(conn_id="test") res = await get_dispatcher().dispatch( "r1", "router.feedback.submit", {"decisionId": "this routing was wrong today", "rating": "down"}, ctx, ) assert res.error is not None assert res.error.code == ERROR_INVALID_REQUEST async def test_feedback_submit_rejects_free_text_rating() -> None: ctx = RpcContext(conn_id="test") res = await get_dispatcher().dispatch( "r1", "router.feedback.submit", {"decisionId": "d" * 32, "rating": "amazing model, keep it!"}, ctx, ) assert res.error is not None assert res.error.code == ERROR_INVALID_REQUEST async def test_feedback_submit_denies_read_only_dispatch() -> None: ctx = RpcContext(conn_id="test", principal=_read_only_principal()) res = await get_dispatcher().dispatch( "r1", "router.feedback.submit", {"decisionId": "d" * 32, "rating": "up"}, ctx, ) assert res.error is not None assert res.error.code == ERROR_UNAUTHORIZED def test_feedback_handler_is_dormant_static() -> None: """The handler module must not touch routing, calibration, or selection. The read-only ``router.selflearning.status`` handler may import the self-learning *state readers* (gates evaluation, pointer/state/store reads) — those observe the loop without feeding routing. What stays forbidden is anything that could route, calibrate, or mutate loop state: the routing engines themselves, and the self-learning mutation surfaces (training, promotion pointer writes, sample writes). """ source = Path("src/opensquilla/gateway/rpc_router.py").read_text() assert "RoutingHistoryStore" not in source import_lines = [ line.strip() for line in source.splitlines() if line.strip().startswith(("import ", "from ")) ] allowed_readonly = ( "squilla_router.self_learning.gates", "squilla_router.self_learning.promotion", "squilla_router.self_learning.state", "squilla_router.self_learning.store", # Feedback intake is this module's own job: append-only sidecar writes, # still nothing that routes, calibrates, or trains. "squilla_router.self_learning.feedback", ) forbidden = ("smart_routing", "router_control", "squilla_router", "calibration", "routing") for line in import_lines: if any(mod in line for mod in allowed_readonly): continue assert not any(token in line for token in forbidden), line # The status handler must stay read-only: no training/mutation imports. # ("train" as a bare token would false-positive on "training"/"trainedAt", # so the mutation modules are matched as import paths.) for mutating in ( "self_learning.orchestrator", "self_learning.train", "write_sample", "write_active_atomic", "promote_candidate", "rollback_active", "quarantine_candidate", ): assert mutating not in source, mutating # --------------------------------------------------------------------------- # Scope classification / boot audit # --------------------------------------------------------------------------- def test_router_rpc_scope_contract() -> None: assert METHOD_SCOPES["router.decisions.list"] == READ_SCOPE assert METHOD_SCOPES["router.feedback.submit"] == WRITE_SCOPE def test_router_rpc_methods_pass_boot_scope_audit() -> None: registry = get_dispatcher() assert "router.decisions.list" in registry.methods() assert "router.feedback.submit" in registry.methods() # Same audit boot runs at the end of opensquilla.gateway.rpc.__init__; # raises ScopeDriftError on declared-vs-table drift. validate_classification(registry)