SupplyAgent/tests/supply_agent/test_find_agent_v2.py

from __future__ import annotations

import json
from dataclasses import replace
from pathlib import Path
from types import SimpleNamespace

import pytest
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
from langchain_core.messages import AIMessage, ToolMessage

from find_agent_v2.graph import FindAgentRoundGraph
from find_agent_v2 import tools as find_agent_tools
from find_agent_v2 import runtime as find_agent_runtime
from find_agent_v2.agent import decide_continued_exploration
from find_agent_v2.runtime import (
    DelegateArgs,
    FindAgentNodeHost,
    _events,
    _message_dict,
    _usage,
)
from find_agent_v2.demand_context import (
    V2DemandContext,
    V2ReferencePoint,
    V2ReferenceVideo,
    _latest_demand_grade_query,
    _points_from_expansions,
    build_v2_user_input,
)
from find_agent_v2.gates import build_rule_snapshot, evaluate_candidate_gate
from find_agent_v2.models import (
    FindAgentV2Candidate,
    FindAgentV2Evidence,
    FindAgentV2Round,
    FindAgentV2Run,
    FindAgentV2Search,
)
from find_agent_v2.observability import (
    GRAPH_SPEC,
    MODULE_TITLES,
    OBAGENT_AGENT,
    OBAGENT_PROJECT,
    OBAGENT_ROUND_ANCHOR,
    NullObserver,
    InputSlot,
    _ModuleHandle,
)
from find_agent_v2.prompts import COMMON_RULES, EVALUATOR_PROMPT
from find_agent_v2.providers import _normalize_search_item, normalize_age_pair
from find_agent_v2.state import (
    DemandBrief,
    DiscoverySnapshot,
    EvaluationBrief,
    EvidenceAssignment,
    ExecutionPlan,
    FindAgentState,
    NodeRun,
    PlanningDecision,
    SearchTask,
    SearchAssignment,
    SupervisorDecision,
)
from find_agent_v2.service import _fails_search_share_gate
from find_agent_v2.tools import (
    CandidateEvaluation,
    EVALUATION_TOOLS,
    EVIDENCE_TOOLS,
    REPORT_TOOLS,
    SEARCH_TOOLS,
    bound_candidate_tools,
    normalize_evaluation_items,
)


def _names(functions) -> set[str]:
    return {getattr(fn, "_tool_name", fn.__name__) for fn in functions}


def test_video_tool_error_code_is_observed_as_failed() -> None:
    message = ToolMessage(
        content=json.dumps({
            "success": False,
            "error": "OSS timeout",
            "error_code": "video_understanding_timeout",
        }),
        name="understand_candidate_video_30s_v2",
        tool_call_id="call-1",
    )

    assert _message_dict(message)["status"] == "error"
    assert _events([message])[0]["status"] == "error"
    assert _events([message])[0]["error_code"] == "video_understanding_timeout"


def test_runtime_usage_reads_openrouter_token_usage_cost() -> None:
    message = AIMessage(
        content="done",
        usage_metadata={"input_tokens": 10, "output_tokens": 2, "total_tokens": 12},
        response_metadata={"token_usage": {"cost": 0.00123}},
    )

    assert _usage([message]) == {
        "input_tokens": 10,
        "output_tokens": 2,
        "total_tokens": 12,
        "cost": 0.00123,
    }


@pytest.mark.asyncio
async def test_host_direct_search_and_evidence_paths_use_no_llm(monkeypatch) -> None:
    calls: list[tuple[str, object]] = []

    async def fake_search(**kwargs):
        calls.append(("search", kwargs["searches"]))
        return '{"run_id":"run","searches":[]}'

    async def fake_details(**kwargs):
        calls.append(("detail", kwargs["candidate_ids"]))
        return '{"run_id":"run","success_count":2,"failed_count":0}'

    monkeypatch.setattr(find_agent_runtime, "search_videos_v2", fake_search)
    monkeypatch.setattr(find_agent_runtime, "fetch_candidate_details_v2", fake_details)
    host = FindAgentNodeHost(observer=NullObserver())

    search_run = await host.run_search_assignment(
        round_index=1,
        assignment=SearchAssignment(
            run_id="run",
            round_index=1,
            tasks=[SearchTask(
                task_id="task-1", keyword="日本间谍", query_reason="测试",
            )],
        ),
        system_prompt="search",
        user_content="{}",
    )
    evidence_run = await host.run_evidence_assignment(
        round_index=1,
        assignment=EvidenceAssignment(
            run_id="run",
            round_index=1,
            candidate_ids=[101, 102],
            evidence_type="detail",
            candidates=[],
        ),
        system_prompt="evidence",
        user_content="{}",
    )

    assert calls == [
        ("search", [{
            "task_id": "task-1",
            "keyword": "日本间谍",
            "query_reason": "测试",
            "source_type": "mixed",
            "provider": None,
            "max_pages": 1,
            "coverage_targets": [],
        }]),
        ("detail", [101, 102]),
    ]
    assert (search_run.iterations, evidence_run.iterations) == (0, 0)
    assert (search_run.tool_calls_made, evidence_run.tool_calls_made) == (1, 1)


def _evaluation(**overrides):
    value = {
        "candidate_id": 11,
        "relevance_score": 0.9,
        "elder_score": 0.8,
        "share_score": 0.7,
        "value_score": 0.85,
        "decision_bucket": "primary",
        "decision_reason": "证据充分且满足推荐要求",
    }
    value.update(overrides)
    return value


def test_evaluation_items_resolve_aweme_id_only_inside_worker_shard() -> None:
    allowed = [
        {"candidate_id": 11, "aweme_id": "video-11"},
        {"candidate_id": 12, "aweme_id": "video-12"},
    ]
    normalized = normalize_evaluation_items([
        _evaluation(candidate_id=None, aweme_id="video-11"),
        _evaluation(candidate_id=12, aweme_id=None, decision_bucket="rejected"),
    ], allowed_candidates=allowed)

    assert [item["candidate_id"] for item in normalized] == [11, 12]
    assert all("aweme_id" not in item for item in normalized)


