fix(Agent工具): 固化结构化参数与验证证据闭包

为规划、候选写入和验证工具提供严格 JSON schema、大小边界与字符串参数归一化，检索成功后原子提交唯一 Evidence。

Validator 现在读取当前 Attempt 的冻结业务 Artifact；同时去重 evidence 字段，并避免跨长 Operation 持有 fencing 锁。

script_build_host/src/script_build_host/agents/presets.py

@@ -194,7 +194,6 @@ def register_script_presets() -> None:
     )
     validator_tools = [
         "read_input_snapshot",
-        "view_frozen_images",
         "query_validation_evidence",
         "deterministic_precheck",
         "submit_validation",
@@ -225,7 +224,7 @@ def register_script_presets() -> None:
         "script_candidate_validator",
         AgentPreset(
             role=AgentRole.VALIDATOR,
-            allowed_tools=validator_tools,
+            allowed_tools=[*validator_tools, "view_frozen_images"],
             denied_tools=validator_denied,
             max_iterations=25,
             temperature=0.0,

script_build_host/src/script_build_host/tools/gateway.py

@@ -188,12 +188,14 @@ class LegacyScriptToolGateway:
     async def dispatch_script_tasks(
         self, *, task_ids: Sequence[str], context: Mapping[str, Any]
     ) -> Sequence[Mapping[str, Any]]:
+        # Dispatch waits for the durable Operation (Worker + Validator) to
+        # finish. Holding the binding row lock across that wait deadlocks the
+        # child submissions, which must independently pass the same fence.
+        # The planning service performs a short owner/fence verification
+        # immediately before it reserves the Operation.
         return cast(
             Sequence[Mapping[str, Any]],
-            await self._execute_fenced(
-                context,
-                lambda: self._planner().dispatch_script_tasks(task_ids=task_ids, context=context),
-            ),
+            await self._planner().dispatch_script_tasks(task_ids=task_ids, context=context),
         )
 
     async def read_input_snapshot(self, context: Mapping[str, Any]) -> dict[str, Any]:
@@ -822,7 +824,44 @@ class LegacyScriptToolGateway:
             limit=limit,
         )
         response = await self.coordinator.query_evidence(dict(context), request)
-        return cast(dict[str, Any], _json_values(asdict(response)))
+        payload = cast(dict[str, Any], _json_values(asdict(response)))
+        ledger = await self.coordinator.task_store.load(request.root_trace_id)
+        task = ledger.tasks.get(request.task_id)
+        snapshot = await self.coordinator.artifact_store.get(
+            request.root_trace_id, request.snapshot_id
+        )
+        binding = await self.bindings.get_by_root(request.root_trace_id)
+        current_artifacts: list[dict[str, Any]] = []
+        for ref in snapshot.artifact_refs:
+            version = await self.artifacts.read_by_ref(
+                ref,
+                script_build_id=binding.script_build_id,
+                task_id=request.task_id,
+                attempt_id=request.attempt_id,
+            )
+            current_artifacts.append(
+                {
+                    "artifact_ref": _ref_dict(ref),
+                    "artifact": _json_values(version.artifact.content_payload()),
+                }
+            )
+        if current_artifacts:
+            payload["current_artifacts"] = current_artifacts
+        if task is not None and task_kind(task.current_spec.context_refs) in RETRIEVAL_KINDS:
+            if len(snapshot.evidence_refs) == 1:
+                ref = snapshot.evidence_refs[0]
+                version = await self.artifacts.read_by_ref(
+                    ref,
+                    script_build_id=binding.script_build_id,
+                    task_id=request.task_id,
+                    attempt_id=request.attempt_id,
+                )
+                if isinstance(version.artifact, EvidenceRecordV1):
+                    payload["current_evidence_ref"] = _ref_dict(ref)
+                    payload["current_evidence"] = _json_values(
+                        version.artifact.content_payload()
+                    )
+        return payload
 
     async def deterministic_precheck(self, context: Mapping[str, Any]) -> dict[str, Any]:
         from agent.orchestration.validation_policy import ValidationContext
@@ -918,7 +957,15 @@ def _json_values(value: Any) -> Any:
 
 
 def _safe_text_tuple(values: Sequence[str]) -> tuple[str, ...]:
-    return tuple(redact_text(str(value)) for value in values)
+    normalized: list[str] = []
+    seen: set[str] = set()
+    for value in values:
+        safe_value = redact_text(str(value))
+        if safe_value in seen:
+            continue
+        seen.add(safe_value)
+        normalized.append(safe_value)
+    return tuple(normalized)
 
 
 def _mapping_payload(

script_build_host/src/script_build_host/tools/registry.py

@@ -27,6 +27,9 @@ TASK_PRESET_BY_KIND = {
     "root-delivery": "script_root_worker",
 }
 
+_MAX_STRUCTURED_ARGUMENT_BYTES = 2 * 1024 * 1024
+_MAX_STRUCTURED_LIST_ITEMS = 1_000
+
 
 def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGateway) -> None:
     """Install Host wrappers into the supplied Runner registry.
@@ -35,6 +38,23 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     isolated tests deterministic while production creates only one composition.
     """
 
+    async def complete_retrieval(
+        source_type: str,
+        tool_name: str,
+        query: dict[str, Any],
+        context: dict[str, Any],
+    ) -> ToolResult:
+        evidence = await gateway.retrieve(source_type, tool_name, query, context)
+        submission = await gateway.submit_current_attempt(context)
+        result_json = _json({"evidence": evidence, "submission": submission})
+        return ToolResult(
+            title="Evidence frozen and attempt submitted",
+            output=result_json,
+            long_term_memory="The unique retrieval Evidence was frozen and submitted",
+            terminate_run=True,
+            result_summary=result_json,
+        )
+
     async def read_input_snapshot(context: dict[str, Any] | None = None) -> str:
         """Read the immutable input snapshot owned by this mission."""
 
@@ -63,6 +83,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str | ToolResult:
         """Validate and atomically create Tasks from complete business contracts."""
 
+        contracts = _structured_object_list(contracts, "contracts", max_items=50)
         return _json(
             await gateway.plan_script_tasks(
                 contracts=contracts,
@@ -71,7 +92,12 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             )
         )
 
-    _register(registry, plan_script_tasks, capabilities=["task_control"])
+    _register(
+        registry,
+        plan_script_tasks,
+        capabilities=["task_control"],
+        schema=_plan_script_tasks_schema(),
+    )
 
     async def decide_script_task(
         task_id: str,
@@ -95,13 +121,20 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             "cancel",
         }:
             raise ValueError("action is not supported by decide_script_task")
+        if replacement_contract is not None:
+            replacement_contract = _structured_object(
+                replacement_contract, "replacement_contract"
+            )
+        child_contracts = _structured_object_list(
+            child_contracts or [], "child_contracts", max_items=50
+        )
         result = await gateway.decide_script_task(
             task_id=task_id,
             action=action,
             reason=reason,
             validation_id=validation_id,
             replacement_contract=replacement_contract,
-            child_contracts=child_contracts or [],
+            child_contracts=child_contracts,
             context=context or {},
         )
         result_json = _json(result)
@@ -120,12 +153,17 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             )
         return result_json
 
