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

from __future__ import annotations

import json
from types import SimpleNamespace

from langchain.agents import create_agent
from langchain_core.language_models.fake_chat_models import (
    FakeMessagesListChatModel,
)
from langchain_core.messages import (
    AIMessage,
    HumanMessage,
    SystemMessage,
    ToolMessage,
)

from production_build_agents.observability.readable_events import (
    BusinessReadableFlowCollector,
    readable_value,
    tool_title,
)
from production_build_agents.observability.business_views import (
    business_response_title,
)
from production_build_agents.agents.invocation import build_agent_run_config
from production_build_agents.run.langgraph_checkpointer import (
    create_in_memory_checkpointer,
)


class _ModuleContext:
    def __init__(self) -> None:
        self.events: list[tuple[int, str, dict]] = []
        self.messages: list[tuple[list, int]] = []

    def emit_at(self, idx: int, type_: str, **fields) -> None:
        self.events.append((idx, type_, fields))

    def record_messages(self, messages: list, *, idx: int) -> None:
        self.messages.append((messages, idx))


class _Run:
    def __init__(self) -> None:
        self.usage = (0, 0)

    def account(self, input_tokens: int, output_tokens: int) -> None:
        self.usage = (input_tokens, output_tokens)


def test_readable_value_turns_json_into_business_sentences() -> None:
    text = readable_value(
        '{"success": true, "task_id": "Task2", '
        '"status": "delivered", "output_path": "/run/result.json"}'
    )

    assert "结果：成功" in text
    assert "Task：Task2" in text
    assert "状态：delivered" in text
    assert not text.startswith("{")


def test_fenced_global_data_plan_is_rendered_as_business_logic() -> None:
    plan = {
        "schema_version": "0.3",
        "plan_id": "global-data-plan",
        "plan_version": 1,
        "goal": "准备正式生产所需的共享素材与约束",
        "stage_requirements": [
            {
                "requirement_id": "Requirement1",
                "description": "确认人物基准图真实可用",
                "artifact_expectations": [
                    {
                        "artifact_type": "image",
                        "minimum_count": 1,
                        "source_asset_ids": ["SourceAsset-" + "a" * 64],
                    }
                ],
            }
        ],
        "tasks": [
            {
                "task_id": "Task1",
                "objective": "检查并采纳人物基准图",
                "skill_id": "reference-inspection",
                "expectation_ids": ["Requirement1-Expectation1"],
                "depends_on": [],
            }
        ],
        "revision_summary": "先验收共享素材，再进入正式生产",
    }

    text = readable_value(
        "```json\n"
        + json.dumps(plan, ensure_ascii=False)
        + "\n```"
    )

    assert "规划目标：准备正式生产所需的共享素材与约束" in text
    assert "Requirement1｜确认人物基准图真实可用" in text
    assert "Task1｜检查并采纳人物基准图" in text
    assert "stage_requirements" not in text
    assert "```json" not in text


def test_validator_candidate_without_verdict_has_business_title() -> None:
    candidate = {
        "task_id": "Task3",
        "criterion_results": [
            {
                "criterion_id": "Criterion1",
                "passed": True,
                "reason": "媒体证据完整",
            }
        ],
        "summary": "任务满足验收要求",
    }
    text = readable_value(json.dumps(candidate, ensure_ascii=False))

    assert business_response_title(candidate) == "模型形成 Task 验收结论"
    assert "业务摘要：任务满足验收要求" in text
    assert "criterion_results" not in text


def test_tool_names_have_business_titles() -> None:
    assert tool_title("read_production_brief") == "读取 Production Brief"
    assert tool_title("assemble_segment_media") == "组装 Segment 媒体"
    assert tool_title("custom_business_action") == "Custom Business Action"


