"""Phase 3 — Events payload context fields for the Tenet 6 surface. Spec: consultations/recoil/bugfix-sprint-post-overhaul/BUILD_SPEC.md, Phase 3. Pre-fix bug: ``recoil/api/adapters/events.py:_receipt_to_event()`` built ``EngineEvent.payload`` with only ``{modality, caller_id, model}``. The Console v2 ChatColumn / activity-pane (Tenet 6) needs more — failure_mode, final_state, cost, eval, shot context — to render usefully. Post-fix invariant: payload carries the original three keys byte-identical plus seven new fields (failure_mode, final_state, cost_usd, eval_score, shot_id, episode, project). All values nullable. No exceptions on malformed / minimal receipts. """ from __future__ import annotations import pytest from recoil.api.adapters.events import _receipt_to_event # ── Helpers ────────────────────────────────────────────────────────────── def _full_receipt(**overrides): """A maximally-populated receipt dict mirroring CP-9 emission shape.""" base = { "receipt_id": "rcpt_1777354477422531_SH01_image_t2i", "modality": "image_t2i", "caller_id": "run_shot", "project": "test-proj", "episode": 3, "shot_id": "EP003_SH01", "timestamp_utc": "2026-04-28T05:34:37Z", "run_result": { "id": "SH01_image_t2i_1777354477", "modality": "image_t2i", "output_path": None, "output_url": None, "metadata": { "final_state": "succeeded", "cost_usd": 0.042, "gate_verdict": "pass", "take_index": 0, "model": "gemini-3-pro-preview", "pipeline": "keyframe", }, "success": True, "error": None, }, "provenance": { "dispatch_path": "run_shot", "model": "gemini-3-pro-preview", }, "eval_scores": { "visual_quality_v1": { "panel_id": "visual_quality_v1", "panel_score": 0.83, "panel_warnings": [], "judges": [], "aggregation": "median", "panel_cost_usd": 0.05, }, }, } base.update(overrides) return base # ── Required tests (Phase 3 exit criteria) ────────────────────────────── def test_receipt_to_event_payload_contains_failure_mode_when_present(): """failure_mode probed from run_result.metadata wins when top-level absent.""" rcpt = _full_receipt() rcpt["run_result"]["metadata"]["failure_mode"] = "extra_limbs" ev = _receipt_to_event(rcpt) assert ev is not None assert ev.payload["failure_mode"] == "extra_limbs" # Top-level wins when both present (top-level is the newer surface). rcpt_top = _full_receipt() rcpt_top["failure_mode"] = "background_contamination" rcpt_top["run_result"]["metadata"]["failure_mode"] = "extra_limbs" ev_top = _receipt_to_event(rcpt_top) assert ev_top.payload["failure_mode"] == "background_contamination" def test_receipt_to_event_payload_contains_cost_usd_when_present(): rcpt = _full_receipt() ev = _receipt_to_event(rcpt) assert ev is not None assert ev.payload["cost_usd"] == 0.042 assert ev.payload["final_state"] == "succeeded" def test_receipt_to_event_payload_contains_shot_episode_project_when_present(): rcpt = _full_receipt() ev = _receipt_to_event(rcpt) assert ev is not None assert ev.payload["shot_id"] == "EP003_SH01" assert ev.payload["episode"] == 3 assert ev.payload["project"] == "test-proj" def test_receipt_to_event_payload_handles_missing_metadata_gracefully(): """Minimal receipt: empty run_result, no metadata, no eval_scores. All extended keys must default to None and no exception may bubble out.""" rcpt = { "receipt_id": "rcpt_minimal", "modality": "image_t2i", "run_result": {}, } ev = _receipt_to_event(rcpt) assert ev is not None p = ev.payload # All extended keys default to None. assert p["failure_mode"] is None assert p["final_state"] is None assert p["cost_usd"] is None assert p["eval_score"] is None assert p["shot_id"] is None assert p["episode"] is None assert p["project"] is None def test_legacy_keys_unchanged(): """modality, caller_id, model must remain byte-identical to pre-Phase-3.""" rcpt = _full_receipt() ev = _receipt_to_event(rcpt) assert ev is not None p = ev.payload assert p["modality"] == "image_t2i" assert p["caller_id"] == "run_shot" assert p["model"] == "gemini-3-pro-preview" # ── Additional coverage on the eval_score collapse ────────────────────── def test_eval_score_collapses_to_mean_of_numeric_panel_scores(): """Multi-panel eval_scores → mean of non-None panel_scores (matches Take.compute_aggregate_score).""" rcpt = _full_receipt() rcpt["eval_scores"] = { "panel_a": {"panel_id": "panel_a", "panel_score": None}, "panel_b": {"panel_id": "panel_b", "panel_score": 0.71}, "panel_c": {"panel_id": "panel_c", "panel_score": 0.95}, } ev = _receipt_to_event(rcpt) assert ev.payload["eval_score"] == pytest.approx(0.83) def test_eval_score_none_when_no_panels(): rcpt = _full_receipt() rcpt["eval_scores"] = {} ev = _receipt_to_event(rcpt) assert ev.payload["eval_score"] is None def test_eval_score_none_when_all_panel_scores_null(): rcpt = _full_receipt() rcpt["eval_scores"] = { "panel_a": {"panel_id": "panel_a", "panel_score": None}, "panel_b": {"panel_id": "panel_b", "panel_score": None}, } ev = _receipt_to_event(rcpt) assert ev.payload["eval_score"] is None def test_eval_score_probes_run_result_when_top_level_absent(): """Forward-compat: if a future migration parks eval_scores under run_result, the adapter still surfaces a score.""" rcpt = _full_receipt() rcpt.pop("eval_scores", None) rcpt["run_result"]["eval_scores"] = { "panel_x": {"panel_score": 0.42}, } ev = _receipt_to_event(rcpt) assert ev.payload["eval_score"] == 0.42 def test_episode_string_is_coerced_to_int_when_possible(): rcpt = _full_receipt(episode="7") ev = _receipt_to_event(rcpt) assert ev.payload["episode"] == 7 def test_episode_unparseable_becomes_none(): rcpt = _full_receipt(episode="not-a-number") ev = _receipt_to_event(rcpt) assert ev.payload["episode"] is None def test_payload_has_exactly_ten_keys(): """Schema lock — three legacy keys + seven new keys = ten payload keys.""" rcpt = _full_receipt() ev = _receipt_to_event(rcpt) expected = { "modality", "caller_id", "model", "failure_mode", "final_state", "cost_usd", "eval_score", "shot_id", "episode", "project", } assert set(ev.payload.keys()) == expected