-    _register(registry, decide_script_task, capabilities=["task_control"])
+    _register(
+        registry,
+        decide_script_task,
+        capabilities=["task_control"],
+        schema=_decide_script_task_schema(),
+    )
 
     async def inspect_script_plan(
         context: dict[str, Any] | None = None,
     ) -> str:
-        """Inspect a bounded business view of the current Task tree and contracts."""
+        """Inspect Tasks; copy complete accepted_decision_ref objects into new contracts."""
 
         return _json(await gateway.inspect_script_plan(context=context or {}))
 
@@ -144,16 +182,16 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, dispatch_script_tasks, capabilities=["task_control"])
 
-    async def query_pattern_qa(message: str, context: dict[str, Any] | None = None) -> str:
+    async def query_pattern_qa(
+        message: str, context: dict[str, Any] | None = None
+    ) -> ToolResult:
         """Query the configured Pattern service and freeze its evidence."""
 
-        return _json(
-            await gateway.retrieve(
-                "pattern",
-                "query_pattern_qa",
-                {"message": message},
-                context or {},
-            )
+        return await complete_retrieval(
+            "pattern",
+            "query_pattern_qa",
+            {"message": message},
+            context or {},
         )
 
     _register(registry, query_pattern_qa, capabilities=["external_send", "write"])
@@ -165,7 +203,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         match_fields: dict[str, str] | None = None,
         top_k: int = 3,
         context: dict[str, Any] | None = None,
-    ) -> str:
+    ) -> ToolResult:
         """Search the frozen script-decode index and freeze ranked evidence."""
 
         if return_field not in {"主脉络", "元素", "主题", "形式", "作用", "感受"}:
@@ -174,25 +212,25 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             raise ValueError("keyword and account_name cannot both be empty")
         if not 1 <= top_k <= 5:
             raise ValueError("top_k must be between 1 and 5")
+        if match_fields is not None:
+            match_fields = _structured_object(match_fields, "match_fields")
         valid_columns = {"主题", "形式", "作用", "感受"}
         valid_subfields = {"维度名", "维度值"}
         if match_fields and (
             set(match_fields) - valid_columns or set(match_fields.values()) - valid_subfields
         ):
             raise ValueError("match_fields contains an unsupported column or subfield")