@pytest.mark.parametrize("items, message", [
    ([_evaluation()], "缺少 candidate_ids=[12]"),
    ([_evaluation(), _evaluation()], "不能重复评估"),
    ([_evaluation(candidate_id=99), _evaluation(candidate_id=12)], "不属于当前 Worker 分片"),
])
def test_evaluation_items_fail_fast_on_incomplete_duplicate_or_foreign_items(items, message) -> None:
    allowed = [
        {"candidate_id": 11, "aweme_id": "video-11"},
        {"candidate_id": 12, "aweme_id": "video-12"},
    ]
    with pytest.raises(ValueError, match=message.replace("[", r"\[").replace("]", r"\]")):
        normalize_evaluation_items(items, allowed_candidates=allowed)


def test_candidate_evaluation_schema_rejects_missing_identity() -> None:
    payload = _evaluation(candidate_id=None)
    with pytest.raises(ValueError, match="至少提供一个"):
        CandidateEvaluation.model_validate(payload)


def test_v2_orm_uses_only_new_table_namespace() -> None:
    assert {
        FindAgentV2Run.__tablename__,
        FindAgentV2Round.__tablename__,
        FindAgentV2Search.__tablename__,
        FindAgentV2Candidate.__tablename__,
        FindAgentV2Evidence.__tablename__,
    } == {
        "find_agent_v2_run",
        "find_agent_v2_round",
        "find_agent_v2_search",
        "find_agent_v2_candidate",
        "find_agent_v2_evidence",
    }
    assert "obagent_run_uid" in FindAgentV2Run.__table__.columns
    assert {"input_tokens", "output_tokens", "total_tokens", "cost_usd"} <= {
        column.name for column in FindAgentV2Run.__table__.columns
    }
    assert not any(table.foreign_key_constraints for table in (
        FindAgentV2Run.__table__,
        FindAgentV2Round.__table__,
        FindAgentV2Search.__table__,
        FindAgentV2Candidate.__table__,
        FindAgentV2Evidence.__table__,
    ))


def test_v2_package_has_no_legacy_business_imports() -> None:
    package = Path(__file__).parents[2] / "find_agent_v2"
    source = "\n".join(path.read_text(encoding="utf-8") for path in package.glob("*.py"))
    assert "agents.find_agent" not in source
    assert "video_discovery_gates" not in source
    assert "services.video_discovery" not in source
    assert "system_prompt.md" not in source


def test_project_orm_defines_no_database_foreign_keys() -> None:
    import supply_infra.db.models  # noqa: F401
    from supply_infra.db.base import Base

    assert not {
        table.name: sorted(fk.name or "<unnamed>" for fk in table.foreign_key_constraints)
        for table in Base.metadata.tables.values()
        if table.foreign_key_constraints
    }


def test_v2_owns_age_portrait_normalization() -> None:
    normalized = normalize_age_pair(
        {"年龄": {"50-": {"percentage": "35%", "preference": 120}}},
        {"年龄": {"50岁以上": {"percentage": 0.25, "preference": 110}}},
    )
    assert normalized["content"]["older_ratio"] == 0.35
    assert normalized["account"]["older_ratio"] == 0.25
    assert normalized["consistency"] == "aligned"


def test_search_metrics_preserve_missing_share_count_and_real_zero() -> None:
    missing = _normalize_search_item({"aweme_id": "missing", "statistics": {}})
    zero = _normalize_search_item({
        "aweme_id": "zero", "statistics": {"share_count": 0},
    })

    assert missing is not None and missing["statistics"]["share_count"] is None
    assert zero is not None and zero["statistics"]["share_count"] == 0


@pytest.mark.parametrize(
    "provider,share_count,expected",
    [
        ("tikhub", 999, True),
        ("tikhub", 1000, False),
        ("tikhub", None, False),
        ("internal_keyword", 0, True),
        ("internal_keyword", 999, True),
        ("internal_keyword", 1000, False),
        ("internal_keyword", None, False),
    ],
)
def test_search_share_gate_applies_to_explicit_metrics_from_all_providers(
    provider, share_count, expected,
) -> None:
    assert _fails_search_share_gate(provider, share_count, 1000) is expected


def test_v2_owns_primary_candidate_gate() -> None:
    rules = build_rule_snapshot()
    candidate = {
        "title": "适合父母的实用生活建议",
        "publish_at": rules["current_datetime"],
        "duration_seconds": 60,
        "share_count": 2000,
        "content_50_plus_ratio": 0.35,
        "account_50_plus_ratio": 0.25,
        "relevance_score": 0.8,
        "elder_score": 0.8,
        "share_score": 0.8,
        "value_score": 0.8,
    }
    result = evaluate_candidate_gate(candidate, rules)
    assert result["status"] == "pass"
    assert result["primary_eligible"] is True


def test_v2_gate_rejects_explicitly_low_evidence() -> None:
    rules = build_rule_snapshot()
    candidate = {
        "title": "普通视频",
        "publish_at": rules["current_datetime"],
        "duration_seconds": 10,
        "share_count": 3,
        "content_50_plus_ratio": 0.01,
        "account_50_plus_ratio": 0.02,
    }
    result = evaluate_candidate_gate(candidate, rules)
    assert result["status"] == "fail"
    assert {"DURATION_TOO_SHORT", "SHARE_COUNT_TOO_LOW", "PORTRAIT_50_PLUS_TOO_LOW"} <= set(
        result["failed_reason_codes"]
    )


def test_v2_gate_ignores_publish_time_but_rejects_missing_elder_portrait() -> None:
    rules = build_rule_snapshot()
    candidate = {
        "title": "高分但缺少硬性证据的视频",
        "publish_at": None,
        "duration_seconds": 60,
        "share_count": 2000,
        "content_50_plus_ratio": None,
        "account_50_plus_ratio": None,
        "relevance_score": 0.99,
        "elder_score": 0.99,
        "share_score": 0.99,
        "value_score": 0.99,
    }

    result = evaluate_candidate_gate(candidate, rules)

    assert result["status"] == "fail"
    assert result["failed_reason_codes"] == ["CONTENT_PORTRAIT_MISSING"]
    assert all(check["name"] != "temporal" for check in result["checks"])


@pytest.mark.parametrize(
    "content_ratio,account_ratio",
    [(0.30, None), (None, 0.30), (0.05, 0.30), (0.30, 0.05)],
)
def test_v2_gate_accepts_either_elder_portrait_side(content_ratio, account_ratio) -> None:
    rules = build_rule_snapshot()
    result = evaluate_candidate_gate({
        "title": "符合要求的视频",
        "publish_at": rules["current_datetime"],
        "duration_seconds": 60,
        "share_count": 2000,
        "content_50_plus_ratio": content_ratio,
        "account_50_plus_ratio": account_ratio,
    }, rules)

    assert result["status"] == "pass"
    assert result["primary_eligible"] is True


