script-build-and-framework-tmp/script_build_host/tests/test_tool_failure_bridge.py

from __future__ import annotations

import json

import pytest
from agent import (
    AgentRunner,
    FailureDetail,
    FailureDisposition,
    RunConfig,
    ToolRegistry,
)
from agent.orchestration import CompletionPolicy
from agent.trace.store import FileSystemTraceStore

from script_build_host.agents.presets import register_script_presets
from script_build_host.application.phase_two_candidates import PhaseTwoCandidateError
from script_build_host.domain.errors import (
    ArtifactDigestMismatch,
    MissionFencingTokenStale,
    PublicationLockTimeout,
    ScriptBuildError,
)
from script_build_host.tools.failures import classify_script_tool_failure
from script_build_host.tools.registry import register_script_tools


class _Coordinator:
    async def ensure_ledger(self, *_args, **_kwargs):
        return None


def _knowledge_off():
    from agent.tools.builtin.knowledge import KnowledgeConfig

    return KnowledgeConfig(
        enable_extraction=False,
        enable_completion_extraction=False,
        enable_injection=False,
    )


@pytest.mark.parametrize(
    ("error", "expected"),
    [
        (
            PhaseTwoCandidateError("INPUT_SCOPE_MISMATCH", "scope is invalid"),
            FailureDisposition.REPLAN_TASK,
        ),
        (
            PhaseTwoCandidateError("GOAL_COVERAGE_INCOMPLETE", "goal is missing"),
            FailureDisposition.REPLAN_TASK,
        ),
        (
            PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "workspace is invalid"),
            FailureDisposition.REPAIR_ATTEMPT,
        ),
        (PublicationLockTimeout(), FailureDisposition.RETRY_CALL),
        (MissionFencingTokenStale(), FailureDisposition.ABORT_RUN),
        (ArtifactDigestMismatch(), FailureDisposition.ABORT_RUN),
        (
            ScriptBuildError("UNCLASSIFIED_DOMAIN_FAILURE", "fail closed"),
            FailureDisposition.ABORT_RUN,
        ),
        (ValueError("bad argument"), FailureDisposition.RETRY_CALL),
    ],
)
def test_script_failure_classification_is_central_and_fail_closed(error, expected):
    failure = classify_script_tool_failure(
        error,
        source_tool="test_tool",
        context={"task_id": "task-1", "attempt_id": "attempt-1"},
    )

    assert failure.disposition is expected
    assert failure.source_tool == "test_tool"
    assert failure.details["task_id"] == "task-1"


@pytest.mark.asyncio
async def test_compose_scope_failure_exits_worker_and_keeps_exact_domain_error(tmp_path):
    class Gateway:
        async def save_structured_script_candidate(self, _notes, _context):
            raise PhaseTwoCandidateError(
                "INPUT_SCOPE_MISMATCH",
                "Paragraph has no adopted covering Structure",
            )

    register_script_presets()
    registry = ToolRegistry()
    register_script_tools(registry, Gateway())  # type: ignore[arg-type]
    calls = 0

    async def llm_call(**_kwargs):
        nonlocal calls
        calls += 1
        return {
            "content": "",
            "tool_calls": [{
                "id": "save-compose",
                "type": "function",
                "function": {
                    "name": "save_structured_script_candidate",
                    "arguments": "{}",
                },
            }],
            "finish_reason": "tool_calls",
        }

    result = await AgentRunner(
        trace_store=FileSystemTraceStore(str(tmp_path)),
        tool_registry=registry,
        llm_call=llm_call,
        task_coordinator=_Coordinator(),
    ).run_result(
        [{"role": "user", "content": "compose"}],
        RunConfig(
            agent_type="script_compose_worker",
            completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
            tools=["save_structured_script_candidate"],
            tool_groups=[],
            context={
                "root_trace_id": "root",
                "task_id": "compose-task",
                "attempt_id": "attempt-1",
            },
            knowledge=_knowledge_off(),
        ),
    )

    assert calls == 1
    assert result["status"] == "failed"
    assert result["failure"] == {
        "code": "INPUT_SCOPE_MISMATCH",
        "message": "Paragraph has no adopted covering Structure",
        "disposition": "replan_task",
        "source_tool": "save_structured_script_candidate",
        "details": {"attempt_id": "attempt-1", "task_id": "compose-task"},
    }