-        return _json(
-            await gateway.retrieve(
-                "decode",
-                "search_script_decode_case",
-                {
-                    "return_field": return_field,
-                    "keyword": keyword,
-                    "account_name": account_name,
-                    "match_fields": match_fields or {},
-                    "top_k": top_k,
-                },
-                context or {},
-            )
+        return await complete_retrieval(
+            "decode",
+            "search_script_decode_case",
+            {
+                "return_field": return_field,
+                "keyword": keyword,
+                "account_name": account_name,
+                "match_fields": match_fields or {},
+                "top_k": top_k,
+            },
+            context or {},
         )
 
     _register(
@@ -207,24 +245,22 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         platform_channel: str = "xhs",
         max_count: int = 5,
         context: dict[str, Any] | None = None,
-    ) -> str:
+    ) -> ToolResult:
         """Search the allowlisted external service without legacy log writes."""
 
         if platform_channel not in {"xhs", "zhihu"}:
             raise ValueError("platform_channel must be xhs or zhihu")
         if not 1 <= max_count <= 20:
             raise ValueError("max_count must be between 1 and 20")
-        return _json(
-            await gateway.retrieve(
-                "external",
-                "external_search_case",
-                {
-                    "keyword": keyword,
-                    "platform_channel": platform_channel,
-                    "max_count": max_count,
-                },
-                context or {},
-            )
+        return await complete_retrieval(
+            "external",
+            "external_search_case",
+            {
+                "keyword": keyword,
+                "platform_channel": platform_channel,
+                "max_count": max_count,
+            },
+            context or {},
         )
 
     _register(
@@ -245,18 +281,16 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         keyword: str,
         max_count: int = 3,
         context: dict[str, Any] | None = None,
-    ) -> str:
+    ) -> ToolResult:
         """Search the frozen internal knowledge corpus and freeze evidence."""
 
         if not 1 <= max_count <= 20:
             raise ValueError("max_count must be between 1 and 20")