def test_llm_tool_decision_card_is_readable_and_keeps_raw_messages() -> None:
    run = _Run()
    context = _ModuleContext()
    collector = BusinessReadableFlowCollector(run, auto_module=False)
    collector._frame = lambda *_args: (context, 3)
    collector._pending_msgs["llm-1"] = [
        AIMessage(content="上一步上下文")
    ]
    collector._model_of["llm-1"] = "test-model"
    message = AIMessage(
        content="",
        tool_calls=[
            {
                "id": "call-1",
                "name": "read_production_brief",
                "args": {"source_paths": ["subject.title", "shots"]},
            }
        ],
        usage_metadata={
            "input_tokens": 20,
            "output_tokens": 5,
            "total_tokens": 25,
        },
    )
    response = SimpleNamespace(
        generations=[[SimpleNamespace(message=message)]],
        llm_output={},
    )

    collector.on_llm_end(response, run_id="llm-1")

    _, type_, event = context.events[0]
    assert type_ == "llm"
    assert event["label"] == "模型决定调用：读取 Production Brief"
    assert "读取 Production Brief" in event["content"]
    assert event["payload"]["tool_calls"][0]["args_summary"] == (
        "来源：subject.title、shots"
    )
    assert context.messages[0][0][-1] is message
    assert run.usage == (20, 5)


def test_llm_start_snapshot_and_terminal_are_joinable_without_raw_content() -> None:
    run = _Run()
    context = _ModuleContext()
    collector = BusinessReadableFlowCollector(run, auto_module=False)
    collector._frame = lambda *_args: (context, 3)
    collector.on_chat_model_start(
        {"name": "test-model"},
        [[
            SystemMessage(content="系统提示词秘密"),
            HumanMessage(content="用户输入秘密"),
        ]],
        run_id="llm-context",
        metadata={
            "business_run_id": "run-1",
            "agent_run_id": "agent-1",
            "domain": "production",
            "role": "segment_executor",
            "segment_id": "Segment1",
            "plan_version": 1,
        },
        invocation_params={"tools": []},
    )
    response_message = AIMessage(
        content="模型输出秘密",
        usage_metadata={
            "input_tokens": 30,
            "output_tokens": 5,
            "total_tokens": 35,
        },
    )
    collector.on_llm_end(
        SimpleNamespace(
            generations=[[SimpleNamespace(message=response_message)]],
            llm_output={},
        ),
        run_id="llm-context",
    )

    assert [event[1] for event in context.events] == ["note", "llm"]
    start = context.events[0][2]["payload"]
    terminal = context.events[1][2]["payload"]["context_metrics"]
    assert start["event_kind"] == "model_context_start"
    assert terminal["event_kind"] == "model_context_terminal"
    assert start["physical_call_id"] == terminal["physical_call_id"]
    assert start["identity"]["segment_id"] == "Segment1"
    assert terminal["usage"]["input_tokens"] == 30
    assert len(context.messages) == 1
    serialized_start = json.dumps(start, ensure_ascii=False)
    assert "系统提示词秘密" not in serialized_start
    assert "用户输入秘密" not in serialized_start


def test_llm_failure_closes_started_context_and_keeps_input_once() -> None:
    context = _ModuleContext()
    collector = BusinessReadableFlowCollector(_Run(), auto_module=False)
    collector._frame = lambda *_args: (context, 4)
    collector.on_chat_model_start(
        {"name": "test-model"},
        [[HumanMessage(content="will fail")]],
        run_id="llm-failure",
        metadata={
            "business_run_id": "run-1",
            "agent_run_id": "agent-1",
            "role": "planner",
        },
    )
    collector.on_llm_error(
        TimeoutError("provider timeout"),
        run_id="llm-failure",
    )

    assert [event[1] for event in context.events] == ["note", "llm"]
    failure = context.events[1][2]
    assert failure["ok"] is False
    terminal = failure["payload"]["context_metrics"]
    assert terminal["status"] == "failure"
    assert terminal["error_type"] == "TimeoutError"
    assert len(context.messages) == 1
    assert len(context.messages[0][0]) == 1


