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,
    ToolExecutionError,
    ToolRegistry,
)
from agent.orchestration import CompletionPolicy, DeterministicWorkerContext
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.application.workbench_workers import ScriptBuildDeterministicWorker
from script_build_host.domain.errors import (
    ArtifactDigestMismatch,
    MissionFencingTokenStale,
    PublicationLockTimeout,
    ScriptBuildError,
)
from script_build_host.domain.task_contracts import TaskContractError
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,
        ),
        (
            ScriptBuildError("TASK_KIND_PRESET_MISMATCH", "Task kind cannot change"),
            FailureDisposition.REPLAN_TASK,
        ),
        (
            ScriptBuildError("TASK_CAPABILITY_MISMATCH", "criterion params are invalid"),
            FailureDisposition.REPLAN_TASK,
        ),
        (
            ScriptBuildError("CONTEXT_HANDLE_UNAUTHORIZED", "outside task"),
            FailureDisposition.RETRY_CALL,
        ),
        (
            ScriptBuildError("CONTEXT_CURSOR_INVALID", "cursor is stale"),
            FailureDisposition.RETRY_CALL,
        ),
        (
            ScriptBuildError("CONTEXT_CURSOR_SCOPE_MISMATCH", "cursor belongs to another query"),
            FailureDisposition.RETRY_CALL,
        ),
        (
            ScriptBuildError("WRITE_SCOPE_CONFLICT", "choose one candidate"),
            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"


def test_phase_policy_failures_keep_semantic_details_in_their_fingerprint():
    structure = classify_script_tool_failure(
        TaskContractError(
            "PHASE_POLICY_VIOLATION",
            "wrong structure placement",
            details={"attempted_task_kinds": ["structure"]},
        ),
        source_tool="plan_script_tasks",
    )
    paragraph = classify_script_tool_failure(
        TaskContractError(
            "PHASE_POLICY_VIOLATION",
            "wrong paragraph placement",
            details={"attempted_task_kinds": ["paragraph"]},
        ),
        source_tool="plan_script_tasks",
    )

    assert structure.details["attempted_task_kinds"] == ["structure"]
    assert structure.fingerprint() != paragraph.fingerprint()


def test_paragraph_batch_tool_exposes_exact_nested_contract():
    registry = ToolRegistry()
    register_script_tools(registry, object())  # type: ignore[arg-type]
    schema = registry._tools["save_script_paragraphs"]["schema"]
    item = schema["function"]["parameters"]["properties"]["paragraphs"]["items"]

    assert {entry["required"][0] for entry in item["oneOf"]} == {
        "client_key",
        "paragraph_target_key",
    }
    assert "index" not in item["properties"]
    assert item["properties"]["paragraph_index"]["minimum"] == 1
    assert item["properties"]["theme"]["type"] == "string"
    assert item["properties"]["theme_elements"]["type"] == "array"
    assert "paragraph_id" not in item["properties"]
    assert item["additionalProperties"] is False


def test_context_read_tool_accepts_only_broker_handles_and_opaque_cursor():
    registry = ToolRegistry()
    register_script_tools(registry, object())  # type: ignore[arg-type]
    schema = registry._tools["read_mission_context"]["schema"]
    properties = schema["function"]["parameters"]["properties"]

    assert schema["function"]["parameters"]["required"] == ["handle"]
    assert properties["cursor"]["type"] == ["string", "null"]


def test_validator_unread_required_context_is_retryable_without_replanning():
    detail = classify_script_tool_failure(
        ScriptBuildError("CONTEXT_NOT_EXHAUSTED", "read remaining pages"),
        source_tool="submit_validation",
    )

    assert detail.code == "CONTEXT_NOT_EXHAUSTED"
    assert detail.disposition is FailureDisposition.RETRY_CALL


def test_element_tool_exposes_only_semantic_targets_and_dimension_contracts():
    registry = ToolRegistry()
    register_script_tools(registry, object())  # type: ignore[arg-type]

    parameters = registry._tools["save_script_elements"]["schema"]["function"]["parameters"]
    element = parameters["properties"]["elements"]["items"]
    assert element["properties"]["dimension_primary"]["enum"] == ["实质", "形式"]
    assert "element_id" not in element["properties"]

    links = parameters["properties"]["links"]["items"]
    assert links["properties"]["paragraph_target_key"]["pattern"].startswith("^pt_")
    assert "paragraph_id" not in links["properties"]
    assert "element_ids" not in links["properties"]
    assert links["additionalProperties"] is False


@pytest.mark.asyncio
async def test_compose_scope_failure_exits_worker_and_keeps_exact_domain_error(tmp_path):
    del tmp_path

    class Candidates:
        async def resolve_attempt_manifest(self, *, context):
            del context
            raise PhaseTwoCandidateError("ARTIFACT_NOT_FOUND", "not frozen")

        async def save_structured_script_candidate(self, *, acceptance_notes, context):
            del acceptance_notes, context
            raise PhaseTwoCandidateError(
                "INPUT_SCOPE_MISMATCH",
                "Paragraph has no adopted covering Structure",
            )

    worker = ScriptBuildDeterministicWorker(Candidates())
    context = DeterministicWorkerContext(
        root_trace_id="root",
        task=type(
            "Task",
            (),
            {
                "task_id": "compose-task",
                "current_spec": type(
                    "Spec", (), {"context_refs": ("script-build://task-kinds/compose",)}
                )(),
                "current_spec_version": 1,
            },
        )(),
        attempt=type("Attempt", (), {"attempt_id": "attempt-1"})(),
        ledger_revision=1,
        accepted_child_results=(),
        operation_id=None,
        execution_epoch=0,
        role_context={},
    )
    with pytest.raises(ToolExecutionError) as captured:
        await worker.execute(context)

    assert captured.value.failure.to_dict() == {
        "code": "INPUT_SCOPE_MISMATCH",
        "message": "Paragraph has no adopted covering Structure",
        "disposition": "replan_task",
        "source_tool": "deterministic_compose",
        "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 == "save_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 = "save_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"}],
                            "theme": "A concrete opening theme",
                            "form": "A visible contrast",
                            "function": "Hooks the audience",
                            "feeling": "Creates curiosity",
                            "description": "A complete opening paragraph",
                            "full_description": "The audience sees the concrete opening copy.",
                        }
                    ],
                    "expected_state_revision": "revision-1",
                }
            )
        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=["save_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_dispatch_converts_child_no_progress_abort_into_planner_replan():
    child_failure = FailureDetail(
        code="NO_PROGRESS",
        message="worker repeated the same invalid write",
        disposition=FailureDisposition.ABORT_RUN,
        source_tool="create_script_paragraphs",
        details={"reason": "same_failure_repeated"},
    )

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

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

    assert result.failure.code == "NO_PROGRESS"
    assert result.failure.disposition is FailureDisposition.REPLAN_TASK
    assert result.failure.source_tool == "dispatch_script_tasks"
    assert result.failure.details["child_disposition"] == "abort_run"
    assert result.failure.details["child_source_tool"] == "create_script_paragraphs"


@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")