-        return _json(
-            await gateway.retrieve(
-                "knowledge",
-                "search_knowledge",
-                {"keyword": keyword, "max_count": max_count},
-                context or {},
-            )
+        return await complete_retrieval(
+            "knowledge",
+            "search_knowledge",
+            {"keyword": keyword, "max_count": max_count},
+            context or {},
         )
 
     _register(
@@ -279,6 +313,9 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Freeze one direction candidate inside the protected Attempt."""
 
+        goals = _structured_object_list(goals, "goals")
+        criteria = _structured_object_list(criteria, "criteria")
+        domain_criteria = _structured_object_list(domain_criteria or [], "domain_criteria")
         return _json(
             await gateway.save_direction_candidate(
                 goals=goals,
@@ -293,7 +330,12 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             )
         )
 
-    _register(registry, save_direction_candidate, capabilities=["write"])
+    _register(
+        registry,
+        save_direction_candidate,
+        capabilities=["write"],
+        schema=_save_direction_candidate_schema(),
+    )
 
     async def read_accepted_input_bundle(
         context: dict[str, Any] | None = None,
@@ -358,6 +400,13 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Create one Paragraph inside the protected Attempt workspace."""
 
+        content_range = _structured_object(content_range, "content_range")
+        theme_elements = _structured_object_list(theme_elements or [], "theme_elements")
+        form_elements = _structured_object_list(form_elements or [], "form_elements")
+        function_elements = _structured_object_list(
+            function_elements or [], "function_elements"
+        )
+        feeling_elements = _structured_object_list(feeling_elements or [], "feeling_elements")
         return _json(
             await gateway.candidate_command(
                 "create_script_paragraph",
@@ -386,6 +435,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Append unique atoms to one allowlisted Paragraph column."""
 
+        atoms = _structured_object_list(atoms, "atoms")
         return _json(
             await gateway.candidate_command(
                 "append_paragraph_atoms",
@@ -426,6 +476,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Atomically validate and update a Paragraph batch in this workspace."""
 
+        updates = _structured_object_list(updates, "updates")
         return _json(
             await gateway.candidate_command(
                 "batch_update_script_paragraphs", updates, context or {}
@@ -446,6 +497,15 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Create one Element inside the protected Attempt workspace."""
 
+        if commonality_analysis is not None:
+            commonality_analysis = _structured_object(
+                commonality_analysis, "commonality_analysis"
+            )
+        if topic_support is not None:
+            topic_support = _structured_object(topic_support, "topic_support")
+        if weight_score is not None:
+            weight_score = _structured_object(weight_score, "weight_score")
+        support_elements = _structured_object_list(support_elements or [], "support_elements")
         return _json(
             await gateway.candidate_command(
                 "create_script_element",
@@ -471,6 +531,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Update one Element with an allowlisted, workspace-local patch."""
 
+        updates = _structured_object(updates, "updates")
         return _json(
             await gateway.candidate_command(
                 "update_script_element",
@@ -487,6 +548,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Atomically link active local Paragraphs and Elements."""
 
+        links = _structured_object_list(links, "links")
         return _json(
             await gateway.candidate_command("batch_link_paragraph_elements", links, context or {})
         )
@@ -499,6 +561,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Atomically remove exact Paragraph/Element links from this workspace."""
 
+        links = _structured_object_list(links, "links")
         return _json(
             await gateway.candidate_command("delete_paragraph_element_links", links, context or {})
         )
@@ -513,6 +576,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Freeze one comparison over the candidates pinned by the Task contract."""
 
+        criterion_results = _structured_object_list(criterion_results, "criterion_results")
         return _json(
             await gateway.candidate_command(
                 "save_comparison_candidate",
@@ -546,6 +610,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> str:
         """Freeze governance closure with one adopted StructuredScript."""
 
+        unresolved_defects = _structured_object_list(unresolved_defects, "unresolved_defects")
         return _json(
             await gateway.candidate_command(
                 "save_candidate_portfolio",
@@ -633,6 +698,23 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     ) -> ToolResult:
         """Submit bounded criterion results plus structured, evidence-bound defects."""
 
+        criterion_results = _structured_object_list(criterion_results, "criterion_results")
+        defects = _structured_object_list(defects, "defects")
+        criterion_results = [
+            {
+                **item,
+                "reason": str(item.get("reason", ""))[:300],
+            }
+            for item in criterion_results
+        ]
+        defects = [
+            {
+                **item,
+                "observed_excerpt": str(item.get("observed_excerpt", ""))[:500],
+            }
+            for item in defects
+        ]
+        recommendation = recommendation[:200]
         result = await gateway.submit_structured_validation(
             verdict=verdict,
             criterion_results=criterion_results,
@@ -651,7 +733,12 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             result_summary=result_json,
         )
 
-    _register(registry, submit_validation, capabilities=["validation_submit"])
+    _register(
+        registry,
+        submit_validation,
+        capabilities=["validation_submit"],
+        schema=_submit_validation_schema(),
+    )
 
 
 def _register(
@@ -707,6 +794,404 @@ def _decode_schema() -> dict[str, Any]:
     }
 
 
+def _plan_script_tasks_schema() -> dict[str, Any]:
+    return {
+        "type": "function",
+        "function": {
+            "name": "plan_script_tasks",
+            "description": (
+                "Create bounded Tasks. Every contracts item must be a complete "
+                "script-task-contract/v1 object; do not send preset or display fields."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "contracts": {
+                        "type": "array",
+                        "items": _task_contract_schema(),
+                        "minItems": 1,
+                        "maxItems": 50,
+                    },
+                    "parent_task_id": {"type": ["string", "null"], "default": None},
+                },
+                "required": ["contracts"],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
+def _decide_script_task_schema() -> dict[str, Any]:
+    return {
+        "type": "function",
+        "function": {
+            "name": "decide_script_task",
+            "description": (
+                "Apply one validated Planner decision. Replacement and child contracts, "
+                "when required by the action, must be complete script-task-contract/v1 objects."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "task_id": {"type": "string"},
+                    "action": {
+                        "type": "string",
+                        "enum": [
+                            "accept",
+                            "repair",
+                            "retry",
+                            "revise",
+                            "split",
+                            "block",
+                            "supersede",
+                            "cancel",
+                        ],
+                    },
+                    "reason": {"type": "string"},
+                    "validation_id": {"type": ["string", "null"], "default": None},
+                    "replacement_contract": {
+                        "oneOf": [_task_contract_schema(), {"type": "null"}],
+                        "default": None,
+                    },
+                    "child_contracts": {
+                        "type": "array",
+                        "items": _task_contract_schema(),
+                        "maxItems": 50,
+                        "default": [],
+                    },
+                },
+                "required": ["task_id", "action", "reason"],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
+def _task_contract_schema() -> dict[str, Any]:
+    task_kinds = [
+        "direction",
+        "pattern-retrieval",
+        "decode-retrieval",
+        "external-retrieval",
+        "knowledge-retrieval",
+        "structure",
+        "paragraph",
+        "element-set",
+        "compare",
+        "compose",
+        "candidate-portfolio",
+    ]
+    artifact_ref = {
+        "type": "object",
+        "properties": {
+            "uri": {"type": "string"},
+            "kind": {"type": "string"},
+            "version": {"type": ["string", "null"]},
+            "digest": {"type": ["string", "null"]},
+            "summary": {"type": ["string", "null"]},
+            "metadata": {"type": "object"},
+        },
+        "required": ["uri", "kind"],
+        "additionalProperties": False,
+    }
+    decision_ref = {
+        "type": "object",
+        "properties": {
+            "decision_id": {"type": "string"},
+            "artifact_ref": artifact_ref,
+            "scope_ref": {
+                "type": "string",
+                "description": "A script-build:// control URI from the accepted decision.",
+            },
+            "expected_task_kind": {"type": "string", "enum": task_kinds},
+        },
+        "required": ["decision_id", "artifact_ref", "scope_ref", "expected_task_kind"],
+        "additionalProperties": False,
+    }
+    return {
+        "type": "object",
+        "properties": {
+            "schema_version": {
+                "type": "string",
+                "const": "script-task-contract/v1",
+            },
+            "task_kind": {"type": "string", "enum": task_kinds},
+            "scope_ref": {
+                "type": "string",
+                "description": "Stable logical scope URI beginning with script-build://.",
+            },
+            "intent_class": {
+                "type": "string",
+                "enum": [
+                    "explore",
+                    "expand",
+                    "replace",
+                    "repair",
+                    "compare",
+                    "compose",
+                    "portfolio",
+                ],
+            },
+            "objective": {"type": "string", "maxLength": 2000},
+            "input_decision_refs": {"type": "array", "items": decision_ref},
+            "base_artifact_ref": {
+                "oneOf": [artifact_ref, {"type": "null"}],
+            },
+            "write_scope": {
+                "type": "array",
+                "items": {
+                    "type": "string",
+                    "description": "A bounded script-build:// control URI.",
+                },
+            },
+            "gap_ref": {"type": ["string", "null"]},
+            "output_schema": {
+                "type": "string",
+                "enum": [
+                    "script-direction/v1",
+                    "evidence-record/v1",
+                    "structure-artifact/v1",
+                    "paragraph-artifact/v1",
+                    "paragraph-patch/v1",
+                    "element-set-artifact/v1",
+                    "element-set-patch/v1",
+                    "comparison-artifact/v1",
+                    "structured-script/v1",
+                    "candidate-portfolio/v1",
+                ],
+                "description": "Must match task_kind; use evidence-record/v1 for retrieval.",
+            },
+            "criteria": {
+                "type": "array",
+                "minItems": 1,
+                "maxItems": 16,
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "criterion_id": {"type": "string"},
+                        "description": {"type": "string"},
+                        "hard": {"type": "boolean"},
+                    },
+                    "required": ["criterion_id", "description", "hard"],
+                    "additionalProperties": False,
+                },
+            },
+            "budget": {
+                "type": "object",
+                "properties": {
+                    "max_attempts": {"type": "integer", "minimum": 1, "maximum": 4},
+                    "max_tokens": {
+                        "type": "integer",
+                        "minimum": 500000,
+                        "maximum": 1000000,
+                    },
+                    "max_seconds": {"type": "integer", "minimum": 1, "maximum": 900},
+                    "max_external_queries": {
+                        "type": "integer",
+                        "minimum": 1,
+                        "maximum": 40,
+                    },
+                    "max_no_improvement": {
+                        "type": "integer",
+                        "minimum": 1,
+                        "maximum": 3,
+                    },
+                },
+                "required": [
+                    "max_attempts",
+                    "max_tokens",
+                    "max_seconds",
+                    "max_external_queries",
+                    "max_no_improvement",
+                ],
+                "additionalProperties": False,
+            },
+            "supersedes_decision_ids": {"type": "array", "items": {"type": "string"}},
+            "candidate_closure_decision_refs": {"type": "array", "items": decision_ref},
+            "adopted_decision_ids": {"type": "array", "items": {"type": "string"}},
+            "held_or_rejected_decision_ids": {
+                "type": "array",
+                "items": {"type": "string"},
+            },
+            "compose_order": {"type": "array", "items": {"type": "string"}},
+            "comparison_decision_refs": {"type": "array", "items": decision_ref},
+        },
+        "required": [
+            "schema_version",
+            "task_kind",
+            "scope_ref",
+            "intent_class",
+            "objective",
+            "input_decision_refs",
+            "base_artifact_ref",
+            "write_scope",
+            "gap_ref",
+            "output_schema",
+            "criteria",
+            "budget",
+            "supersedes_decision_ids",
+            "candidate_closure_decision_refs",
+            "adopted_decision_ids",
+            "held_or_rejected_decision_ids",
+            "compose_order",
+            "comparison_decision_refs",
+        ],
+        "additionalProperties": False,
+    }
+
+
+def _save_direction_candidate_schema() -> dict[str, Any]:
+    criterion = {
+        "type": "object",
+        "properties": {
+            "criterion_id": {"type": "string"},
+            "description": {"type": "string"},
+        },
+        "required": ["criterion_id", "description"],
+        "additionalProperties": False,
+    }
+    return {
+        "type": "function",
+        "function": {
+            "name": "save_direction_candidate",
+            "description": (
+                "Freeze one Direction. evidence_refs must be exact artifact_ref.uri values "
+                "from accepted retrieval children; never invent a reference."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "goals": {
+                        "type": "array",
+                        "minItems": 1,
+                        "maxItems": 3,
+                        "items": {
+                            "type": "object",
+                            "properties": {
+                                "goal_id": {"type": "string"},
+                                "statement": {"type": "string"},
+                                "rationale": {"type": "string"},
+                            },
+                            "required": ["goal_id", "statement", "rationale"],
+                            "additionalProperties": False,
+                        },
+                    },
+                    "criteria": {"type": "array", "items": criterion},
+                    "domain_criteria": {"type": "array", "items": criterion, "default": []},
+                    "topic_refs": {"type": "array", "items": {"type": "string"}, "default": []},
+                    "persona_refs": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "default": [],
+                    },
+                    "strategy_refs": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "default": [],
+                    },
+                    "evidence_refs": {
+                        "type": "array",
+                        "minItems": 1,
+                        "items": {"type": "string"},
+                    },
+                    "legacy_markdown": {"type": "string"},
+                },
+                "required": ["goals", "criteria", "evidence_refs", "legacy_markdown"],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
+def _submit_validation_schema() -> dict[str, Any]:
+    artifact_ref = {
+        "type": "object",
+        "properties": {
+            "uri": {"type": "string"},
+            "kind": {"type": "string"},
+            "version": {"type": ["string", "null"]},
+            "digest": {"type": ["string", "null"]},
+        },
+        "required": ["uri", "kind"],
+        "additionalProperties": False,
+    }
+    defect = {
+        "type": "object",
+        "properties": {
+            "defect_code": {"type": "string"},
+            "criterion_id": {"type": "string"},
+            "scope_ref": {"type": "string"},
+            "observed_excerpt": {"type": "string", "maxLength": 500},
+            "evidence_refs": {"type": "array", "items": artifact_ref},
+            "severity": {"type": "string", "enum": ["warning", "hard", "critical"]},
+            "invalidated_inputs": {"type": "array", "items": {"type": "string"}},
+            "recommended_action_class": {
+                "type": "string",
+                "enum": [
+                    "repair",
+                    "retry",
+                    "revise",
+                    "split",
+                    "retrieve",
+                    "replace",
+                    "compare",
+                    "block",
+                ],
+            },
+        },
+        "required": [
+            "defect_code",
+            "criterion_id",
+            "scope_ref",
+            "observed_excerpt",
+            "evidence_refs",
+            "severity",
+            "invalidated_inputs",
+            "recommended_action_class",
+        ],
+        "additionalProperties": False,
+    }
+    return {
+        "type": "function",
+        "function": {
+            "name": "submit_validation",
+            "description": (
+                "Submit every Task criterion exactly once. Passed reports have no hard or "
+                "critical defects; failed/inconclusive reports must contain defects."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "verdict": {
+                        "type": "string",
+                        "enum": ["passed", "failed", "inconclusive"],
+                    },
+                    "criterion_results": {
+                        "type": "array",
+                        "items": {
+                            "type": "object",
+                            "properties": {
+                                "criterion_id": {"type": "string"},
+                                "verdict": {
+                                    "type": "string",
+                                    "enum": ["passed", "failed", "inconclusive"],
+                                },
+                                "reason": {"type": "string", "maxLength": 300},
+                            },
+                            "required": ["criterion_id", "verdict", "reason"],
+                            "additionalProperties": False,
+                        },
+                    },
+                    "defects": {"type": "array", "items": defect},
+                    "recommendation": {"type": "string", "maxLength": 200, "default": ""},
+                },
+                "required": ["verdict", "criterion_results", "defects"],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
 def _external_schema() -> dict[str, Any]:
     return {
         "type": "function",
@@ -754,6 +1239,47 @@ def _json(value: Any) -> str:
     return json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
 
 
+def _structured_object(value: Any, label: str) -> dict[str, Any]:
+    normalized = _decode_bounded_structured_value(value, label)
+    if not isinstance(normalized, dict):
+        raise ValueError(f"{label} must be a JSON object")
+    return normalized
+
+
+def _structured_object_list(
+    value: Any,
+    label: str,
+    *,
+    max_items: int = _MAX_STRUCTURED_LIST_ITEMS,
+) -> list[dict[str, Any]]:
+    normalized = _decode_bounded_structured_value(value, label)
+    if not isinstance(normalized, list):
+        raise ValueError(f"{label} must be a JSON array")
+    if len(normalized) > max_items:
+        raise ValueError(f"{label} exceeds the item limit")
+    return [
+        _structured_object(item, f"{label}[{index}]")
+        for index, item in enumerate(normalized)
+    ]
+
+
+def _decode_bounded_structured_value(value: Any, label: str) -> Any:
+    if isinstance(value, str):
+        if len(value.encode("utf-8")) > _MAX_STRUCTURED_ARGUMENT_BYTES:
+            raise ValueError(f"{label} exceeds the byte limit")
+        try:
+            value = json.loads(value)
+        except json.JSONDecodeError as exc:
+            raise ValueError(f"{label} must contain valid JSON") from exc
+    try:
+        encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"{label} must be JSON-serializable") from exc
+    if len(encoded.encode("utf-8")) > _MAX_STRUCTURED_ARGUMENT_BYTES:
+        raise ValueError(f"{label} exceeds the byte limit")
+    return value
+
+
 def _validated_images(
     values: Any,
 ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:

script_build_host/tests/test_agent_surface.py

@@ -77,8 +77,12 @@ def test_script_presets_are_exact_and_never_expose_legacy_control_tools() -> Non
     portfolio = get_preset("script_candidate_portfolio_worker")
     assert "save_candidate_portfolio" in (portfolio.allowed_tools or [])
     assert "save_structured_script_candidate" not in (portfolio.allowed_tools or [])
-    for validator_name in ("script_retrieval_validator", "script_candidate_validator"):
-        assert "view_frozen_images" in (get_preset(validator_name).allowed_tools or [])
+    assert "view_frozen_images" not in (
+        get_preset("script_retrieval_validator").allowed_tools or []
+    )
+    assert "view_frozen_images" in (
+        get_preset("script_candidate_validator").allowed_tools or []
+    )
 
     assert set(get_preset("script_root_worker").allowed_tools or []) == {
         "read_input_snapshot",

script_build_host/tests/test_phase_two_agent_tools.py

@@ -8,6 +8,7 @@ from typing import Any
 import pytest
 from agent import ToolRegistry
 from agent.orchestration import ArtifactRef, ValidationVerdict
+from agent.orchestration.evidence import EvidenceResponse
 
 from script_build_host.domain.artifacts import ArtifactState
 from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation
@@ -139,8 +140,9 @@ class _CountingRetrievalAdapter:
     async def retrieve(self, **_kwargs: Any) -> RetrievalResult:
         self.calls += 1
         return RetrievalResult(
-            source_refs=("decode://case/1",),
+            source_refs=("decode://case/1", "decode://case/1"),
             summary="one immutable result",
+            supports=("opening", "opening"),
         )
 
 
@@ -190,6 +192,65 @@ def _gateway(ref: ArtifactRef) -> tuple[LegacyScriptToolGateway, _Coordinator, _
     return gateway, coordinator, candidates
 
 
+@pytest.mark.asyncio
+async def test_validation_query_includes_exact_current_business_artifact() -> None:
+    ref = ArtifactRef(
+        uri="script-build://artifact-versions/9",
+        kind="paragraph",
+        version="9",
+        digest="sha256:" + "c" * 64,
+    )
+    gateway, coordinator, _candidates = _gateway(ref)
+
+    async def query_evidence(_context: Mapping[str, Any], _request: Any) -> EvidenceResponse:
+        return EvidenceResponse(items=[], evidence_refs=[], queries_used=1, queries_remaining=4)
+
+    coordinator.query_evidence = query_evidence  # type: ignore[attr-defined]
+
+    class Artifacts:
+        async def read_by_ref(self, _ref: ArtifactRef, **_owners: Any) -> Any:
+            return SimpleNamespace(
+                artifact=SimpleNamespace(
+                    content_payload=lambda: {
+                        "schema_version": "paragraph-artifact/v1",
+                        "paragraph_key": "opening",
+                        "text": "grounded opening",
+                    }
+                )
+            )
+
+    gateway.artifacts = Artifacts()  # type: ignore[assignment]
+    result = await gateway.query_validation_evidence(
+        "current candidate",
+        5,
+        {
+            "root_trace_id": "root-a",
+            "task_id": "task-1",
+            "attempt_id": "attempt-a",
+            "snapshot_id": "snapshot-a",
+            "validation_id": "validation-a",
+        },
+    )
+
+    assert result["current_artifacts"] == [
+        {
+            "artifact_ref": {
+                "uri": ref.uri,
+                "kind": ref.kind,
+                "version": ref.version,
+                "digest": ref.digest,
+                "summary": ref.summary,
+                "metadata": ref.metadata,
+            },
+            "artifact": {
+                "schema_version": "paragraph-artifact/v1",
+                "paragraph_key": "opening",
+                "text": "grounded opening",
+            },
+        }
+    ]
+
+
 @pytest.mark.asyncio
 async def test_attempt_submission_is_derived_from_protected_attempt_artifact() -> None:
     ref = ArtifactRef(
@@ -269,6 +330,8 @@ async def test_one_retrieval_attempt_never_calls_adapter_twice() -> None:
         "spec_version": 1,
     }
     await gateway.retrieve("decode", "search_script_decode_case", {"query": "opening"}, context)
+    assert artifacts.version.artifact.source_refs == ("decode://case/1",)
+    assert artifacts.version.artifact.supports == ("opening",)
     with pytest.raises(ProtocolViolation, match="only one Evidence"):
         await gateway.retrieve(
             "decode", "search_script_decode_case", {"query": "different"}, context

script_build_host/tests/test_tool_argument_normalization.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from agent import ToolRegistry
+
+from script_build_host.tools.registry import (
+    _structured_object,
+    _structured_object_list,
+)
+
+
+def test_structured_arguments_accept_native_and_provider_encoded_json() -> None:
+    assert _structured_object({"name": "native"}, "payload") == {"name": "native"}
+    assert _structured_object('{"name":"encoded"}', "payload") == {"name": "encoded"}
+    assert _structured_object_list('[{"index":1},{"index":2}]', "items") == [
+        {"index": 1},
+        {"index": 2},
+    ]
+    assert _structured_object_list(['{"index":1}'], "items") == [{"index": 1}]
+
+
+@pytest.mark.parametrize("value", ["not-json", "123", '{"not":"an-array"}'])
+def test_structured_object_list_rejects_non_object_arrays(value: str) -> None:
+    with pytest.raises(ValueError):
+        _structured_object_list(value, "items")
+
+
+def test_structured_arguments_enforce_item_and_byte_limits() -> None:
+    with pytest.raises(ValueError, match="item limit"):
+        _structured_object_list([{}, {}], "items", max_items=1)
+    with pytest.raises(ValueError, match="byte limit"):
+        _structured_object({"value": "x" * (2 * 1024 * 1024)}, "payload")
+
+
+def test_planner_schema_exposes_the_complete_contract_shape() -> None:
+    from script_build_host.tools.registry import register_script_tools
+
+    registry = ToolRegistry()
+    register_script_tools(registry, object())  # type: ignore[arg-type]
+
+    schema = registry.get_schemas(["plan_script_tasks"])[0]
+    contracts = schema["function"]["parameters"]["properties"]["contracts"]
+    contract = contracts["items"]
+
+    assert contracts["type"] == "array"
+    assert contract["properties"]["schema_version"]["const"] == "script-task-contract/v1"
+    assert "schema_version" in contract["required"]
+    assert contract["properties"]["task_kind"]["enum"] == [
+        "direction",
+        "pattern-retrieval",
+        "decode-retrieval",
+        "external-retrieval",
+        "knowledge-retrieval",
+        "structure",
+        "paragraph",
+        "element-set",
+        "compare",
+        "compose",
+        "candidate-portfolio",
+    ]
+    assert contract["additionalProperties"] is False
+
+
+@pytest.mark.asyncio
+async def test_retrieval_tool_freezes_and_submits_in_one_terminal_call() -> None:
+    class Gateway:
+        def __init__(self) -> None:
+            self.calls: list[tuple[str, Any]] = []
+
+        async def retrieve(self, source: str, tool: str, query: Any, context: Any) -> Any:
+            self.calls.append(("retrieve", (source, tool, query, context)))
+            return {"artifact_ref": {"uri": "script-build://artifact-versions/1"}}
+
+        async def submit_current_attempt(self, context: Any) -> Any:
+            self.calls.append(("submit", context))
+            return {"attempt_id": "attempt-1", "status": "awaiting_validation"}
+
+    from script_build_host.tools.registry import register_script_tools
+
+    gateway = Gateway()
+    registry = ToolRegistry()
+    register_script_tools(registry, gateway)  # type: ignore[arg-type]
+    tool = registry._tools["search_knowledge"]["func"]  # noqa: SLF001
+
+    result = await tool(keyword="hard tech", max_count=3, context={"task_id": "task-1"})
+
+    assert result.terminate_run is True
+    assert [name for name, _ in gateway.calls] == ["retrieve", "submit"]
+    assert "attempt-1" in result.output