def test_langgraph_config_metadata_reaches_real_model_callback() -> None:
    context = _ModuleContext()
    collector = BusinessReadableFlowCollector(_Run(), auto_module=False)
    collector._frame = lambda *_args: (context, 2)
    model = FakeMessagesListChatModel(
        responses=[AIMessage(content="done")],
        callbacks=[collector],
    )
    agent = create_agent(
        model=model,
        tools=[],
        system_prompt="system",
        checkpointer=create_in_memory_checkpointer(),
    )
    config = build_agent_run_config(
        business_run_id="run-1",
        agent_run_id="run-1:Segment1:v1:validator",
        domain="production",
        role="segment_validator",
        invocation_mode="VALIDATE",
        segment_id="Segment1",
        plan_version=1,
    )

    agent.invoke(
        {"messages": [{"role": "user", "content": "validate"}]},
        config,
    )

    start = next(
        fields["payload"]
        for _, type_, fields in context.events
        if type_ == "note"
    )
    assert start["identity"] == config["metadata"]


def test_llm_plan_card_has_business_title_and_keeps_raw_message() -> None:
    run = _Run()
    context = _ModuleContext()
    collector = BusinessReadableFlowCollector(run, auto_module=False)
    collector._frame = lambda *_args: (context, 4)
    collector._pending_msgs["llm-plan"] = [
        AIMessage(content="规划上下文")
    ]
    plan_json = (
        '```json\n{"schema_version":"0.3","plan_id":"plan-1",'
        '"plan_version":1,"goal":"准备共享素材",'
        '"stage_requirements":[],"tasks":[]}\n```'
    )
    message = AIMessage(content=plan_json)
    response = SimpleNamespace(
        generations=[[SimpleNamespace(message=message)]],
        llm_output={},
    )

    collector.on_llm_end(response, run_id="llm-plan")

    _, type_, event = context.events[0]
    assert type_ == "llm"
    assert event["label"] == "初版候选 · Global Data 规划"
    assert event["content"].startswith("规划目标：准备共享素材")
    assert "```json" not in event["content"]
    assert event["payload"]["business_phase"] == "initial_candidate"
    assert context.messages[0][0][-1] is message


def test_contract_correction_is_labeled_before_upload() -> None:
    run = _Run()
    context = _ModuleContext()
    collector = BusinessReadableFlowCollector(run, auto_module=False)
    collector._frame = lambda *_args: (context, 5)
    collector._pending_msgs["llm-correction"] = [
        HumanMessage(content='{"task_id":"Task3"}'),
        AIMessage(content='{"summary":"初版"}'),
        HumanMessage(
            content=(
                "上一版未通过运行时校验。错误："
                "missing_binding_evidence: 该 Expectation 必须声明工具证据。"
            )
        ),
    ]
    corrected = AIMessage(
        content=json.dumps(
            {
                "task_id": "Task3",
                "artifacts": [{"uri": "/run/image.png"}],
                "artifact_binding_claims": [
                    {
                        "expectation_id": "Expectation1",
                        "evidence_tool_call_ids": ["call-1"],
                    }
                ],
                "summary": "已补充工具证据",
            },
            ensure_ascii=False,
        ),
        usage_metadata={
            "input_tokens": 30,
            "output_tokens": 8,
            "total_tokens": 38,
        },
    )
    response = SimpleNamespace(
        generations=[[SimpleNamespace(message=corrected)]],
        llm_output={},
    )

    collector.on_llm_end(response, run_id="llm-correction")

    _, type_, event = context.events[0]
    assert type_ == "llm"
    assert event["label"] == "合同校正 · 第 1 次"
    assert event["content"].startswith(
        "校正原因：Artifact Binding 缺少工具调用证据\n校正结果："
    )
    assert event["payload"]["business_phase"] == "contract_correction"
    assert event["payload"]["attempt"] == 2
    assert event["payload"]["previous_candidate_ok"] is False
    assert event["payload"]["correction_reason_codes"] == [
        "missing_binding_evidence"
    ]