def test_v2_gate_does_not_compensate_missing_duration_or_share_count() -> None:
    rules = build_rule_snapshot()
    result = evaluate_candidate_gate({
        "title": "其他信号很强但硬指标缺失",
        "publish_at": rules["current_datetime"],
        "duration_seconds": None,
        "share_count": None,
        "content_50_plus_ratio": 0.30,
        "like_count": 100000,
        "play_count": 1000000,
        "relevance_score": 0.99,
        "elder_score": 0.99,
        "share_score": 0.99,
        "value_score": 0.99,
    }, rules)

    assert result["status"] == "fail"
    assert {"DURATION_UNKNOWN", "SHARE_COUNT_UNKNOWN"} <= set(
        result["failed_reason_codes"]
    )


def test_v2_context_deduplicates_expansion_points() -> None:
    rows = [
        SimpleNamespace(
            video_id="v1", point_type="purpose", expanded_text="照顾父母",
            point_desc="描述一",
        ),
        SimpleNamespace(
            video_id="v1", point_type="purpose", expanded_text="照顾父母",
            point_desc="重复描述",
        ),
        SimpleNamespace(
            video_id="v1", point_type="invalid", expanded_text="忽略",
            point_desc=None,
        ),
    ]
    order, points = _points_from_expansions(rows)
    assert order == ["v1"]
    assert len(points["v1"]) == 1
    assert points["v1"][0].point == "照顾父母"


def test_v2_context_builds_self_contained_user_input() -> None:
    context = V2DemandContext(
        biz_dt="20260811",
        demand_grade_id=123,
        demand_name="测试需求",
        grade="S",
        score=95.0,
        videos=[V2ReferenceVideo(
            video_id="video-1",
            title="参考视频",
            points=[V2ReferencePoint("关键内容", "key", "关键描述")],
        )],
    )
    raw = build_v2_user_input(
        context,
        run_id="v2-test-run",
        rules={"rule_version": "v2-test"},
    )
    assert '"run_id": "v2-test-run"' in raw
    assert '"demand_grade_id": 123' in raw
    assert '"reference_videos"' in raw
    assert '"关键内容"' in raw


def test_demand_word_lookup_uses_exact_name_and_newest_record_order() -> None:
    sql = str(
        _latest_demand_grade_query("照顾父母").compile(
            compile_kwargs={"literal_binds": True},
        )
    ).lower()
    assert "demand_grade.demand_name = '照顾父母'" in sql
    assert "order by demand_grade.biz_dt desc, demand_grade.create_time desc, demand_grade.id desc" in sql


def test_obagent_identity_and_round_structure_are_stable() -> None:
    assert OBAGENT_PROJECT == "find_agent_v2"
    assert OBAGENT_AGENT == "find_agent_v2"
    assert OBAGENT_ROUND_ANCHOR == {"in": "run", "on": ["graph"]}
    assert [node["key"] for node in GRAPH_SPEC["nodes"]] == [
        "supervisor", "search", "evidence", "evaluator",
    ]
    assert set(MODULE_TITLES) == {"supervisor", "search", "evidence", "evaluator", "report"}


def test_round_graph_is_a_compiled_langgraph() -> None:
    service = _FakeService(pending_after_search=0)
    graph = FindAgentRoundGraph(service=service, runner=_FakeRunner(service))
    drawable = graph.app.get_graph()
    assert {"supervisor", "search", "evidence", "evaluator"} <= set(drawable.nodes)
    assert graph.obagent_spec.get("nodes")


def test_runtime_exposes_bounded_delegate_schema() -> None:
    field = DelegateArgs.model_fields["requests"]
    assert field.metadata
    assert FindAgentNodeHost.__module__ == "find_agent_v2.runtime"


def test_runtime_usage_can_be_reset_between_resume_attempts() -> None:
    host = FindAgentNodeHost(observer=NullObserver())
    host.usage["total_tokens"] = 99
    host.reset_usage()
    assert host.usage == {
        "input_tokens": 0,
        "output_tokens": 0,
        "total_tokens": 0,
        "cost": 0.0,
    }


@pytest.mark.asyncio
async def test_langchain_runtime_runs_without_network(monkeypatch) -> None:
    host = FindAgentNodeHost(observer=NullObserver())
    fake_model = FakeMessagesListChatModel(responses=[AIMessage(content="done")])
    monkeypatch.setattr(host, "_model", lambda _role: fake_model)

    result = await host.run_node(
        node="supervisor",
        round_index=1,
        system_prompt="plan",
        user_content="task",
        tools=(),
        max_iterations=2,
        allow_delegation=False,
    )

    assert result.content == "done"
    assert result.iterations == 1
    assert result.tool_calls_made == 0


class _FakeSearchService:
    def __init__(self) -> None:
        self.saved: list[dict] = []

    def save_search(self, **kwargs):
        self.saved.append(kwargs)
        results = list((kwargs.get("payload") or {}).get("search_results") or [])
        return {
            "search_id": len(self.saved),
            "new_candidate_count": len(results),
            "result_count": len(results),
            "share_gate_rejected_count": 0,
        }


def _empty_search_payload(*, provider: str, error: str | None = None) -> dict:
    return {
        "provider": provider,
        "search_results": [],
        "has_more": False,
        "next_cursor": "",
        "error": error,
    }


def _hit_search_payload(*, provider: str, aweme_id: str) -> dict:
    return {
        "provider": provider,
        "search_results": [{"aweme_id": aweme_id, "desc": "命中"}],
        "has_more": False,
        "next_cursor": "",
    }


@pytest.mark.asyncio
async def test_default_search_falls_back_to_tikhub_when_internal_is_empty(
    monkeypatch,
) -> None:
    service = _FakeSearchService()
    providers: list[str] = []

    async def fake_internal(**_kwargs):
        providers.append("internal_keyword")
        return _empty_search_payload(provider="internal_keyword")

    async def fake_tikhub(**_kwargs):
        providers.append("tikhub")
        return _hit_search_payload(provider="tikhub", aweme_id="1")

    monkeypatch.setattr(find_agent_tools, "get_find_agent_v2_service", lambda: service)
    monkeypatch.setattr(find_agent_tools, "search_internal", fake_internal)
    monkeypatch.setattr(find_agent_tools, "search_tikhub", fake_tikhub)

    raw = await find_agent_tools.search_videos_v2(
        run_id="run",
        round_index=1,
        searches=[{
            "keyword": "甲午战争 民族觉醒 历史真相",
            "query_reason": "扩大候选池",
        }],
    )
    payload = json.loads(raw)

    assert providers == ["internal_keyword", "tikhub"]
    assert [item["provider"] for item in payload["searches"]] == [
        "internal_keyword",
        "tikhub",
    ]
    assert payload["searches"][1]["fallback_from"] == "internal_keyword"
    assert [item["provider"] for item in service.saved] == [
        "internal_keyword",
        "tikhub",
    ]


