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, ), ( 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_strategy_tool_accepts_only_workbench_handles(): registry = ToolRegistry() register_script_tools(registry, object()) # type: ignore[arg-type] schema = registry._tools["load_frozen_strategy"]["schema"] strategy_ref = schema["function"]["parameters"]["properties"]["strategy_handle"] assert strategy_ref["pattern"] == "^strategy_[0-9a-f]{20}$" 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")