@pytest.mark.asyncio
async def test_workspace_error_can_be_repaired_in_same_attempt(tmp_path):
    class Gateway:
        def __init__(self):
            self.write_calls = 0

        async def candidate_command(self, name, _payload, _context):
            assert name == "create_script_paragraphs"
            self.write_calls += 1
            if self.write_calls == 1:
                raise PhaseTwoCandidateError(
                    "LEGACY_WRITE_INVALID",
                    "paragraph dimension is duplicated",
                )
            return {"created": True}

        async def submit_current_attempt(self, _context):
            return {"attempt_id": "attempt-1", "status": "awaiting_validation"}

    register_script_presets()
    gateway = Gateway()
    registry = ToolRegistry()
    register_script_tools(registry, gateway)  # type: ignore[arg-type]
    calls = 0

    async def llm_call(**_kwargs):
        nonlocal calls
        calls += 1
        if calls <= 2:
            tool_name = "create_script_paragraphs"
            arguments = json.dumps({
                "paragraphs": [{
                    "client_key": "opening",
                    "paragraph_index": 1,
                    "name": "opening",
                    "content_range": {"start": 0, "end": 100},
                    "level": 1,
                    "theme_elements": [{"原子点": "theme", "维度": "topic", "维度类型": "主维度"}],
                    "form_elements": [{"原子点": "form", "维度": "contrast", "维度类型": "主维度"}],
                    "function_elements": [{"原子点": "hook", "维度": "role", "维度类型": "主维度"}],
                    "feeling_elements": [{"原子点": "curious", "维度": "tone"}],
                }],
            })
        else:
            tool_name = "submit_attempt"
            arguments = "{}"
        return {
            "content": "",
            "tool_calls": [{
                "id": f"call-{calls}",
                "type": "function",
                "function": {"name": tool_name, "arguments": arguments},
            }],
            "finish_reason": "tool_calls",
        }

    result = await AgentRunner(
        trace_store=FileSystemTraceStore(str(tmp_path)),
        tool_registry=registry,
        llm_call=llm_call,
        task_coordinator=_Coordinator(),
    ).run_result(
        [{"role": "user", "content": "write paragraph"}],
        RunConfig(
            agent_type="script_paragraph_worker",
            completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
            tools=["create_script_paragraphs", "submit_attempt"],
            tool_groups=[],
            context={"task_id": "paragraph-task", "attempt_id": "attempt-1"},
            knowledge=_knowledge_off(),
        ),
    )

    assert gateway.write_calls == 2
    assert calls == 3
    assert result["status"] == "completed"
    assert result["failure"] is None


@pytest.mark.asyncio
async def test_dispatch_preserves_batch_and_surfaces_child_failure():
    child_failure = FailureDetail(
        code="INPUT_SCOPE_MISMATCH",
        message="scope mismatch",
        disposition=FailureDisposition.REPLAN_TASK,
        source_tool="save_structured_script_candidate",
    )

    class Gateway:
        async def dispatch_script_tasks(self, *, task_ids, context):
            assert task_ids == ["compose-task", "other-task"]
            assert context["root_trace_id"] == "root"
            return [
                {
                    "task_id": "compose-task",
                    "task_status": "needs_replan",
                    "attempt_id": "attempt-1",
                    "failure": child_failure.to_dict(),
                },
                {"task_id": "other-task", "task_status": "completed"},
            ]

    registry = ToolRegistry()
    register_script_tools(registry, Gateway())  # type: ignore[arg-type]
    result = await registry._tools["dispatch_script_tasks"]["func"](
        task_ids=["compose-task", "other-task"],
        context={"root_trace_id": "root"},
    )

    assert "other-task" in result.output
    assert result.failure.code == "INPUT_SCOPE_MISMATCH"
    assert result.failure.source_tool == "save_structured_script_candidate"
    assert result.failure.details == {
        "attempt_id": "attempt-1",
        "task_id": "compose-task",
    }


@pytest.mark.asyncio
async def test_planner_breaks_after_same_dispatched_failure_twice(tmp_path):
    child_failure = FailureDetail(
        code="INPUT_SCOPE_MISMATCH",
        message="scope mismatch",
        disposition=FailureDisposition.REPLAN_TASK,
        source_tool="save_structured_script_candidate",
    )

    class Gateway:
        async def dispatch_script_tasks(self, *, task_ids, context):
            return [{
                "task_id": task_ids[0],
                "task_status": "needs_replan",
                "attempt_id": "volatile-attempt-id",
                "failure": child_failure.to_dict(),
            }]

    register_script_presets()
    registry = ToolRegistry()
    register_script_tools(registry, Gateway())  # type: ignore[arg-type]
    calls = 0

    async def llm_call(**_kwargs):
        nonlocal calls
        calls += 1
        return {
            "content": "",
            "tool_calls": [{
                "id": f"dispatch-{calls}",
                "type": "function",
                "function": {
                    "name": "dispatch_script_tasks",
                    "arguments": json.dumps({"task_ids": ["compose-task"]}),
                },
            }],
            "finish_reason": "tool_calls",
        }

    result = await AgentRunner(
        trace_store=FileSystemTraceStore(str(tmp_path)),
        tool_registry=registry,
        llm_call=llm_call,
        task_coordinator=_Coordinator(),
    ).run_result(
        [{"role": "user", "content": "plan"}],
        RunConfig(
            agent_type="script_planner",
            completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
            tools=["dispatch_script_tasks"],
            tool_groups=[],
            root_task_spec={
                "objective": "deliver script",
                "acceptance_criteria": [{"description": "script is accepted"}],
            },
            context={"root_trace_id": "root"},
            knowledge=_knowledge_off(),
        ),
    )

    assert calls == 2
    assert result["status"] == "incomplete"
    assert result["failure"]["code"] == "NO_PROGRESS"
    assert result["failure"]["details"]["last_failure"]["code"] == (
        "INPUT_SCOPE_MISMATCH"
    )