@pytest.mark.asyncio
async def test_default_search_skips_tikhub_when_internal_has_results(
    monkeypatch,
) -> None:
    service = _FakeSearchService()
    providers: list[str] = []

    async def fake_internal(**_kwargs):
        providers.append("internal_keyword")
        return _hit_search_payload(provider="internal_keyword", aweme_id="2")

    async def fake_tikhub(**_kwargs):
        providers.append("tikhub")
        return _hit_search_payload(provider="tikhub", aweme_id="3")

    monkeypatch.setattr(find_agent_tools, "get_find_agent_v2_service", lambda: service)
    monkeypatch.setattr(find_agent_tools, "search_internal", fake_internal)
    monkeypatch.setattr(find_agent_tools, "search_tikhub", fake_tikhub)

    raw = await find_agent_tools.search_videos_v2(
        run_id="run",
        round_index=1,
        searches=[{
            "keyword": "国家安全 反间谍",
            "query_reason": "未指定 provider 且内部有结果",
        }],
    )
    payload = json.loads(raw)

    assert providers == ["internal_keyword"]
    assert [item["provider"] for item in payload["searches"]] == ["internal_keyword"]
    assert "fallback_from" not in payload["searches"][0]


@pytest.mark.asyncio
async def test_explicit_internal_search_does_not_fallback_when_empty(
    monkeypatch,
) -> None:
    service = _FakeSearchService()
    providers: list[str] = []

    async def fake_internal(**_kwargs):
        providers.append("internal_keyword")
        return _empty_search_payload(provider="internal_keyword")

    async def fake_tikhub(**_kwargs):
        providers.append("tikhub")
        return _hit_search_payload(provider="tikhub", aweme_id="3")

    monkeypatch.setattr(find_agent_tools, "get_find_agent_v2_service", lambda: service)
    monkeypatch.setattr(find_agent_tools, "search_internal", fake_internal)
    monkeypatch.setattr(find_agent_tools, "search_tikhub", fake_tikhub)

    raw = await find_agent_tools.search_videos_v2(
        run_id="run",
        round_index=1,
        searches=[{
            "keyword": "国家安全 反间谍",
            "query_reason": "指定只用内部搜索",
            "provider": "internal_keyword",
        }],
    )
    payload = json.loads(raw)

    assert providers == ["internal_keyword"]
    assert [item["provider"] for item in payload["searches"]] == ["internal_keyword"]
    assert "fallback_from" not in payload["searches"][0]


@pytest.mark.asyncio
async def test_explicit_tikhub_search_does_not_call_internal(monkeypatch) -> None:
    service = _FakeSearchService()
    providers: list[str] = []

    async def fake_internal(**_kwargs):
        providers.append("internal_keyword")
        return _hit_search_payload(provider="internal_keyword", aweme_id="4")

    async def fake_tikhub(**_kwargs):
        providers.append("tikhub")
        return _hit_search_payload(provider="tikhub", aweme_id="5")

    monkeypatch.setattr(find_agent_tools, "get_find_agent_v2_service", lambda: service)
    monkeypatch.setattr(find_agent_tools, "search_internal", fake_internal)
    monkeypatch.setattr(find_agent_tools, "search_tikhub", fake_tikhub)

    raw = await find_agent_tools.search_videos_v2(
        run_id="run",
        round_index=1,
        searches=[{
            "keyword": "抗日战争 历史档案",
            "query_reason": "指定 TikHub",
            "provider": "tikhub",
        }],
    )
    payload = json.loads(raw)

    assert providers == ["tikhub"]
    assert [item["provider"] for item in payload["searches"]] == ["tikhub"]


def test_stage_tool_allowlists_are_physical_and_isolated() -> None:
    assert _names(SEARCH_TOOLS) == {"search_videos_v2", "query_find_agent_v2_state"}
    assert _names(EVIDENCE_TOOLS) == {
        "fetch_candidate_details_v2",
        "fetch_candidate_portraits_v2",
        "query_pending_candidates_v2",
    }
    assert _names(EVALUATION_TOOLS) == {
        "understand_candidate_video_30s_v2",
        "evaluate_candidates_v2",
        "query_pending_candidates_v2",
    }
    assert _names(REPORT_TOOLS) == {"query_find_agent_v2_state"}
    all_names = _names((*SEARCH_TOOLS, *EVIDENCE_TOOLS, *EVALUATION_TOOLS, *REPORT_TOOLS))
    assert not any(name.startswith("batch_update_video_discovery") for name in all_names)
    assert "query_video_discovery_state" not in all_names


def test_bound_worker_tools_query_exact_ids_beyond_global_display_limit(monkeypatch) -> None:
    class Service:
        def require_run(self, run_id):
            return {"run_id": run_id, "status": "running"}

        def candidate_inputs(self, run_id, candidate_ids):
            assert run_id == "large-run"
            assert candidate_ids == [101, 102]
            return [
                {
                    "candidate_id": candidate_id,
                    "aweme_id": f"video-{candidate_id}",
                    "decision_bucket": "pending_evaluation",
                }
                for candidate_id in candidate_ids
            ]

        def get_full_state(self, *_args, **_kwargs):
            raise AssertionError("Worker 不应通过全局展示投影查询自己的分片")

        def evaluate(self, run_id, items):
            assert run_id == "large-run"
            return [
                {"candidate_id": item["candidate_id"], "decision_bucket": "primary"}
                for item in items
            ]

    monkeypatch.setattr(find_agent_tools, "get_find_agent_v2_service", lambda: Service())
    query_tool = bound_candidate_tools(
        (EVIDENCE_TOOLS[2],), run_id="large-run", candidate_ids=[101, 102],
    )[0]
    query_result = json.loads(query_tool(run_id="large-run", limit=100))
    assert [item["candidate_id"] for item in query_result["candidates"]] == [101, 102]

    evaluation_tool = bound_candidate_tools(
        (EVALUATION_TOOLS[1],), run_id="large-run", candidate_ids=[101, 102],
    )[0]
    evaluation_result = json.loads(evaluation_tool(
        run_id="large-run",
        items=[
            _evaluation(candidate_id=101),
            _evaluation(candidate_id=102),
        ],
    ))
    assert [item["candidate_id"] for item in evaluation_result["updated"]] == [101, 102]


