video-image-production-build/tests/observability/test_global_data_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.global_data import (
    GlobalDataObservation,
    start_graph_node_observation,
)
from production_build_agents.observability import global_data
from production_build_agents.observability import collector as collector_module


class _RunHandle:
    def __init__(self) -> None:
        self.finished: list[dict] = []

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


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

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

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


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

    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))


def test_global_data_context_forces_sanitized_manual_collection(
    monkeypatch,
    tmp_path: Path,
) -> None:
    captured: dict = {}
    handle = _RunHandle()
    manager = _RunManager(handle)

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

    monkeypatch.setattr(observe, "run", fake_run)
    monkeypatch.setattr(
        langgraph,
        "graph_spec",
        lambda _graph: {"nodes": [{"key": "preprocess"}], "edges": []},
    )
    settings = SimpleNamespace(
        configure_sdk=lambda: True,
        project="safe-project",
        project_name="安全项目",
        capture_mode="redacted",
    )
    result = {
        "run_id": "safe-run",
        "protocol_version": "0.3",
        "status": "COMPLETED",
        "phase": "FINALIZE",
        "input_path": "/private/客户标题.json",
        "output_dir": str(tmp_path),
        "event_log": ["客户正文"],
    }

    with GlobalDataObservation(
        project_root=tmp_path,
        thread_id="safe-run",
        input_sha256="a" * 64,
        protocol_version="0.3",
        execution_mode="fresh",
        graph=object(),
        settings=settings,
    ) as observation:
        observation.finish(result, graph_invoked=True)

    encoded = json.dumps(
        {"run": captured, "finish": handle.finished},
        ensure_ascii=False,
    )
    assert captured["auto_collect"] is False
    assert captured["round_anchor"] == {
        "in": "run",
        "on": ["prepare_next_task"],
    }
    assert captured["payload"]["input_sha256"] == "a" * 64
    assert captured["meta"]["capture_mode"] == "redacted_allowlist_v1"
    assert handle.finished[0]["ok"] is True
    assert "/private/客户标题.json" not in encoded
    assert "客户正文" not in encoded


def test_global_data_context_full_capture_reports_real_records(
    monkeypatch,
    tmp_path: Path,
) -> None:
    captured: dict = {}
    collector_calls: list[tuple[str, object]] = []
    handle = _RunHandle()
    manager = _RunManager(handle)
    input_path = tmp_path / "production_final.json"
    delivery_path = tmp_path / "GlobalDataStageDelivery.v1.json"
    input_path.write_text(
        json.dumps(
            {
                "title": "客户真实标题",
                "reference_url": "https://internal.example/input.png",
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    delivery_path.write_text(
        json.dumps(
            {
                "summary": "完整交付正文",
                "media_url": "https://internal.example/output.mp4",
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )

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

    monkeypatch.setattr(observe, "run", fake_run)
    monkeypatch.setattr(
        collector_module,
        "_start_business_collector",
        lambda run_handle: (
            collector_calls.append(("start", run_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": "preprocess"}], "edges": []},
    )
    settings = SimpleNamespace(
        configure_sdk=lambda: True,
        project="full-project",
        project_name="完整观测项目",
        capture_mode="full",
    )
    result = {
        "run_id": "full-run",
        "protocol_version": "0.3",
        "status": "COMPLETED",
        "phase": "FINALIZE",
        "input_path": str(input_path),
        "output_dir": str(tmp_path),
        "global_data_delivery_path": str(delivery_path),
        "event_log": ["完整事件正文"],
    }

    with GlobalDataObservation(
        project_root=tmp_path,
        thread_id="full-run",
        input_sha256="b" * 64,
        protocol_version="0.3",
        execution_mode="fresh",
        graph=object(),
        input_path=input_path,
        parent_run_id="pipeline-run",
        parent_inst_id="global-stage",
        pipeline_round_id="round-1",
        settings=settings,
    ) as observation:
        observation.finish(result, graph_invoked=True)

    encoded = json.dumps(
        {"run": captured, "finish": handle.finished},
        ensure_ascii=False,
    )
    assert captured["auto_collect"] is False
    assert captured["round_anchor"] == {
        "in": "run",
        "on": ["prepare_next_task"],
    }
    assert captured["parent_run_id"] == "pipeline-run"
    assert captured["parent_inst_id"] == "global-stage"
    assert captured["payload"]["pipeline_round_id"] == "round-1"
    assert collector_calls == [
        ("start", handle),
        ("stop", "collector-token"),
    ]
    assert captured["meta"]["capture_mode"] == "full_v1"
    assert captured["payload"]["input"]["content"]["title"] == (
        "客户真实标题"
    )
    assert handle.finished[0]["ok"] is True
    assert "客户真实标题" in encoded
    assert "完整交付正文" not in encoded
    assert "https://internal.example/output.mp4" not in encoded
    assert handle.finished[0]["final_output"]["结论"] == "已完成"
    assert "state_after" not in encoded


def test_full_node_capture_expands_records_and_enables_media_scan(
    monkeypatch,
    tmp_path: Path,
) -> None:
    task_path = tmp_path / "Task1.TaskPackage.v1.json"
    delivery_path = tmp_path / "Task1.ExecutorDelivery.v1.json"
    task_path.write_text(
        json.dumps(
            {"task_id": "Task1", "plan_version": 1},
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    delivery_path.write_text(
        json.dumps(
            {
                "text": "完整工具交付",
                "image_url": "https://internal.example/generated.png",
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    handle = _ModuleHandle()
    manager = _RunManager(handle)
    monkeypatch.setattr(observe, "module", lambda *_args, **_kwargs: manager)
    owner = SimpleNamespace(
        active=True,
        full_capture=True,
        warn_once=lambda *_args, **_kwargs: None,
    )
    token = global_data._ACTIVE.set(owner)
    try:
        observation = start_graph_node_observation(
            "execute_task",
            {
                "run_id": "full-run",
                "current_task_path": str(task_path),
            },
        )
        observation.finish(
            {
                "current_executor_delivery_path": str(delivery_path),
                "status": "RUNNING",
                "phase": "VALIDATE_TASK",
                "event_log": ["Executor 完整事件"],
            }
        )
    finally:
        global_data._ACTIVE.reset(token)

    input_blocks = handle.declared[0]["blocks"]
    assert [block.key for block in input_blocks] == [
        "execute_task_identity",
        "task_package",
    ]
    assert "Task：Task1" in input_blocks[1].value
    assert not input_blocks[1].value.startswith("{")
    assert "详情" not in handle.outputs[0][0]
    assert "完整工具交付" in handle.notes[0][0]
    assert handle.notes[0][1]["label"] == "正式记录详情"
    assert handle.outputs[0][0]["执行结果"] == "Executor 完整事件"
    assert "state_after" not in handle.outputs[0][0]
    assert handle.outputs[0][1]["images"] == "auto"