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

from __future__ import annotations

import json
from pathlib import Path
from types import SimpleNamespace

from obagent_sdk import observe
from obagent_sdk.integrations import langgraph

from production_build_agents.observability import (
    ProductionObservation,
    start_graph_node_observation,
)
from production_build_agents.observability import production
from production_build_agents.observability import collector as collector_module
from production_build_agents.production.graph import create_production_graph


class _Handle:
    def __init__(self) -> None:
        self.finished: list[dict] = []
        self.declared: list[dict] = []
        self.outputs: list[tuple[object, dict]] = []
        self.notes: list[tuple[str, dict]] = []

    def finish(self, **payload) -> None:
        self.finished.append(payload)

    def declare(self, **payload) -> None:
        self.declared.append(payload)

    def set_output(self, output, **payload) -> None:
        self.outputs.append((output, payload))

    def record_note(self, text: str, **payload) -> None:
        self.notes.append((text, payload))


class _Manager:
    def __init__(self, handle: _Handle) -> None:
        self.handle = handle
        self.exits: list[tuple] = []

    def __enter__(self) -> _Handle:
        return self.handle

    def __exit__(self, *args):
        self.exits.append(args)
        if args[1] is not None:
            raise args[1]
        return False


def _settings(mode: str = "full"):
    return SimpleNamespace(
        configure_sdk=lambda: True,
        project="production-project",
        project_name="Production 完整观测",
        capture_mode=mode,
    )


def test_production_observer_covers_every_real_graph_node() -> None:
    spec = langgraph.graph_spec(create_production_graph())
    graph_nodes = {node["key"] for node in spec["nodes"]}

    assert graph_nodes == set(production._NODE_SPECS)
    assert len(graph_nodes) == 13


def test_production_planner_titles_distinguish_role_modes(
    tmp_path: Path,
) -> None:
    initial = tmp_path / "initial.json"
    adapt = tmp_path / "adapt.json"
    replan = tmp_path / "replan.json"
    for path, mode in (
        (initial, "INITIAL"),
        (adapt, "ADAPT"),
        (replan, "REPLAN"),
    ):
        path.write_text(json.dumps({"mode": mode}), encoding="utf-8")

    assert production._planning_mode_title(
        "plan_production",
        {"current_planning_package_path": str(initial)},
        "fallback",
    ) == "Production Planner · Initial Plan"
    assert production._planning_mode_title(
        "plan_production",
        {"current_planning_package_path": str(adapt)},
        "fallback",
    ) == "Production Planner · Plan Adaptation"
    assert production._planning_mode_title(
        "plan_production",
        {"current_planning_package_path": str(replan)},
        "fallback",
    ) == "Production Planner · Plan Revision"
    assert production._NODE_SPECS["review_production_progress"] == (
        "agent",
        "Production Planner · Progress Review",
    )
    assert production._NODE_SPECS["adapt_production"][0] == "code"