def test_common_prompt_points_to_v2_tables_and_tools() -> None:
    assert "find_agent_v2_run" in COMMON_RULES
    assert "video_discovery_run" not in COMMON_RULES
    assert "batch_search_and_record" not in COMMON_RULES
    assert "batch_update_video_discovery_candidates" not in COMMON_RULES


def test_evaluator_prompt_defines_video_capability_and_judgment_boundaries() -> None:
    assert "understand_candidate_video_30s_v2" in EVALUATOR_PROMPT
    assert "同一候选" in EVALUATOR_PROMPT and "最多调用一次" in EVALUATOR_PROMPT
    assert "已经通过硬门禁" in EVALUATOR_PROMPT
    assert "不得把视频人物年龄当作受众画像" in EVALUATOR_PROMPT
    assert "不得用点赞量冒充分享量" in EVALUATOR_PROMPT
    assert "没有强日期依赖线索" in EVALUATOR_PROMPT
    assert "无法确认时按关键时间证据不足 rejected" in EVALUATOR_PROMPT
    assert "均属于强日期" in EVALUATOR_PROMPT
    assert "不能当作常青内容处理" in EVALUATOR_PROMPT
    assert "仍无法确认时，应 rejected" in EVALUATOR_PROMPT


@pytest.mark.parametrize(
    "previous,current,expected,reason",
    [
        (
            DiscoverySnapshot("running", 1, 10, 0, 3, 3, 7),
            DiscoverySnapshot("running", 2, 10, 0, 3, 3, 7),
            False,
            "没有新增候选",
        ),
        (
            DiscoverySnapshot("running", 0, 0, 0, 0, 0, 0),
            DiscoverySnapshot("running", 1, 4, 0, 1, 1, 3),
            True,
            "样本不足",
        ),
        (
            DiscoverySnapshot("running", 1, 4, 0, 1, 1, 3),
            DiscoverySnapshot("running", 2, 12, 0, 1, 1, 11),
            False,
            "通过率为 0",
        ),
        (
            DiscoverySnapshot("running", 1, 8, 0, 1, 1, 7),
            DiscoverySnapshot("running", 2, 12, 0, 2, 2, 10),
            True,
            "正向通过率",
        ),
    ],
)
def test_exploration_decision_uses_volume_and_pass_rate(
    previous, current, expected, reason,
) -> None:
    decision = decide_continued_exploration(previous, current)
    assert decision.continue_exploring is expected
    assert reason in decision.reason


def test_video_understanding_only_allows_candidates_passing_all_hard_gates(monkeypatch) -> None:
    service = _FakeService(pending_after_search=0)
    graph = FindAgentRoundGraph(service=service, runner=object())
    monkeypatch.setattr(service, "require_run", lambda _run_id: {
        "run_id": "run", "rule_config": build_rule_snapshot(),
    })
    common = {
        "video_url": "https://example.test/video.mp4",
        "publish_at": build_rule_snapshot()["current_datetime"],
        "content_50_plus_ratio": 0.30,
        "duration_seconds": 60,
        "share_count": 2000,
    }
    items = [
        {"candidate_id": 1, **common},
        {"candidate_id": 2, **common, "content_50_plus_ratio": 0.05},
        {"candidate_id": 3, **common, "publish_at": None},
        {"candidate_id": 4, **common, "video_url": None},
        {"candidate_id": 5, **common, "duration_seconds": 10},
        {"candidate_id": 6, **common, "share_count": 1},
        {"candidate_id": 7, **common, "duration_seconds": None},
        {"candidate_id": 8, **common, "share_count": None},
    ]

    assert graph._video_understanding_ids({"run_id": "run"}, items) == [1, 3]