def test_media_review_is_labeled_and_reports_scope() -> None:
    run = _Run()
    context = _ModuleContext()
    collector = BusinessReadableFlowCollector(run, auto_module=False)
    collector._frame = lambda *_args: (context, 6)
    collector._pending_msgs["llm-media-review"] = [
        HumanMessage(content='{"task_id":"Task3"}'),
        HumanMessage(
            content=[
                {
                    "type": "text",
                    "text": (
                        "最终媒体证据自检："
                        '{"artifact_uris":["/run/image.png","/run/video.mp4"]}'
                    ),
                },
                {
                    "type": "image_url",
                    "image_url": {"url": "data:image/png;base64,AA=="},
                },
            ]
        ),
    ]
    reviewed = AIMessage(
        content=json.dumps(
            {
                "task_id": "Task3",
                "artifacts": [
                    {"uri": "/run/image.png"},
                    {"uri": "/run/video.mp4"},
                ],
                "artifact_binding_claims": [],
                "summary": "媒体内容与制作要求一致",
            },
            ensure_ascii=False,
        )
    )
    response = SimpleNamespace(
        generations=[[SimpleNamespace(message=reviewed)]],
        llm_output={},
    )

    collector.on_llm_end(response, run_id="llm-media-review")

    _, type_, event = context.events[0]
    assert type_ == "llm"
    assert event["label"] == "媒体证据复核"
    assert event["content"].startswith(
        "复核范围：2 个媒体产物\n复核结果："
    )
    assert event["payload"]["business_phase"] == "media_evidence_review"
    assert event["payload"]["reviewed_media_count"] == 2


def test_correction_tool_call_keeps_action_and_adds_phase() -> None:
    run = _Run()
    context = _ModuleContext()
    collector = BusinessReadableFlowCollector(run, auto_module=False)
    collector._frame = lambda *_args: (context, 7)
    collector._pending_msgs["llm-correction-tool"] = [
        HumanMessage(
            content="上一版未通过运行时格式校验。错误：缺少字段。"
        )
    ]
    message = AIMessage(
        content="重新检查图片",
        tool_calls=[
            {
                "id": "call-2",
                "name": "view_images",
                "args": {"image_sources": ["/run/image.png"]},
            }
        ],
    )
    response = SimpleNamespace(
        generations=[[SimpleNamespace(message=message)]],
        llm_output={},
    )

    collector.on_llm_end(response, run_id="llm-correction-tool")

    _, type_, event = context.events[0]
    assert type_ == "llm"
    assert event["label"] == "校正中 · 模型决定调用：查看图片"
    assert event["payload"]["business_phase"] == "contract_correction"
    assert event["payload"]["tool_calls"][0]["name"] == "view_images"


def test_tool_result_card_uses_summary_instead_of_raw_json() -> None:
    context = _ModuleContext()
    collector = BusinessReadableFlowCollector(_Run(), auto_module=False)
    collector._frame = lambda *_args: (context, 2)
    collector._pending_tool["tool-1"] = {
        "name": "probe_media",
        "args": {"artifact_uri": "/run/segment.mp4"},
        "tok": None,
    }
    output = ToolMessage(
        content=(
            '{"success": true, "status": "ready", '
            '"duration_ms": 1250, "video_url": "https://media/segment.mp4"}'
        ),
        tool_call_id="call-1",
        name="probe_media",
    )

    collector.on_tool_end(output, run_id="tool-1")

    _, type_, event = context.events[0]
    assert type_ == "tool"
    assert event["label"] == "检查媒体技术信息 · 已返回"
    assert event["tool_args"] == "Artifact：/run/segment.mp4"
    assert "结果：成功" in event["tool_result"]
    assert "状态：ready" in event["tool_result"]
    assert not event["tool_result"].startswith("{")
    contribution = event["payload"]["context_contribution"]
    assert contribution["message_role"] == "tool"
    assert contribution["serialized_bytes"] > 0
    assert "segment.mp4" not in json.dumps(contribution)