video-image-production-build/tests/agents/test_invocation.py

from pathlib import Path

from langchain_core.messages import AIMessage

from production_build_agents.agents.invocation import (
    build_agent_run_config,
    extract_final_json,
)
from production_build_agents.contracts.models import ExecutorCandidate


def test_extracts_final_candidate_after_schema_fragment() -> None:
    fixture = (
        Path(__file__).parents[1]
        / "fixtures"
        / "v20_executor_output_recovery.txt"
    ).read_text(encoding="utf-8")
    payload = extract_final_json(
        {"messages": [AIMessage(content=fixture)]},
        role="Executor",
        error_type=ValueError,
    )

    assert ExecutorCandidate.model_validate_json(payload).run_id == "fixture-v20"


def test_agent_run_config_exposes_stable_business_identity() -> None:
    config = build_agent_run_config(
        business_run_id="run-1",
        agent_run_id="run-1:Segment1:v2:executor",
        domain="production",
        role="segment_executor",
        invocation_mode="EXECUTE",
        segment_id="Segment1",
        plan_version=2,
        recursion_limit=64,
    )

    assert config == {
        "configurable": {
            "thread_id": "run-1:Segment1:v2:executor",
        },
        "metadata": {
            "business_run_id": "run-1",
            "agent_run_id": "run-1:Segment1:v2:executor",
            "domain": "production",
            "role": "segment_executor",
            "invocation_mode": "EXECUTE",
            "segment_id": "Segment1",
            "plan_version": 2,
        },
        "recursion_limit": 64,
    }