class _FakeService:
    def __init__(self, *, pending_after_search: int) -> None:
        self.pending_after_search = pending_after_search
        self.stage = "start"
        self.updates: list[dict] = []
        self.rejected_ids: set[int] = set()

    def require_run(self, run_id: str):
        snapshot = self.snapshot(run_id)
        return {
            "run_id": run_id,
            "status": "running",
            "outcome_status": None,
            "current_round": 1,
            "search_count": snapshot.search_count,
            "candidate_count": snapshot.candidate_count,
            "valid_primary_count": snapshot.valid_primary_count,
            "rule_config": build_rule_snapshot(),
        }

    def get_search_summaries(self, _run_id: str):
        if self.stage == "start":
            return []
        return [{
            "search_id": 1,
            "round_index": 1,
            "keyword": "测试需求",
            "query_reason": "建立候选池",
            "provider": "internal_keyword",
            "page_no": 1,
            "has_more": False,
            "next_cursor": None,
            "status": "success",
            "result_count": self.pending_after_search,
        }]

    def _candidate_rows(self):
        if self.stage == "start":
            return []
        evidence_status = "pending" if self.stage == "searched" else "success"
        return [{
            "candidate_id": index,
            "decision_bucket": (
                "rejected"
                if self.stage == "evaluated" or index in self.rejected_ids
                else "pending_evaluation"
            ),
            "detail_status": evidence_status,
            "portrait_status": evidence_status,
            "publish_at": build_rule_snapshot()["current_datetime"],
            "duration_seconds": 60,
            "share_count": 2000,
            "content_50_plus_ratio": 0.30,
        } for index in range(1, self.pending_after_search + 1)]

    def get_full_state(self, run_id: str, **_kwargs):
        return {
            "run": self.require_run(run_id),
            "searches": self.get_search_summaries(run_id),
            "candidates": self._candidate_rows(),
        }

    def candidate_inputs(self, _run_id: str, candidate_ids: list[int]):
        allowed = {int(value) for value in candidate_ids}
        return [
            item for item in self._candidate_rows()
            if int(item["candidate_id"]) in allowed
        ]

    def get_candidate_progress(self, _run_id: str):
        candidates = self._candidate_rows()
        pending = [
            item for item in candidates
            if item["decision_bucket"] == "pending_evaluation"
        ]
        return {
            "total_count": len(candidates),
            "pending_count": len(pending),
            "primary_count": 0,
            "rejected_count": len(candidates) - len(pending),
            "detail_pending_count": sum(item["detail_status"] == "pending" for item in pending),
            "detail_success_count": sum(item["detail_status"] == "success" for item in pending),
            "detail_failed_count": 0,
            "portrait_pending_count": sum(item["portrait_status"] == "pending" for item in pending),
            "portrait_success_count": sum(item["portrait_status"] == "success" for item in pending),
            "portrait_failed_count": 0,
            "evidence_completed_count": sum(
                item["detail_status"] != "pending" and item["portrait_status"] != "pending"
                for item in pending
            ),
            "evidence_success_count": sum(
                item["detail_status"] == "success" and item["portrait_status"] == "success"
                for item in pending
            ),
        }

    def list_pending_evidence_ids(self, _run_id: str, evidence_type: str, *, limit: int):
        key = f"{evidence_type}_status"
        return [
            int(item["candidate_id"]) for item in self._candidate_rows()
            if item["decision_bucket"] == "pending_evaluation" and item[key] == "pending"
        ][:limit]

    def list_ready_evaluation_ids(self, _run_id: str, *, limit: int):
        return [
            int(item["candidate_id"]) for item in self._candidate_rows()
            if item["decision_bucket"] == "pending_evaluation"
            and item["detail_status"] != "pending"
            and item["portrait_status"] != "pending"
        ][:limit]

    def count_pending_candidates(self, run_id: str) -> int:
        return int(self.get_candidate_progress(run_id)["pending_count"])

    def get_report_state(self, run_id: str):
        return {
            "run": self.require_run(run_id),
            "summary": self.get_candidate_progress(run_id),
            "primary_candidates": [],
            "rejection_reason_distribution": {},
        }

    def snapshot(self, _run_id: str) -> DiscoverySnapshot:
        base = DiscoverySnapshot("running", 0, 0, 0, 0, 0, 0)
        if self.stage == "searched":
            return replace(base, search_count=1, candidate_count=self.pending_after_search,
                           pending_count=self.pending_after_search)
        if self.stage in {"evidenced", "batched"}:
            return replace(base, search_count=1, candidate_count=self.pending_after_search,
                           pending_count=self.pending_after_search)
        if self.stage == "evaluated":
            return replace(base, search_count=1, candidate_count=self.pending_after_search,
                           rejected_count=self.pending_after_search)
        return base

    def update_round(self, _run_id: str, _round_index: int, **kwargs) -> None:
        self.updates.append(kwargs)

    def recount_valid_primary(self, _run_id: str) -> int:
        return 0

    def reject_failed_gates(self, _run_id: str, failures) -> list[int]:
        rejected = [candidate_id for candidate_id, _gate in failures]
        self.rejected_ids.update(rejected)
        return rejected


class _FakeRunner:
    def __init__(self, service: _FakeService) -> None:
        self.service = service
        self.calls: list[tuple[str, set[str]]] = []
        self.inputs: list[tuple[str, str]] = []
        self.supervisor_visual_outputs: list[dict] = []

    async def run_node(self, *, node, round_index, tools=(), user_content="", **_kwargs) -> NodeRun:
        self.calls.append((node, _names(tools)))
        self.inputs.append((node, user_content))
        if node == "search":
            self.service.stage = "searched"
        elif node == "evidence":
            self.service.stage = "evidenced"
        elif node == "evaluator":
            self.service.stage = "evaluated"
        if node == "supervisor":
            if self.service.stage == "evaluated":
                next_action = "finish"
            elif self.service.stage in {"evidenced", "batched"}:
                next_action = "evaluator"
            elif self.service.stage == "searched" and not self.service.pending_after_search:
                next_action = "finish"
            else:
                next_action = "search"
            content = json.dumps({
                "next_action": next_action,
                "worker_count": 4,
                "evidence_scope": "both",
                "plan": {"searches": []},
            })
        else:
            content = '{"searches": []}'
        return NodeRun(node, round_index, content, 1, 0)

    @staticmethod
    def execution_plan() -> ExecutionPlan:
        return ExecutionPlan(
            demand_brief=DemandBrief(
                core_intent="测试需求",
                relevance_criteria=["内容直接匹配测试需求"],
            ),
            search_tasks=[SearchTask(
                task_id="search-1", keyword="测试需求", query_reason="建立候选池",
            )],
            evaluation_brief=EvaluationBrief(
                relevance_criteria=["内容直接匹配测试需求"],
            ),
        )

    async def run_planning(self, *, round_index, **kwargs):
        run = await self.run_node(node="supervisor", round_index=round_index, **kwargs)
        proposal = json.loads(run.content)
        return run, PlanningDecision(
            next_action=proposal["next_action"],
            worker_count=proposal["worker_count"],
            evidence_scope=proposal["evidence_scope"],
            execution_plan=self.execution_plan(),
        )

    async def run_supervision(self, *, round_index, **kwargs):
        output_enricher = kwargs.pop("output_enricher", None)
        run = await self.run_node(node="supervisor", round_index=round_index, **kwargs)
        proposal = json.loads(run.content)
        decision = SupervisorDecision(
            next_action=proposal["next_action"],
            worker_count=proposal["worker_count"],
            evidence_scope=proposal["evidence_scope"],
        )
        if output_enricher is not None:
            self.supervisor_visual_outputs.append(output_enricher(decision))
        return run, decision


class _WrongEvidenceProposalRunner(_FakeRunner):
    async def run_node(self, *, node, round_index, tools=(), user_content="", **kwargs):
        run = await super().run_node(
            node=node,
            round_index=round_index,
            tools=tools,
            user_content=user_content,
            **kwargs,
        )
        if node == "supervisor" and self.service.stage == "evidenced":
            return NodeRun(
                node,
                round_index,
                json.dumps({
                    "next_action": "evidence",
                    "worker_count": 8,
                    "evidence_scope": "both",
                }),
                1,
                0,
            )
        return run


