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

from __future__ import annotations

import json

from langchain_core.messages import (
    AIMessage,
    HumanMessage,
    SystemMessage,
    ToolMessage,
)

from production_build_agents.observability.context_metrics import (
    build_context_snapshot,
    build_terminal_metrics,
    summarize_tool_contribution,
)


def test_context_snapshot_measures_messages_tools_images_and_growth() -> None:
    previous = AIMessage(
        content="调用工具",
        tool_calls=[
            {
                "id": "call-1",
                "name": "inspect_tool",
                "args": {"工具参数": "不能进入指标正文"},
            }
        ],
        usage_metadata={
            "input_tokens": 120,
            "output_tokens": 10,
            "total_tokens": 130,
        },
    )
    messages = [
        SystemMessage(content="系统提示词秘密"),
        HumanMessage(
            content=[
                {"type": "text", "text": "中文输入"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "data:image/png;base64,YWJj",
                    },
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://secret.example/image.png",
                    },
                },
            ]
        ),
        previous,
        ToolMessage(
            content='{"secret_result":"不能进入指标正文"}',
            tool_call_id="call-1",
            name="inspect_tool",
        ),
    ]
    snapshot = build_context_snapshot(
        messages,
        invocation_params={
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": "inspect_tool",
                        "description": "工具说明秘密",
                        "parameters": {"type": "object"},
                    },
                }
            ]
        },
        identity={
            "business_run_id": "run-1",
            "agent_run_id": "agent-1",
            "role": "executor",
        },
        physical_call_id="physical-1",
    )

    assert snapshot["logical_call_index"] == 2
    assert snapshot["input"]["message_role_counts"] == {
        "system": 1,
        "human": 1,
        "ai": 1,
        "tool": 1,
        "other": 0,
    }
    assert snapshot["input"]["tool_messages"]["count"] == 1
    assert snapshot["input"]["history_tool_calls"]["count"] == 1
    assert snapshot["input"]["tool_schemas"]["count"] == 1
    assert snapshot["input"]["images"]["count"] == 2
    assert snapshot["input"]["images"]["decoded_bytes"] == 3
    assert snapshot["input"]["images"]["unknown_bytes_count"] == 1
    assert snapshot["previous"]["input_tokens"] == 120
    assert snapshot["previous"]["delta"]["message_count"] == 2

    serialized = json.dumps(snapshot, ensure_ascii=False)
    for secret in (
        "系统提示词秘密",
        "中文输入",
        "不能进入指标正文",
        "https://secret.example/image.png",
        "YWJj",
        "工具说明秘密",
    ):
        assert secret not in serialized


def test_first_call_has_no_previous_delta_and_terminal_uses_real_tokens() -> None:
    snapshot = build_context_snapshot(
        [SystemMessage(content="system"), HumanMessage(content="hello")],
        invocation_params={},
        identity={"agent_run_id": "agent-1"},
        physical_call_id="physical-1",
    )
    output = AIMessage(
        content="done",
        usage_metadata={
            "input_tokens": 25,
            "output_tokens": 5,
            "total_tokens": 30,
            "input_token_details": {"cache_read": 4},
        },
    )
    terminal = build_terminal_metrics(
        snapshot,
        message=output,
        response=object(),
        status="success",
        duration_ms=17,
        transport={
            "observed": False,
            "attempt_count": None,
            "retry_count": None,
            "attempts": [],
        },
    )

    assert snapshot["previous"] is None
    assert terminal["usage"]["input_tokens"] == 25
    assert terminal["usage"]["input_tokens_delta"] is None
    assert terminal["output"]["content_bytes"] == 4
    assert "done" not in json.dumps(terminal)


def test_tool_contribution_only_contains_size_and_hash() -> None:
    contribution = summarize_tool_contribution(
        {"secret": "tool result", "success": True}
    )

    assert contribution["message_role"] == "tool"
    assert contribution["serialized_bytes"] > 0
    assert "tool result" not in json.dumps(contribution)