def test_production_context_captures_root_and_all_run_records(
    monkeypatch,
    tmp_path: Path,
) -> None:
    captured: dict = {}
    collector_calls: list[tuple[str, object]] = []
    run_handle = _Handle()
    run_manager = _Manager(run_handle)
    delivery = tmp_path / "GlobalDataStageDelivery.v1.json"
    anchor = tmp_path / "anchor.png"
    run_dir = tmp_path / "production-run"
    final = run_dir / "final_production_delivery.json"
    delivery.write_text(
        json.dumps({"summary": "完整 Global Data 交付"}, ensure_ascii=False),
        encoding="utf-8",
    )
    anchor.write_bytes(b"\x89PNG")
    final.parent.mkdir()
    final.write_text(
        json.dumps(
            {
                "status": "PASS",
                "video_url": "https://internal.example/final.mp4",
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )

    def fake_run(**kwargs):
        captured.update(kwargs)
        return run_manager

    monkeypatch.setattr(observe, "run", fake_run)
    monkeypatch.setattr(
        collector_module,
        "_start_business_collector",
        lambda handle: (
            collector_calls.append(("start", handle))
            or "collector-token"
        ),
    )
    monkeypatch.setattr(
        collector_module,
        "_stop_business_collector",
        lambda token: collector_calls.append(("stop", token)),
    )
    monkeypatch.setattr(
        langgraph,
        "graph_spec",
        lambda _graph: {"nodes": [{"key": "prepare_inputs"}]},
    )
    result = {
        "run_id": "round-production",
        "protocol_version": "0.7",
        "status": "COMPLETED",
        "phase": "FINALIZE",
        "final_production_delivery_path": str(final),
        "event_log": ["Production 完整完成"],
    }

    with ProductionObservation(
        project_root=tmp_path,
        thread_id="round-production",
        input_sha256="c" * 64,
        protocol_version="0.7",
        execution_mode="fresh",
        graph=object(),
        input_path=delivery,
        shared_visual_anchor_path=anchor,
        run_dir=run_dir,
        parent_run_id="pipeline-run",
        parent_inst_id="production-stage",
        pipeline_round_id="round-1",
        upstream_global_data_thread_id="round-1-global-data",
        settings=_settings(),
    ) as observation:
        observation.finish(result, graph_invoked=True)

    encoded = json.dumps(
        {"run": captured, "finish": run_handle.finished},
        ensure_ascii=False,
    )
    assert captured["agent"] == "production"
    assert captured["auto_collect"] is False
    assert captured["round_anchor"] == {
        "in": "run",
        "on": [
            "prepare_inputs",
            "review_production_progress",
        ],
    }
    assert captured["parent_run_id"] == "pipeline-run"
    assert captured["parent_inst_id"] == "production-stage"
    assert captured["payload"]["Pipeline 轮次"] == "round-1"
    assert captured["payload"]["上游 Global Data Run"] == (
        "round-1-global-data"
    )
    assert collector_calls == [
        ("start", run_handle),
        ("stop", "collector-token"),
    ]
    assert captured["meta"]["capture_mode"] == "full_v1"
    assert captured["payload"]["Global Data 交付摘要"]["业务摘要"] == (
        "完整 Global Data 交付"
    )
    assert captured["payload"]["共享视觉锚点"] == str(anchor.resolve())
    assert captured["meta"]["audit_records"][
        "Global Data 正式交付原件"
    ]["content"]["summary"] == "完整 Global Data 交付"
    assert captured["meta"]["audit_records"]["共享视觉锚点原件"][
        "content_omitted"
    ] == "non_text_record"
    assert "https://internal.example/final.mp4" not in encoded
    assert run_handle.finished[0]["final_output"]["结论"] == "已完成"
    assert "state_after" not in encoded


def test_shared_node_router_uses_production_context_and_segment_branch(
    monkeypatch,
    tmp_path: Path,
) -> None:
    run_handle = _Handle()
    run_manager = _Manager(run_handle)
    node_handle = _Handle()
    node_manager = _Manager(node_handle)
    run_dir = tmp_path / "production-run"
    run_dir.mkdir()
    delivery = tmp_path / "GlobalDataStageDelivery.v1.json"
    anchor = tmp_path / "anchor.png"
    delivery.write_text("{}", encoding="utf-8")
    anchor.write_bytes(b"\x89PNG")
    package = run_dir / "segment_packages" / "Segment2.v3.json"
    delivery_path = run_dir / "segment_deliveries" / "Segment2.v3.json"
    package.parent.mkdir()
    delivery_path.parent.mkdir()
    package.write_text(
        json.dumps(
            {"segment_id": "Segment2", "plan_version": 3},
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )

    monkeypatch.setattr(observe, "run", lambda **_kwargs: run_manager)
    monkeypatch.setattr(
        observe,
        "module",
        lambda *_args, **kwargs: (
            setattr(node_manager, "kwargs", kwargs) or node_manager
        ),
    )
    monkeypatch.setattr(
        langgraph,
        "graph_spec",
        lambda _graph: {"nodes": [], "edges": []},
    )

    with ProductionObservation(
        project_root=tmp_path,
        thread_id="round-production",
        input_sha256="d" * 64,
        protocol_version="0.7",
        execution_mode="fresh",
        graph=object(),
        input_path=delivery,
        shared_visual_anchor_path=anchor,
        run_dir=run_dir,
        settings=_settings(),
    ):
        observation = start_graph_node_observation(
            "execute_segment",
            {
                "run_id": "round-production",
                "output_dir": str(run_dir),
                "status": "RUNNING",
                "current_segment_id": "Segment2",
                "current_segment_package_path": str(package),
                "segment_records": {
                    "Segment2": {
                        "status": "ready",
                        "active_plan_version": 3,
                    }
                },
            },
        )
        delivery_path.write_text(
            json.dumps(
                {
                    "summary": "Segment2 完整 Candidate",
                    "audio_url": "https://internal.example/audio.mp3",
                },
                ensure_ascii=False,
            ),
            encoding="utf-8",
        )
        observation.finish(
            {
                "current_segment_delivery_path": str(delivery_path),
                "status": "RUNNING",
                "phase": "VALIDATE_SEGMENT",
            }
        )

    assert node_manager.kwargs["module_key"] == "execute_segment"
    assert node_manager.kwargs["branch_key"] == "Segment2.v3"
    assert node_manager.kwargs["view"].summary == "结论"
    assert node_manager.kwargs["view"].hide == []
    assert "正式产物" in {
        field.key for field in node_manager.kwargs["view"].fields
    }
    input_blocks = node_handle.declared[0]["blocks"]
    assert [block.key for block in input_blocks] == [
        "execute_segment_identity",
        "segment_package",
    ]
    assert "Segment：Segment2" in input_blocks[1].value
    assert not input_blocks[1].value.startswith("{")
    output, options = node_handle.outputs[0]
    assert "详情" not in output
    assert "Segment2 完整 Candidate" in node_handle.notes[0][0]
    assert "https://internal.example/audio.mp3" in node_handle.notes[0][0]
    assert "state_after" not in output
    assert options["images"] == "auto"