class _OptimizedRunner(_FakeRunner):
    def __init__(self, service: _FakeService) -> None:
        super().__init__(service)
        self.host_calls: list[str] = []

    async def run_search_assignment(self, *, round_index, user_content, **_kwargs):
        self.host_calls.append("search")
        self.inputs.append(("search", user_content))
        self.service.stage = "searched"
        return NodeRun("search", round_index, "host search", 0, 1)

    async def run_evidence_assignment(
        self, *, round_index, user_content, **_kwargs,
    ):
        self.host_calls.append("evidence")
        self.inputs.append(("evidence", user_content))
        self.service.stage = "evidenced"
        return NodeRun("evidence", round_index, "host evidence", 0, 1)

    async def run_host_supervision(
        self, *, round_index, decision, output_enricher=None, **_kwargs,
    ):
        self.host_calls.append(f"supervisor:{decision.next_action}")
        if output_enricher is not None:
            self.supervisor_visual_outputs.append(output_enricher(decision))
        return NodeRun("supervisor", round_index, decision.model_dump_json(), 0, 0), decision


@pytest.mark.asyncio
async def test_host_executes_deterministic_stages_without_redundant_llm_calls() -> None:
    service = _FakeService(pending_after_search=2)
    runner = _OptimizedRunner(service)
    graph = FindAgentRoundGraph(service=service, runner=runner)

    result = await graph.invoke(FindAgentState(
        run_id="optimized-run", user_input="task", round_index=1,
    ))

    assert runner.host_calls == [
        "search",
        "supervisor:evidence",
        "evidence",
        "evidence",
        "supervisor:evaluator",
    ]
    assert [node for node, _tools in runner.calls] == [
        "supervisor", "evaluator", "supervisor",
    ]
    assert result.snapshot is not None and result.snapshot.pending_count == 0


def test_supervisor_progress_counts_missing_evidence_only_for_pending_candidates() -> None:
    class ProgressService:
        @staticmethod
        def require_run(_run_id: str):
            return {}

        @staticmethod
        def get_search_summaries(_run_id: str):
            return []

        @staticmethod
        def get_candidate_progress(_run_id: str):
            return {
                "total_count": 4,
                "pending_count": 3,
                "primary_count": 0,
                "rejected_count": 1,
                "detail_pending_count": 0,
                "detail_success_count": 2,
                "detail_failed_count": 1,
                "portrait_pending_count": 1,
                "portrait_success_count": 2,
                "portrait_failed_count": 0,
                "evidence_completed_count": 2,
                "evidence_success_count": 1,
            }

    graph = FindAgentRoundGraph(service=ProgressService(), runner=object())
    progress = graph._supervisor_state({"run_id": "run"})["candidate_progress"]

    assert progress["pending_count"] == 3
    assert progress["detail_pending_count"] == 0
    assert progress["detail_success_count"] == 2
    assert progress["detail_failed_count"] == 1
    assert progress["portrait_pending_count"] == 1
    assert progress["portrait_success_count"] == 2
    assert progress["portrait_failed_count"] == 0
    assert progress["evidence_completed_count"] == 2
    assert progress["evidence_success_count"] == 1


@pytest.mark.asyncio
async def test_supervisor_visualization_records_host_override() -> None:
    service = _FakeService(pending_after_search=1)
    runner = _WrongEvidenceProposalRunner(service)
    graph = FindAgentRoundGraph(service=service, runner=runner)

    await graph.invoke(FindAgentState(run_id="override-run", user_input="task", round_index=1))

    validations = [
        item["Supervisor决策校验"] for item in runner.supervisor_visual_outputs
    ]
    overridden = next(
        item
        for item in validations
        if item["proposed_action"] == "evidence"
        and item["approved_action"] == "evaluator"
    )
    assert overridden["proposed_action"] == "evidence"
    assert overridden["approved_action"] == "evaluator"


@pytest.mark.asyncio
async def test_round_graph_supervisor_routes_with_guarded_allowlists() -> None:
    service = _FakeService(pending_after_search=2)
    runner = _FakeRunner(service)
    graph = FindAgentRoundGraph(service=service, runner=runner)
    state = FindAgentState(run_id="new-run", user_input="task", round_index=1)

    result = await graph.invoke(state)

    assert [name for name, _ in runner.calls] == [
        "supervisor", "search", "supervisor", "evidence", "evidence",
        "supervisor", "evaluator", "supervisor",
    ]
    assert runner.calls[0][1] == set()
    assert runner.calls[1][1] == _names(SEARCH_TOOLS)
    assert runner.calls[2][1] == set()
    assert runner.calls[3][1] <= _names(EVIDENCE_TOOLS)
    assert runner.calls[4][1] <= _names(EVIDENCE_TOOLS)
    assert runner.calls[6][1] == {
        "evaluate_candidates_v2", "query_pending_candidates_v2",
    }
    assert result.phase == "done"
    assert result.snapshot is not None and result.snapshot.pending_count == 0

@pytest.mark.asyncio
async def test_raw_demand_is_only_sent_to_initial_planner() -> None:
    service = _FakeService(pending_after_search=1)
    runner = _FakeRunner(service)
    graph = FindAgentRoundGraph(service=service, runner=runner)
    marker = "RAW-DEMAND-MUST-NOT-LEAK"

    await graph.invoke(FindAgentState(
        run_id="assignment-run", user_input=marker, round_index=1,
    ))

    assert marker in runner.inputs[0][1]
    assert all(marker not in content for _node, content in runner.inputs[1:])
    search_payload = json.loads(next(
        content for node, content in runner.inputs if node == "search"
    ))
    evidence_payload = json.loads(next(
        content for node, content in runner.inputs if node == "evidence"
    ))
    supervisor_payload = json.loads([
        content for node, content in runner.inputs if node == "supervisor"
    ][1])
    assert set(search_payload) == {"run_id", "round_index", "tasks"}
    assert set(evidence_payload) == {
        "run_id", "round_index", "candidate_ids", "evidence_type", "candidates",
    }
    assert "candidates" not in supervisor_payload["execution_state"]
    assert "candidate_progress" in supervisor_payload["execution_state"]


def test_obagent_declaration_cannot_override_actual_model_input() -> None:
    class Context:
        declared = False

        def declare(self, **_kwargs):
            self.declared = True
            return "SDK-rendered-input-that-must-not-reach-model"

    context = Context()
    handle = _ModuleHandle(context)
    actual = handle.declare(
        fallback='{"run_id":"r","tasks":[]}',
        system_prompt="prompt",
        slots=(InputSlot("阶段任务", "{}", "stage_assignment", "test", False),),
        tools=(),
        model="model",
    )

    assert context.declared is True
    assert actual == '{"run_id":"r","tasks":[]}'


