video-image-production-build/tests/observability/test_production_runtime_observation.py

from __future__ import annotations

from pathlib import Path

from production_build_agents.production import runtime


class _Graph:
    def __init__(self, saved: dict, result: dict | None = None) -> None:
        self.saved = saved
        self.result = result
        self.invoke_calls = 0

    def get_state(self, _config):
        return type("_Snapshot", (), {"values": self.saved})()

    def invoke(self, _state, _config):
        self.invoke_calls += 1
        return self.result or self.saved


class _Observation:
    created: list["_Observation"] = []

    def __init__(self, **payload) -> None:
        self.payload = payload
        self.finished: list[tuple[dict, bool]] = []
        self.__class__.created.append(self)

    def __enter__(self):
        return self

    def finish(self, result, *, graph_invoked: bool) -> None:
        self.finished.append((dict(result), graph_invoked))

    def __exit__(self, *_args):
        return False


def _paths(tmp_path: Path) -> tuple[Path, Path]:
    delivery = tmp_path / "GlobalDataStageDelivery.v1.json"
    anchor = tmp_path / "anchor.png"
    delivery.write_text("{}", encoding="utf-8")
    anchor.write_bytes(b"\x89PNG")
    return delivery, anchor


def test_runtime_reports_completed_noop_without_invoking_graph(
    monkeypatch,
    tmp_path: Path,
) -> None:
    delivery, anchor = _paths(tmp_path)
    saved = {
        "run_id": "round-production",
        "protocol_version": "0.7",
        "status": "COMPLETED",
        "phase": "FINALIZE",
        "max_plan_revisions": 5,
    }
    graph = _Graph(saved)
    _Observation.created.clear()
    monkeypatch.setattr(runtime, "create_production_graph", lambda **_k: graph)
    monkeypatch.setattr(runtime, "ProductionObservation", _Observation)
    monkeypatch.setattr(runtime, "validate_resume_identity", lambda *_a, **_k: None)
    monkeypatch.setattr(
        runtime,
        "validate_completed_production_run",
        lambda *_a, **_k: None,
    )

    result = runtime.run_production_graph(
        delivery,
        output_root=tmp_path / "runs",
        thread_id="round-production",
        shared_visual_anchor_path=anchor,
        registry_dir=tmp_path / "registry",
    )

    observation = _Observation.created[0]
    assert result == saved
    assert graph.invoke_calls == 0
    assert observation.payload["execution_mode"] == "completed_noop"
    assert observation.finished == [(saved, False)]


def test_runtime_reports_fresh_graph_execution(
    monkeypatch,
    tmp_path: Path,
) -> None:
    delivery, anchor = _paths(tmp_path)
    completed = {
        "run_id": "fresh-production",
        "protocol_version": "0.7",
        "status": "COMPLETED",
        "phase": "FINALIZE",
        "max_plan_revisions": 5,
    }
    graph = _Graph({}, completed)
    _Observation.created.clear()
    monkeypatch.setattr(runtime, "create_production_graph", lambda **_k: graph)
    monkeypatch.setattr(runtime, "ProductionObservation", _Observation)
    monkeypatch.setattr(
        runtime,
        "record_run_invocation",
        lambda *_a, **_k: None,
    )

    result = runtime.run_production_graph(
        delivery,
        output_root=tmp_path / "runs",
        thread_id="fresh-production",
        shared_visual_anchor_path=anchor,
        registry_dir=tmp_path / "registry",
    )

    observation = _Observation.created[0]
    assert result == completed
    assert graph.invoke_calls == 1
    assert observation.payload["execution_mode"] == "fresh"
    assert observation.finished == [(completed, True)]