"""Tests for evidence resolver, workspace info, and maintenance origination SDK functions.""" from __future__ import annotations import unittest from klbr_runtime.context import ( Bridge, ExecutionScope, assess_maintenance, current, originate_maintenance, resolve_evidence, workspace_info, ) from klbr.maintenance import load_maintenance_evidence class MockWire: def __init__(self, responses: dict | None = None): self.responses = responses or {} self.sent = [] self.bridge: Bridge | None = None async def send(self, msg): self.sent.append(msg) req_method = msg.get("request", {}).get("method") if req_method in self.responses and self.bridge is not None: self.bridge.reply({ "kind": "host_reply", "id": msg["id"], "execution_id": msg["execution_id"], "reply": {"kind": "ok", "value": self.responses[req_method]}, }) class EvidenceResolverTests(unittest.IsolatedAsyncioTestCase): """Verify evidence resolver and workspace info SDK functions.""" async def test_resolve_evidence_parses_refs(self): expected_result = { "resolved": [ { "ref": "event:42", "kind": "event", "event": {"id": 42, "kind": "tool.result", "payload": {"error": "TypeError"}, "created_ms": 1000}, "workspace": {"path": "/measured/workspace", "session_id": "test-session"}, "revision": {"id": "rev-abc", "epoch": 1}, } ], "unavailable": [ {"ref": "event:999", "reason": "not_found"} ], } wire = MockWire({"evidence.resolve": expected_result}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope) try: result = await resolve_evidence(["event:42", "event:999"]) self.assertEqual(len(result["resolved"]), 1) self.assertEqual(result["resolved"][0]["ref"], "event:42") self.assertEqual(result["resolved"][0]["workspace"]["path"], "/measured/workspace") self.assertEqual(len(result["unavailable"]), 1) self.assertEqual(result["unavailable"][0]["ref"], "event:999") self.assertEqual(result["unavailable"][0]["reason"], "not_found") finally: current.reset(token) async def test_workspace_info_returns_measured_path_and_session(self): expected_result = { "path": "/measured/workspace/project", "session_id": "session-xyz", "active_revision": "rev-123", "epoch": 2, } wire = MockWire({"workspace.info": expected_result}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope) try: result = await workspace_info() self.assertEqual(result["path"], "/measured/workspace/project") self.assertEqual(result["session_id"], "session-xyz") self.assertEqual(result["active_revision"], "rev-123") self.assertEqual(result["epoch"], 2) finally: current.reset(token) async def test_load_maintenance_evidence_builds_packet_with_missing(self): wire_data = { "resolved": [ { "ref": "event:10", "kind": "event", "event": {"id": 10, "kind": "scratchpad", "payload": {"error": "IndexError: out of bounds"}}, } ], "unavailable": [ {"ref": "source:missing.py", "reason": "file not found in workspace or release"} ], } wire = MockWire({"evidence.resolve": wire_data}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope) try: evidence = await load_maintenance_evidence(["event:10", "source:missing.py"]) self.assertEqual(evidence.observation_id, "event:10") self.assertEqual(evidence.error_trace, "IndexError: out of bounds") self.assertIn("source:missing.py", evidence.missing_evidence) self.assertEqual(evidence.evidence_refs, ("event:10", "source:missing.py")) finally: current.reset(token) async def test_originate_maintenance_requests_runnable_work(self): expected_reply = { "item_id": "work-maint-123", "work_item_id": "work-maint-123", "seq": 45, } wire = MockWire({"maintenance.originate": expected_reply}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope) try: res = await originate_maintenance( purpose="repair", reason="failure observed in check", evidence_refs=["event:10"], impact="high", ) self.assertEqual(res["item_id"], "work-maint-123") self.assertEqual(res["seq"], 45) self.assertEqual(wire.sent[0]["request"]["method"], "maintenance.originate") self.assertEqual(wire.sent[0]["request"]["purpose"], "repair") finally: current.reset(token) async def test_assess_maintenance_requests_host_recording(self): expected_reply = { "status": "recorded", "work_item_id": "work-assess-1", "seq": 46, } wire = MockWire({"maintenance.assess": expected_reply}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope) try: res = await assess_maintenance( work_item_id="work-assess-1", outcome="needs-evidence", reason="missing file", missing_refs=["source:foo.py"], revisit_condition="awaiting file", candidate_revision="rev-1", check_receipts=["check-1"], ) self.assertEqual(res["status"], "recorded") self.assertEqual(res["work_item_id"], "work-assess-1") self.assertEqual(wire.sent[0]["request"]["method"], "maintenance.assess") self.assertEqual(wire.sent[0]["request"]["outcome"], "needs-evidence") self.assertEqual(wire.sent[0]["request"]["missing_refs"], ["source:foo.py"]) finally: current.reset(token) class AssessMaintenanceBoundaryTests(unittest.IsolatedAsyncioTestCase): """The SDK validates the closed assessment outcome set before forwarding.""" def _scope_and_wire(self, expected_reply: dict) -> tuple: wire = MockWire({"maintenance.assess": expected_reply}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope) return scope, wire, token async def test_invalid_outcome_fails_locally_and_is_not_forwarded(self): scope, wire, token = self._scope_and_wire({"status": "recorded"}) try: with self.assertRaises(ValueError) as ctx: await assess_maintenance( work_item_id="work-assess-1", outcome="actviated", reason="fixture typo", ) self.assertIn("is not a disposition for a work item", str(ctx.exception)) self.assertEqual(wire.sent, [], "an invalid outcome must never reach the host") finally: current.reset(token) async def test_missing_required_variant_fields_fail_locally(self): scope, wire, token = self._scope_and_wire({"status": "recorded"}) try: with self.assertRaises(ValueError): await assess_maintenance( work_item_id="work-assess-1", outcome="no-change", ) with self.assertRaises(ValueError): await assess_maintenance( work_item_id="work-assess-1", outcome="needs-evidence", ) with self.assertRaises(ValueError): await assess_maintenance( work_item_id="work-assess-1", outcome="activated", check_receipts=["event:10"], ) self.assertEqual(wire.sent, []) finally: current.reset(token) async def test_valid_assessments_forward_the_closed_fields(self): expected_reply = {"status": "recorded", "work_item_id": "work-assess-1", "seq": 46} wire = MockWire({"maintenance.assess": expected_reply}) bridge = Bridge(wire) wire.bridge = bridge scope = ExecutionScope(id="test-op", bridge=bridge) token = current.set(scope) try: res = await assess_maintenance( work_item_id="work-assess-1", outcome="needs-evidence", reason="missing file", missing_refs=["source:foo.py"], revisit_condition="awaiting file", revisit_deadline_ms=50_000, ) self.assertEqual(res["status"], "recorded") self.assertEqual(res["work_item_id"], "work-assess-1") self.assertEqual(wire.sent[0]["request"]["method"], "maintenance.assess") self.assertEqual(wire.sent[0]["request"]["outcome"], "needs-evidence") self.assertEqual(wire.sent[0]["request"]["missing_refs"], ["source:foo.py"]) self.assertEqual(wire.sent[0]["request"]["revisit_deadline_ms"], 50_000) res = await assess_maintenance( work_item_id="work-assess-1", outcome="deferred", reason="checkpoint retained", trigger_ms=60_000, ) self.assertEqual(res["status"], "recorded") self.assertEqual(wire.sent[-1]["request"]["outcome"], "deferred") self.assertEqual(wire.sent[-1]["request"]["trigger_ms"], 60_000) finally: current.reset(token) if __name__ == "__main__": unittest.main()