@pytest.mark.asyncio
async def test_round_graph_skips_evidence_and_evaluation_without_candidates() -> None:
    service = _FakeService(pending_after_search=0)
    runner = _FakeRunner(service)
    graph = FindAgentRoundGraph(service=service, runner=runner)

    await graph.invoke(FindAgentState(run_id="new-run", user_input="task", round_index=1))

    assert [name for name, _ in runner.calls] == [
        "supervisor", "search", "supervisor",
    ]


@pytest.mark.asyncio
async def test_evaluator_model_never_receives_hard_gate_failures() -> None:
    service = _FakeService(pending_after_search=1)
    service.stage = "evidenced"
    rejected: list[int] = []

    bucket = ["pending_evaluation"]

    def candidate_inputs(_run_id: str, candidate_ids: list[int]):
        if 1 not in candidate_ids:
            return []
        return [{
            "candidate_id": 1,
            "decision_bucket": bucket[0],
            "detail_status": "success",
            "portrait_status": "success",
            "publish_at": build_rule_snapshot()["current_datetime"],
            "duration_seconds": 10,
            "share_count": 2000,
            "content_50_plus_ratio": 0.30,
            "video_url": "https://example.test/video.mp4",
        }]

    def reject(_run_id: str, failures):
        ids = [candidate_id for candidate_id, _ in failures]
        rejected.extend(ids)
        bucket[0] = "rejected"
        return ids

    service.candidate_inputs = candidate_inputs  # type: ignore[method-assign]
    service.list_ready_evaluation_ids = (  # type: ignore[method-assign]
        lambda _run_id, limit: [1] if bucket[0] == "pending_evaluation" else []
    )
    service.reject_failed_gates = reject  # type: ignore[method-assign]
    runner = _FakeRunner(service)
    graph = FindAgentRoundGraph(service=service, runner=runner)

    await graph._evaluator({
        "run_id": "r", "round_index": 1, "worker_count": 1,
        "action_count": 0, "node_runs": [], "user_input": "需求",
    })

    assert rejected == [1]
    assert not any(name == "evaluator" for name, _tools in runner.calls)


@pytest.mark.asyncio
async def test_evaluator_drains_database_batches_until_pending_queue_is_empty() -> None:
    class BatchedService(_FakeService):
        def __init__(self) -> None:
            super().__init__(pending_after_search=3)
            self.remaining_ids = {1, 2, 3}

        def _candidate_rows(self):
            if self.stage == "start":
                return []
            evidence_status = "pending" if self.stage == "searched" else "success"
            return [{
                "candidate_id": index,
                "decision_bucket": (
                    "pending_evaluation" if index in self.remaining_ids else "rejected"
                ),
                "detail_status": evidence_status,
                "portrait_status": evidence_status,
                "publish_at": build_rule_snapshot()["current_datetime"],
                "duration_seconds": 60,
                "share_count": 2000,
                "content_50_plus_ratio": 0.30,
            } for index in range(1, 4)]

        def list_ready_evaluation_ids(self, _run_id: str, *, limit: int):
            del limit
            return sorted(self.remaining_ids)[:1]

        def snapshot(self, _run_id: str) -> DiscoverySnapshot:
            if self.stage == "start":
                return DiscoverySnapshot("running", 0, 0, 0, 0, 0, 0)
            return DiscoverySnapshot(
                "running", 1, 3, len(self.remaining_ids), 0, 0,
                3 - len(self.remaining_ids),
            )

    service = BatchedService()

    class BatchedRunner(_FakeRunner):
        def __init__(self, fake_service: _FakeService) -> None:
            super().__init__(fake_service)
        async def run_node(
            self, *, node, round_index, tools=(), user_content="", **kwargs,
        ) -> NodeRun:
            if node == "evaluator":
                self.calls.append((node, _names(tools)))
                self.inputs.append((node, user_content))
                candidate_ids = json.loads(user_content)["candidate_ids"]
                self.service.remaining_ids.difference_update(candidate_ids)
                self.service.stage = (
                    "evaluated" if not self.service.remaining_ids else "batched"
                )
                return NodeRun(node, round_index, "", 1, 0)
            return await super().run_node(
                node=node,
                round_index=round_index,
                tools=tools,
                user_content=user_content,
                **kwargs,
            )

    runner = BatchedRunner(service)
    graph = FindAgentRoundGraph(service=service, runner=runner)

    result = await graph.invoke(
        FindAgentState(run_id="batched-run", user_input="task", round_index=1),
    )

    assert [name for name, _ in runner.calls].count("evaluator") == 3
    assert result.snapshot is not None and result.snapshot.pending_count == 0


def test_supervisor_policy_overrides_unsafe_finish_and_clamps_workers() -> None:
    service = _FakeService(pending_after_search=2)
    service.stage = "searched"
    graph = FindAgentRoundGraph(service=service, runner=_FakeRunner(service))
    action, _reason, workers, scope = graph._approve_action(
        {"run_id": "r", "search_actions": 1, "action_count": 1},
        {"next_action": "finish", "worker_count": 99, "evidence_scope": "invalid"},
    )
    assert action == "evidence"
    assert workers == 8
    assert scope == "both"


def test_supervisor_can_choose_an_extra_search_within_budget() -> None:
    service = _FakeService(pending_after_search=0)
    service.stage = "searched"
    graph = FindAgentRoundGraph(service=service, runner=_FakeRunner(service))
    action, *_ = graph._approve_action(
        {"run_id": "r", "search_actions": 1, "action_count": 1},
        {"next_action": "search", "worker_count": 2},
    )
    assert action == "search"


@pytest.mark.asyncio
async def test_round_graph_rejects_incomplete_evaluator_batch() -> None:
    service = _FakeService(pending_after_search=2)

    class RetryRunner(_FakeRunner):
        evaluator_calls = 0

        async def run_node(self, *, node, round_index, tools=(), **kwargs) -> NodeRun:
            if node == "evaluator":
                self.calls.append((node, _names(tools)))
                self.evaluator_calls += 1
                return NodeRun(node, round_index, "", 1, 0)
            return await super().run_node(
                node=node, round_index=round_index, tools=tools, **kwargs,
            )

    runner = RetryRunner(service)
    graph = FindAgentRoundGraph(service=service, runner=runner)

    with pytest.raises(RuntimeError, match="评估分批未完整消费"):
        await graph.invoke(
            FindAgentState(run_id="retry-run", user_input="task", round_index=1),
        )

    assert runner.evaluator_calls == 1