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

from __future__ import annotations

import json
from pathlib import Path

from production_build_agents.observability.production_capture import (
    collect_production_run_records,
    expand_production_node_output,
    summarize_production_state,
)


def test_production_run_record_scan_is_incremental(
    tmp_path: Path,
) -> None:
    run_dir = tmp_path / "production-run"
    candidate = run_dir / "segment_candidates" / "Segment1.v1.json"
    candidate.parent.mkdir(parents=True)
    candidate.write_text(
        json.dumps(
            {
                "summary": "完整 Candidate",
                "image_url": "https://internal.example/segment.png",
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    (run_dir / "video.mp4").write_bytes(b"\x00\x01")
    seen: dict[str, str] = {}

    first = collect_production_run_records(
        run_dir,
        seen_hashes=seen,
        changed_only=True,
    )
    second = collect_production_run_records(
        run_dir,
        seen_hashes=seen,
        changed_only=True,
    )
    candidate.write_text(
        json.dumps(
            {"summary": "更新后的完整 Candidate"},
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    third = collect_production_run_records(
        run_dir,
        seen_hashes=seen,
        changed_only=True,
    )

    key = "segment_candidates/Segment1.v1.json"
    assert first[key]["content"]["summary"] == "完整 Candidate"
    assert "video.mp4" not in first
    assert second == {}
    assert third[key]["content"]["summary"] == "更新后的完整 Candidate"


def test_production_node_output_keeps_segment_and_error_details(
    tmp_path: Path,
) -> None:
    report = tmp_path / "Segment1.v1.json"
    report.write_text(
        json.dumps(
            {"verdict": "FAIL", "reason": "真实验收原因"},
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    state = {
        "run_id": "round-production",
        "status": "RUNNING",
        "current_segment_id": "Segment1",
    }
    update = {
        "status": "FAILED",
        "error": "真实 Production 错误",
        "current_segment_validation_report_path": str(report),
    }

    observed = expand_production_node_output(
        "validate_segment",
        state,
        update,
        run_records={},
        exception=RuntimeError("工具结果未知"),
    )

    assert observed["state_after"]["error"] == "真实 Production 错误"
    assert observed["exception"]["message"] == "工具结果未知"
    assert observed["referenced_records"][
        "current_segment_validation_report_path"
    ]["content"]["reason"] == "真实验收原因"


def test_redacted_production_state_preserves_scheduling_facts() -> None:
    summary = summarize_production_state(
        {
            "run_id": "round-production",
            "protocol_version": "0.7",
            "status": "RUNNING",
            "phase": "EXECUTE_SEGMENT",
            "current_segment_id": "Segment2",
            "executor_calls": 3,
            "event_log": ["正文不会进入摘要"],
            "segment_records": {
                "Segment2": {
                    "status": "ready",
                    "active_plan_version": 2,
                    "delivery_path": "/private/Delivery.json",
                }
            },
        }
    )

    assert summary["phase"] == "EXECUTE_SEGMENT"
    assert summary["counters"]["executor_calls"] == 3
    assert summary["segments"]["Segment2"]["status"] == "ready"
    assert summary["event_count"] == 1
    assert "/private/Delivery.json" not in json.dumps(summary)