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

from __future__ import annotations

import json
from pathlib import Path

from production_build_agents.observability.redaction import (
    summarize_global_data_delivery,
    summarize_global_data_state,
    summarize_run_metrics,
)


def test_state_summary_omits_raw_content_urls_paths_and_errors() -> None:
    secret_path = "/Users/private/客户标题_production_final.json"
    state = {
        "run_id": "global-data-safe-1",
        "protocol_version": "0.3",
        "input_path": secret_path,
        "input_sha256": "a" * 64,
        "output_dir": "/Users/private/output",
        "status": "FAILED",
        "phase": "VALIDATE_TASK",
        "failure_code": "VALIDATOR_OUTPUT_INVALID",
        "error": "客户正文 https://secret.example/token?sig=secret",
        "plan_history": {"1": {"json_uri": secret_path}},
        "task_records": {
            "Task1": {
                "task_id": "Task1",
                "status": "failed",
                "active_plan_version": 1,
                "executor_delivery_path": secret_path,
                "validation_report_path": secret_path,
            }
        },
        "event_log": ["客户正文"],
        "planner_calls": 1,
    }

    summary = summarize_global_data_state(state)
    encoded = json.dumps(summary, ensure_ascii=False)

    assert summary["error_present"] is True
    assert summary["tasks"][0]["task_id"] == "Task1"
    assert summary["records_present"]["global_data_plan"] is False
    assert "/Users/private" not in encoded
    assert "客户正文" not in encoded
    assert "secret.example" not in encoded
    assert "sig=secret" not in encoded


def test_delivery_summary_keeps_artifact_identity_but_omits_locations(
    tmp_path: Path,
) -> None:
    path = tmp_path / "global_data_stage_delivery.json"
    path.write_text(
        json.dumps(
            {
                "schema_version": "0.3",
                "run_id": "safe-run",
                "plan_id": "safe-plan",
                "final_plan_version": 2,
                "active_artifacts": [
                    {
                        "artifact_id": "Artifact-1",
                        "artifact_type": "image",
                        "mime_type": "image/png",
                        "size_bytes": 123,
                        "content_sha256": "b" * 64,
                        "description": "客户人物和真实文案",
                        "source_uri": "https://secret.example/source.png",
                        "uri": "/private/output.png",
                    }
                ],
                "requirement_evaluations": [
                    {
                        "expectation_evaluations": [
                            {"satisfied": True}
                        ]
                    }
                ],
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )

    summary = summarize_global_data_delivery(str(path))
    encoded = json.dumps(summary, ensure_ascii=False)

    assert summary is not None
    assert summary["artifact_count"] == 1
    assert summary["artifacts"][0]["artifact_id"] == "Artifact-1"
    assert summary["satisfied_requirement_count"] == 1
    assert "客户人物" not in encoded
    assert "secret.example" not in encoded
    assert "/private/output.png" not in encoded


def test_run_metrics_summary_omits_internal_event_ids_and_details(
    tmp_path: Path,
) -> None:
    (tmp_path / "run_metrics.json").write_text(
        json.dumps(
            {
                "model_totals": {
                    "calls": 2,
                    "input_tokens": 10,
                    "output_tokens": 5,
                    "total_tokens": 15,
                    "reported_cost_usd": 0.1,
                    "cost_status": "reported",
                },
                "tools": {
                    "calls": 1,
                    "successful_calls": 1,
                    "failed_calls": 0,
                    "replayed_calls": 0,
                    "total_duration_ms": 20,
                    "by_name": {
                        "inspect_tool": {
                            "calls": 1,
                            "successful_calls": 1,
                            "failed_calls": 0,
                            "replayed_calls": 0,
                            "total_duration_ms": 20,
                        }
                    },
                },
                "phases": {
                    "execute_task": {
                        "calls": 1,
                        "total_duration_ms": 50,
                        "last_duration_ms": 50,
                        "last_outcome": "RUNNING",
                    }
                },
                "corrections": [{"secret": "raw details"}],
                "replans": [{"details": "raw failure text"}],
                "_seen_events": ["private-event-id"],
            }
        ),
        encoding="utf-8",
    )

    summary = summarize_run_metrics(str(tmp_path))
    encoded = json.dumps(summary, ensure_ascii=False)

    assert summary is not None
    assert summary["model_totals"]["total_tokens"] == 15
    assert summary["tools_by_name"]["inspect_tool"]["calls"] == 1
    assert summary["correction_count"] == 1
    assert "raw details" not in encoded
    assert "raw failure text" not in encoded
    assert "private-event-id" not in encoded