feat(脚本工具): 暴露 create_script_paragraphs 专用批量入口

贯通工具端口、Gateway 分派、注册 schema 和 Structure preset。批量输入保持结构化边界并转发到候选服务，同时收紧 load_frozen_strategy 的工具说明，要求精确复制快照 URI。

script_build_host/src/script_build_host/agents/presets.py

@@ -111,6 +111,7 @@ def register_script_presets() -> None:
     paragraph_writes = [
         "read_attempt_workspace",
         "create_script_paragraph",
+        "create_script_paragraphs",
         "append_paragraph_atoms",
         "delete_paragraph_atom",
         "batch_update_script_paragraphs",

script_build_host/src/script_build_host/tools/contracts.py

@@ -81,6 +81,10 @@ class ScriptCandidateToolPort(Protocol):
         self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
     ) -> Mapping[str, Any]: ...
 
+    async def create_script_paragraphs(
+        self, *, paragraphs: Sequence[Mapping[str, Any]], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
     async def append_paragraph_atoms(
         self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
     ) -> Mapping[str, Any]: ...

script_build_host/src/script_build_host/tools/gateway.py

@@ -193,10 +193,7 @@ class LegacyScriptToolGateway:
         # 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._planner().dispatch_script_tasks(task_ids=task_ids, context=context),
-        )
+        return 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]:
         binding, snapshot = await self._scope(context)
@@ -478,6 +475,10 @@ class LegacyScriptToolGateway:
             return await tools.create_script_paragraph(
                 payload=_mapping_payload(payload), context=context
             )
+        if operation == "create_script_paragraphs":
+            return await tools.create_script_paragraphs(
+                paragraphs=_sequence_payload(payload), context=context
+            )
         if operation == "append_paragraph_atoms":
             return await tools.append_paragraph_atoms(
                 payload=_mapping_payload(payload), context=context
@@ -858,9 +859,7 @@ class LegacyScriptToolGateway:
                 )
                 if isinstance(version.artifact, EvidenceRecordV1):
                     payload["current_evidence_ref"] = _ref_dict(ref)
-                    payload["current_evidence"] = _json_values(
-                        version.artifact.content_payload()
-                    )
+                    payload["current_evidence"] = _json_values(version.artifact.content_payload())
         return payload
 
     async def deterministic_precheck(self, context: Mapping[str, Any]) -> dict[str, Any]:

script_build_host/src/script_build_host/tools/registry.py

@@ -63,7 +63,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     _register(registry, read_input_snapshot, capabilities=["read"])
 
     async def load_frozen_strategy(strategy_ref: str, context: dict[str, Any] | None = None) -> str:
-        """Load one on-demand strategy explicitly pinned by this Task."""
+        """Load an exact on-demand strategy URI copied from read_input_snapshot; never guess it."""
 
         return _json(await gateway.load_frozen_strategy(strategy_ref, context or {}))
 
@@ -122,9 +122,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         }:
             raise ValueError("action is not supported by decide_script_task")
         if replacement_contract is not None:
-            replacement_contract = _structured_object(
-                replacement_contract, "replacement_contract"
-            )
+            replacement_contract = _structured_object(replacement_contract, "replacement_contract")
         child_contracts = _structured_object_list(
             child_contracts or [], "child_contracts", max_items=50
         )
@@ -182,9 +180,7 @@ 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
-    ) -> ToolResult:
+    async def query_pattern_qa(message: str, context: dict[str, Any] | None = None) -> ToolResult:
         """Query the configured Pattern service and freeze its evidence."""
 
         return await complete_retrieval(
@@ -403,9 +399,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         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"
-        )
+        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(
@@ -427,6 +421,22 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, create_script_paragraph, capabilities=["write"])
 
+    async def create_script_paragraphs(
+        paragraphs: list[dict[str, Any]],
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Atomically create a parent-linked Paragraph batch using client keys."""
+
+        return _json(
+            await gateway.candidate_command(
+                "create_script_paragraphs",
+                [_structured_object(item, "paragraph") for item in paragraphs],
+                context or {},
+            )
+        )
+
+    _register(registry, create_script_paragraphs, capabilities=["write"])
+
     async def append_paragraph_atoms(
         paragraph_id: int,
         column: str,
@@ -498,9 +508,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         """Create one Element inside the protected Attempt workspace."""
 
         if commonality_analysis is not None:
-            commonality_analysis = _structured_object(
-                commonality_analysis, "commonality_analysis"
-            )
+            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:
@@ -1257,10 +1265,7 @@ def _structured_object_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)
-    ]
+    return [_structured_object(item, f"{label}[{index}]") for index, item in enumerate(normalized)]
 
 
 def _decode_bounded_structured_value(value: Any, label: str) -> Any: