工具协议：收敛方向比较与验证证据输入

Direction 使用 client_key 和 Evidence Decision，Compare 只推荐候选 Decision，Validator 只提交 evidence handle；Host 统一映射稳定业务 ID、Artifact 与受保护证据。移除 Agent 可见的物理 ID、URI、digest、旧 workspace 读写和低层修补工具，并增加静态表面审计。

script_build_host/src/script_build_host/tools/contracts.py

@@ -74,7 +74,7 @@ class ScriptCandidateToolPort(Protocol):
     async def view_frozen_images(
         self,
         *,
-        raw_artifact_refs: Sequence[str],
+        image_handles: Sequence[str],
         context: Mapping[str, Any],
     ) -> Sequence[Mapping[str, Any]]: ...
 
@@ -92,6 +92,14 @@ class ScriptCandidateToolPort(Protocol):
         self, *, paragraphs: Sequence[Mapping[str, Any]], context: Mapping[str, Any]
     ) -> Mapping[str, Any]: ...
 
+    async def save_script_paragraphs(
+        self,
+        *,
+        paragraphs: Sequence[Mapping[str, Any]],
+        expected_state_revision: str,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]: ...
+
     async def append_paragraph_atoms(
         self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
     ) -> Mapping[str, Any]: ...
@@ -108,6 +116,23 @@ class ScriptCandidateToolPort(Protocol):
         self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
     ) -> Mapping[str, Any]: ...
 
+    async def create_script_elements(
+        self,
+        *,
+        elements: Sequence[Mapping[str, Any]],
+        links: Sequence[Mapping[str, Any]],
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]: ...
+
+    async def save_script_elements(
+        self,
+        *,
+        elements: Sequence[Mapping[str, Any]],
+        links: Sequence[Mapping[str, Any]],
+        expected_state_revision: str,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]: ...
+
     async def update_script_element(
         self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
     ) -> Mapping[str, Any]: ...

script_build_host/src/script_build_host/tools/failures.py

@@ -9,30 +9,39 @@ from agent import FailureDetail, FailureDisposition
 
 from script_build_host.domain.errors import ScriptBuildError
 
-_REPLAN_CODES = frozenset({
-    "INPUT_SCOPE_MISMATCH",
-    "CHILD_DECISION_INVALID",
-    "PHASE_POLICY_VIOLATION",
-    "PHASE_TWO_BOUNDARY_NOT_READY",
-    "PHASE_THREE_BOUNDARY_NOT_READY",
-    "PROTOCOL_VIOLATION",
-    "TASK_CHILD_BLOCKED",
-    "TASK_BUDGET_EXCEEDED",
-    "ATTEMPT_WORKSPACE_FROZEN",
-    "WRITE_SCOPE_VIOLATION",
-    "SUPERSESSION_CYCLE",
-})
-_REPAIR_CODES = frozenset({
-    "LEGACY_WRITE_INVALID",
-    "LEGACY_REFERENCE_INVALID",
-    "ARTIFACT_NOT_FOUND",
-})
-_RETRY_CODES = frozenset({
-    "PUBLICATION_LOCK_TIMEOUT",
-    "PUBLICATION_DEADLOCK_RETRYABLE",
-    "STALE_BASE_REVISION",
-    "VALIDATION_EVIDENCE_INVALID",
-})
+_REPLAN_CODES = frozenset(
+    {
+        "INPUT_SCOPE_MISMATCH",
+        "CHILD_DECISION_INVALID",
+        "PHASE_POLICY_VIOLATION",
+        "PHASE_TWO_BOUNDARY_NOT_READY",
+        "PHASE_THREE_BOUNDARY_NOT_READY",
+        "PROTOCOL_VIOLATION",
+        "TASK_CHILD_BLOCKED",
+        "TASK_BUDGET_EXCEEDED",
+        "TASK_KIND_PRESET_MISMATCH",
+        "TASK_STATE_CONFLICT",
+        "ATTEMPT_WORKSPACE_FROZEN",
+        "WRITE_SCOPE_VIOLATION",
+        "SUPERSESSION_CYCLE",
+    }
+)
+_REPAIR_CODES = frozenset(
+    {
+        "LEGACY_WRITE_INVALID",
+        "LEGACY_REFERENCE_INVALID",
+        "ARTIFACT_NOT_FOUND",
+    }
+)
+_RETRY_CODES = frozenset(
+    {
+        "PUBLICATION_LOCK_TIMEOUT",
+        "PUBLICATION_DEADLOCK_RETRYABLE",
+        "STALE_BASE_REVISION",
+        "STALE_WORKBENCH_STATE",
+        "VALIDATION_EVIDENCE_INVALID",
+    }
+)
 _ABORT_MARKERS = (
     "STOP",
     "FENCING",
@@ -59,7 +68,7 @@ def classify_script_tool_failure(
             message=str(error),
             disposition=FailureDisposition.RETRY_CALL,
             source_tool=source_tool,
-            details=_identity_details(context),
+            details=_failure_details(error, context),
         )
 
     code = error.code
@@ -89,18 +98,25 @@ def classify_script_tool_failure(
         message=error.summary,
         disposition=disposition,
         source_tool=source_tool,
-        details=_identity_details(context),
+        details=_failure_details(error, context),
     )
 
 
-def _identity_details(context: Mapping[str, Any] | None) -> dict[str, str]:
-    if not isinstance(context, Mapping):
-        return {}
-    return {
-        key: str(context[key])
-        for key in ("task_id", "attempt_id", "validation_id", "operation_id")
-        if context.get(key) not in (None, "")
-    }
+def _failure_details(
+    error: ScriptBuildError | ValueError,
+    context: Mapping[str, Any] | None,
+) -> dict[str, Any]:
+    raw = getattr(error, "details", None)
+    details = dict(raw) if isinstance(raw, Mapping) else {}
+    if isinstance(context, Mapping):
+        details.update(
+            {
+                key: str(context[key])
+                for key in ("task_id", "attempt_id", "validation_id", "operation_id")
+                if context.get(key) not in (None, "")
+            }
+        )
+    return details
 
 
 __all__ = ["classify_script_tool_failure"]

script_build_host/src/script_build_host/tools/gateway.py

@@ -47,6 +47,13 @@ from script_build_host.domain.ports import (
     MissionBindingRepository,
     ScriptBusinessArtifactRepository,
 )
+from script_build_host.domain.task_contracts import stable_semantic_id
+from script_build_host.domain.workbench import (
+    protect_model_value,
+    semantic_handle,
+    strategy_handle,
+    strategy_ref,
+)
 from script_build_host.infrastructure.redaction import redact, redact_text
 
 from .contracts import RootDeliveryToolPort, ScriptCandidateToolPort, ScriptPlannerToolPort
@@ -112,6 +119,7 @@ class LegacyScriptToolGateway:
         candidate_tools: ScriptCandidateToolPort | None = None,
         budget_guard: PhaseTwoBudgetGuard | None = None,
         root_tools: RootDeliveryToolPort | None = None,
+        workbench: Any | None = None,
     ) -> None:
         self.bindings = bindings
         self.snapshots = snapshots
@@ -124,6 +132,7 @@ class LegacyScriptToolGateway:
         self.candidate_tools = candidate_tools
         self.budget_guard = budget_guard
         self.root_tools = root_tools
+        self.workbench = workbench
         self._active_retrieval_attempts: set[tuple[str, str, str]] = set()
 
     async def ensure_dispatch_allowed(self, context: Mapping[str, Any]) -> None:
@@ -190,6 +199,28 @@ class LegacyScriptToolGateway:
     async def inspect_script_plan(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
         return await self._planner().inspect_script_plan(context=context)
 
+    async def read_workbench_detail(
+        self,
+        *,
+        view: str,
+        handle: str | None,
+        known_revision: str | None,
+        cursor: int,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        if self.workbench is None:
+            raise ProtocolViolation("Mission Workbench is not configured")
+        return cast(
+            Mapping[str, Any],
+            await self.workbench.detail(
+                root_trace_id=_required(context, "root_trace_id"),
+                view=view,
+                handle=handle,
+                known_revision=known_revision,
+                cursor=cursor,
+            ),
+        )
+
     async def dispatch_script_tasks(
         self, *, task_ids: Sequence[str], context: Mapping[str, Any]
     ) -> Sequence[Mapping[str, Any]]:
@@ -216,7 +247,18 @@ class LegacyScriptToolGateway:
             {key: value for key, value in item.items() if key != "content"}
             for item in payload.get("prompt_manifest", [])
         ]
-        return cast(dict[str, Any], _json_values(payload))
+        view = await self._snapshot_view(context)
+        return cast(dict[str, Any], _json_values(_project_snapshot(payload, view=view)))
+
+    async def _snapshot_view(self, context: Mapping[str, Any]) -> str:
+        task_id = context.get("task_id")
+        if not isinstance(task_id, str) or not task_id.strip():
+            return "planner"
+        ledger = await self.coordinator.task_store.load(_required(context, "root_trace_id"))
+        task = ledger.tasks.get(task_id)
+        if task is None:
+            raise ProtocolViolation("protected Task is not in the mission ledger")
+        return task_kind(task.current_spec.context_refs) or "unknown"
 
     async def read_accepted_portfolio(self, context: Mapping[str, Any]) -> Mapping[str, Any]:
         return await self._root_tools().read_accepted_portfolio(context=context)
@@ -248,7 +290,11 @@ class LegacyScriptToolGateway:
         )
 
     async def load_frozen_strategy(
-        self, strategy_ref: str, context: Mapping[str, Any]
+        self,
+        selected_handle: str,
+        context: Mapping[str, Any],
+        *,
+        cursor: int = 0,
     ) -> dict[str, Any]:
         """Explicitly load one digest-verified on-demand strategy from the bound snapshot."""
 
@@ -263,21 +309,39 @@ class LegacyScriptToolGateway:
         } != {expected_input_ref}:
             raise ProtocolViolation("current task is not bound to the frozen InputSnapshot")
         matches: list[dict[str, Any]] = []
-        for item in snapshot.strategies:
-            expected = (
-                f"script-build://strategies/{item.get('strategy_id')}"
-                f"/versions/{item.get('version')}"
-            )
-            if expected == strategy_ref and item.get("mode") == "on_demand":
+        for index, item in enumerate(snapshot.strategies):
+            if (
+                strategy_handle(snapshot.snapshot_id, index, item) == selected_handle
+                and item.get("mode") == "on_demand"
+            ):
                 matches.append(item)
         if len(matches) != 1:
-            raise ProtocolViolation("strategy_ref is outside the frozen on-demand strategies")
+            raise ValueError("strategy_handle must select one frozen on-demand strategy")
         item = matches[0]
         content = item.get("content")
         digest = item.get("content_sha256")
         if not isinstance(content, str) or canonical_sha256(content).wire != digest:
             raise ProtocolViolation("frozen on-demand strategy digest does not match")
-        return cast(dict[str, Any], _json_values(item))
+        start = max(cursor, 0)
+        fragment = content[start : start + 7_000]
+        metadata = {
+            key: value
+            for key, value in item.items()
+            if key not in {"content", "content_sha256", "sha256", "source_uri"}
+        }
+        return cast(
+            dict[str, Any],
+            _json_values(
+                {
+                    "strategy_handle": selected_handle,
+                    "metadata": metadata,
+                    "content_fragment": fragment,
+                    "next_cursor": (
+                        start + len(fragment) if start + len(fragment) < len(content) else None
+                    ),
+                }
+            ),
+        )
 
     async def read_accepted_artifacts(self, context: Mapping[str, Any]) -> list[dict[str, Any]]:
         binding, _ = await self._scope(context)
@@ -447,10 +511,10 @@ class LegacyScriptToolGateway:
         return await self.image_adapter.load(urls=urls, snapshot=snapshot)
 
     async def view_frozen_images(
-        self, raw_artifact_refs: Sequence[str], context: Mapping[str, Any]
+        self, image_handles: Sequence[str], context: Mapping[str, Any]
     ) -> Sequence[Mapping[str, Any]]:
         return await self._candidates().view_frozen_images(
-            raw_artifact_refs=raw_artifact_refs,
+            image_handles=image_handles,
             context=context,
         )
 
@@ -476,41 +540,20 @@ class LegacyScriptToolGateway:
     ) -> Mapping[str, Any]:
         await self.ensure_dispatch_allowed(context)
         tools = self._candidates()
-        if operation == "create_script_paragraph":
-            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 == "save_script_paragraphs":
+            batch = _mapping_payload(payload)
+            return await tools.save_script_paragraphs(
+                paragraphs=_sequence_payload(batch.get("paragraphs", ())),
+                expected_state_revision=str(batch.get("expected_state_revision", "")),
+                context=context,
             )
-        if operation == "append_paragraph_atoms":
-            return await tools.append_paragraph_atoms(
-                payload=_mapping_payload(payload), context=context
-            )
-        if operation == "delete_paragraph_atom":
-            return await tools.delete_paragraph_atom(
-                payload=_mapping_payload(payload), context=context
-            )
-        if operation == "batch_update_script_paragraphs":
-            return await tools.batch_update_script_paragraphs(
-                updates=_sequence_payload(payload), context=context
-            )
-        if operation == "create_script_element":
-            return await tools.create_script_element(
-                payload=_mapping_payload(payload), context=context
-            )
-        if operation == "update_script_element":
-            return await tools.update_script_element(
-                payload=_mapping_payload(payload), context=context
-            )
-        if operation == "batch_link_paragraph_elements":
-            return await tools.batch_link_paragraph_elements(
-                links=_sequence_payload(payload), context=context
-            )
-        if operation == "delete_paragraph_element_links":
-            return await tools.delete_paragraph_element_links(
-                links=_sequence_payload(payload), context=context
+        if operation == "save_script_elements":
+            batch = _mapping_payload(payload)
+            return await tools.save_script_elements(
+                elements=_sequence_payload(batch.get("elements", ())),
+                links=_sequence_payload(batch.get("links", ())),
+                expected_state_revision=str(batch.get("expected_state_revision", "")),
+                context=context,
             )
         if operation == "save_comparison_candidate":
             return await tools.save_comparison_candidate(
@@ -624,22 +667,19 @@ class LegacyScriptToolGateway:
         allowed_values = [*snapshot.artifact_refs, *snapshot.evidence_refs]
         kind = task_kind(task.current_spec.context_refs)
         if kind in PHASE_TWO_KINDS:
-            evidence_reader = getattr(
-                self._candidates(), "validation_evidence_refs", None
-            )
+            evidence_reader = getattr(self._candidates(), "validation_evidence_refs", None)
             if callable(evidence_reader):
                 allowed_values.extend(await evidence_reader(context=context))
         elif kind == ROOT_DELIVERY_KIND:
-            evidence_reader = getattr(
-                self._root_tools(), "validation_evidence_refs", None
-            )
+            evidence_reader = getattr(self._root_tools(), "validation_evidence_refs", None)
             if callable(evidence_reader):
                 allowed_values.extend(await evidence_reader(context=context))
-        allowed_refs = {
-            (ref.uri, ref.version, ref.digest, ref.kind): ref
-            for ref in allowed_values
-        }
-        normalized_defects = _normalize_defects(defects, allowed_refs)
+        allowed_refs = {(ref.uri, ref.version, ref.digest, ref.kind): ref for ref in allowed_values}
+        refs_by_handle = {_evidence_handle(ref): ref for ref in allowed_refs.values()}
+        normalized_defects = _normalize_defects(
+            [_expand_validation_handles(item, refs_by_handle) for item in defects],
+            allowed_refs,
+        )
         expected_ids = {item.criterion_id for item in task.current_spec.acceptance_criteria}
         if any(item["criterion_id"] not in expected_ids for item in normalized_defects):
             raise ProtocolViolation("validation defect references an unknown Task criterion")
@@ -657,10 +697,16 @@ class LegacyScriptToolGateway:
             )
             if defect_codes:
                 reason = f"{reason}; defect_codes={','.join(defect_codes)}"[:500]
+            evidence_handles = raw.get("evidence_handle_ids", ())
+            if not isinstance(evidence_handles, list) or len(evidence_handles) > 64:
+                raise ProtocolViolation("criterion evidence handles are invalid")
             evidence = _normalize_evidence_refs(
-                raw.get("evidence_refs"),
+                [
+                    _ref_dict(_required_evidence_handle(value, refs_by_handle))
+                    for value in evidence_handles
+                ],
                 allowed_refs,
-                label="criterion evidence_refs",
+                label="criterion evidence_handle_ids",
             )
             if criterion_verdict is ValidationVerdict.PASSED and not evidence:
                 raise ProtocolViolation("passed criterion requires frozen evidence_refs")
@@ -679,9 +725,7 @@ class LegacyScriptToolGateway:
         criterion_verdicts = {item.verdict for item in result_by_id.values()}
         every_criterion_passed = criterion_verdicts == {ValidationVerdict.PASSED}
         if (overall is ValidationVerdict.PASSED) != every_criterion_passed:
-            raise ProtocolViolation(
-                "overall validation verdict must match all criterion verdicts"
-            )
+            raise ProtocolViolation("overall validation verdict must match all criterion verdicts")
         from agent.orchestration.validation_policy import ValidationContext
 
         attempt = ledger.attempts.get(_required(context, "attempt_id"))
@@ -778,27 +822,43 @@ class LegacyScriptToolGateway:
         goals: Sequence[Mapping[str, Any]],
         constraints: Sequence[Mapping[str, Any]],
         preferences: Sequence[Mapping[str, Any]],
-        topic_refs: Sequence[str],
-        persona_refs: Sequence[str],
-        strategy_refs: Sequence[str],
-        evidence_refs: Sequence[str],
+        strategy_handles: Sequence[str],
+        evidence_decision_ids: Sequence[str],
         context: Mapping[str, Any],
     ) -> dict[str, Any]:
         await self.ensure_dispatch_allowed(context)
-        binding, _ = await self._scope(context)
+        binding, snapshot = await self._scope(context)
         accepted = await self.read_accepted_artifacts(context)
-        allowed_evidence = {item["artifact_ref"]["uri"] for item in accepted}
-        if any(ref not in allowed_evidence for ref in evidence_refs):
+        evidence_by_decision = {
+            str(item["decision_id"]): str(item["artifact_ref"]["uri"]) for item in accepted
+        }
+        if len(set(evidence_decision_ids)) != len(evidence_decision_ids) or any(
+            item not in evidence_by_decision for item in evidence_decision_ids
+        ):
             raise ProtocolViolation(
                 "direction references evidence outside frozen child ACCEPT decisions"
             )
         task_id = _required(context, "task_id")
         direction_goals, goal_ids_by_client_key = _direction_goals(task_id, goals)
+        strategy_by_handle = {
+            strategy_handle(snapshot.snapshot_id, index, item): strategy_ref(
+                snapshot.snapshot_id, index, item
+            )
+            for index, item in enumerate(snapshot.strategies)
+        }
+        if len(set(strategy_handles)) != len(strategy_handles) or any(
+            item not in strategy_by_handle for item in strategy_handles
+        ):
+            raise ProtocolViolation("direction contains an unknown strategy handle")
         artifact = DirectionArtifact(
             goals=direction_goals,
             constraints=tuple(
                 DirectionConstraint(
-                    constraint_id=redact_text(str(item["constraint_id"])),
+                    constraint_id=stable_semantic_id(
+                        task_id,
+                        "constraint",
+                        _normalized_client_key(item.get("client_key"), "constraint client_key"),
+                    ),
                     statement=redact_text(str(item["statement"])),
                     rationale=redact_text(str(item.get("rationale", ""))),
                 )
@@ -806,17 +866,21 @@ class LegacyScriptToolGateway:
             ),
             preferences=tuple(
                 DirectionPreference(
-                    preference_id=redact_text(str(item["preference_id"])),
+                    preference_id=stable_semantic_id(
+                        task_id,
+                        "preference",
+                        _normalized_client_key(item.get("client_key"), "preference client_key"),
+                    ),
                     statement=redact_text(str(item["statement"])),
                     rationale=redact_text(str(item.get("rationale", ""))),
                     priority=(int(item["priority"]) if item.get("priority") is not None else None),
                 )
                 for item in preferences
             ),
-            topic_refs=_safe_text_tuple(topic_refs),
-            persona_refs=_safe_text_tuple(persona_refs),
-            strategy_refs=_safe_text_tuple(strategy_refs),
-            evidence_refs=tuple(evidence_refs),
+            topic_refs=(f"script-build://inputs/{snapshot.snapshot_id}/topic",),
+            persona_refs=(f"script-build://inputs/{snapshot.snapshot_id}/persona",),
+            strategy_refs=tuple(strategy_by_handle[item] for item in strategy_handles),
+            evidence_refs=tuple(evidence_by_decision[item] for item in evidence_decision_ids),
         )
         version, ref = await self.artifacts.freeze(
             script_build_id=binding.script_build_id,
@@ -849,40 +913,20 @@ class LegacyScriptToolGateway:
         )
         response = await self.coordinator.query_evidence(dict(context), request)
         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
+        refs = tuple(response.evidence_refs)
+        payload.pop("evidence_refs", None)
+        payload["evidence_handles"] = [
+            {
+                "evidence_handle": _evidence_handle(ref),
+                "kind": ref.kind,
+                "summary": ref.summary,
+            }
+            for ref in refs
+        ]
+        payload["items"] = protect_model_value(
+            payload.get("items", []),
+            namespace=f"validation:{request.validation_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]:
@@ -908,7 +952,9 @@ class LegacyScriptToolGateway:
             snapshot=snapshot,
             prior_validation_count=max(0, len(task.validation_ids) - 1),
         )
-        rules = [item.to_dict() for item in precheck_snapshot(validation_context)]
+        rules: list[Mapping[str, Any]] = [
+            item.to_dict() for item in precheck_snapshot(validation_context)
+        ]
         kind = task_kind(task.current_spec.context_refs)
         if kind in PHASE_TWO_KINDS:
             rules.extend(await self._candidates().deterministic_precheck(context=context))
@@ -941,6 +987,150 @@ class LegacyScriptToolGateway:
         return self.root_tools
 
 
+_SNAPSHOT_IDENTITY_FIELDS = (
+    "snapshot_id",
+    "script_build_id",
+    "execution_id",
+    "topic_build_id",
+    "topic_id",
+    "canonical_sha256",
+    "business_input_sha256",
+    "parent_snapshot_id",
+    "parent_snapshot_sha256",
+    "created_at",
+)
+_TOPIC_ITEM_FIELDS = (
+    "point_type",
+    "dimension",
+    "element_name",
+    "note",
+    "category_path",
+    "item_level",
+    "derivation_type",
+    "content",
+    "result",
+    "name",
+)
+
+
+def _project_snapshot(payload: Mapping[str, Any], *, view: str) -> dict[str, Any]:
+    """Project immutable inputs into a role-sized business view."""
+
+    persona_limits = {
+        "planner": 24,
+        "direction": 32,
+        "pattern-retrieval": 16,
+        "structure": 24,
+        "paragraph": 12,
+        "element-set": 12,
+        "root-delivery": 16,
+    }
+    persona = list(payload.get("persona_points") or ())
+    section_patterns = list(payload.get("section_patterns") or ())
+    output = {key: payload.get(key) for key in _SNAPSHOT_IDENTITY_FIELDS if key in payload}
+    output.update(
+        {
+            "view": view,
+            "account": payload.get("account") or {},
+            "topic": _project_topic(payload.get("topic")),
+            "strategies": list(payload.get("strategies") or ()),
+            "persona_points_total": len(persona),
+            "section_patterns_total": len(section_patterns),
+        }
+    )
+    limit = persona_limits.get(view, 16)
+    if limit:
+        output["persona_points"] = _select_persona_points(persona, limit=limit)
+    if view in {
+        "planner",
+        "direction",
+        "pattern-retrieval",
+        "structure",
+        "paragraph",
+        "element-set",
+        "root-delivery",
+    }:
+        output["section_patterns"] = section_patterns
+    return output
+
+
+def _project_topic(raw: Any) -> dict[str, Any]:
+    if not isinstance(raw, Mapping):
+        return {}
+    topic = raw.get("topic")
+    topic_row = dict(topic) if isinstance(topic, Mapping) else {}
+    projected: dict[str, Any] = {
+        "topic": {
+            key: topic_row[key]
+            for key in (
+                "result",
+                "topic_direction",
+                "topic_understanding",
+                "inspiration_evaluation_config",
+            )
+            if key in topic_row
+        }
+    }
+    for collection in ("composition_items", "points", "relations", "sources"):
+        values = raw.get(collection)
+        if not isinstance(values, Sequence) or isinstance(values, (str, bytes)):
+            continue
+        rows = []
+        for value in values:
+            if not isinstance(value, Mapping) or value.get("is_active") is False:
+                continue
+            rows.append({key: value[key] for key in _TOPIC_ITEM_FIELDS if key in value})
+        if rows:
+            projected[collection] = rows
+    return projected
+
+
+def _select_persona_points(values: Sequence[Any], *, limit: int) -> list[dict[str, Any]]:
+    groups: dict[str, list[Mapping[str, Any]]] = {}
+    for value in values:
+        if isinstance(value, Mapping):
+            groups.setdefault(str(value.get("列") or "其他"), []).append(value)
+    for rows in groups.values():
+        rows.sort(
+            key=lambda item: (
+                -_numeric(item.get("人设权重分")),
+                -_numeric(item.get("帖子覆盖率")),
+                str(item.get("点名称") or ""),
+            )
+        )
+    selected: list[dict[str, Any]] = []
+    group_names = sorted(groups)
+    while len(selected) < limit and group_names:
+        remaining = []
+        for name in group_names:
+            rows = groups[name]
+            if rows and len(selected) < limit:
+                selected.append(_project_persona_point(rows.pop(0)))
+            if rows:
+                remaining.append(name)
+        group_names = remaining
+    return selected
+
+
+def _project_persona_point(value: Mapping[str, Any]) -> dict[str, Any]:
+    output = {
+        key: value[key]
+        for key in ("列", "点名称", "点类型", "人设权重分", "帖子覆盖率")
+        if key in value
+    }
+    paths = value.get("分类路径")
+    if isinstance(paths, Sequence) and not isinstance(paths, (str, bytes)):
+        output["分类路径"] = [str(item)[:160] for item in paths[:2]]
+    return output
+
+
+def _numeric(value: Any) -> float:
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return 0.0
+
+
 def _required(context: Mapping[str, Any], key: str) -> str:
     value = context.get(key)
     if not isinstance(value, str) or not value.strip():
@@ -1031,9 +1221,7 @@ def _direction_goals(
     for item in goals:
         client_key = _normalized_client_key(item.get("client_key"), "goal client_key")
         parent_key = (
-            _normalized_client_key(
-                item.get("parent_client_key"), "goal parent_client_key"
-            )
+            _normalized_client_key(item.get("parent_client_key"), "goal parent_client_key")
             if item.get("parent_client_key") is not None
             else None
         )
@@ -1049,13 +1237,11 @@ def _direction_goals(
             raise ValueError(f"goal parent_client_key does not exist: {parent_key}")
     parent_by_key = {key: parent for key, parent, _ in normalized}
     if any(
-        parent is not None and parent_by_key.get(parent) is not None
-        for _, parent, _ in normalized
+        parent is not None and parent_by_key.get(parent) is not None for _, parent, _ in normalized
     ):
         raise ValueError("direction goal hierarchy may contain only two levels")
     goal_ids = {
-        key: "goal-" + sha256(f"{task_id}\0{key}".encode()).hexdigest()[:16]
-        for key in keys
+        key: "goal-" + sha256(f"{task_id}\0{key}".encode()).hexdigest()[:16] for key in keys
     }
     if len(set(goal_ids.values())) != len(goal_ids):
         raise ValueError("generated goal identity collision")
@@ -1084,6 +1270,61 @@ def _attempt_summary(ref: ArtifactRef, scope_ref: str) -> str:
     return f"kind={ref.kind};scope={scope};digest={digest};status=frozen"[:500]
 
 
+def _evidence_handle(ref: ArtifactRef) -> str:
+    return semantic_handle("src", ref.uri, ref.digest or ref.version or "")
+
+
+def _required_evidence_handle(
+    value: object, refs_by_handle: Mapping[str, ArtifactRef]
+) -> ArtifactRef:
+    handle = str(value or "").strip()
+    ref = refs_by_handle.get(handle)
+    if ref is None:
+        raise ProtocolViolation("validation references an unknown evidence handle")
+    return ref
+
+
+def _expand_validation_handles(
+    value: Mapping[str, Any], refs_by_handle: Mapping[str, ArtifactRef]
+) -> dict[str, Any]:
+    unknown = set(value) - {
+        "defect_code",
+        "criterion_id",
+        "scope_selector",
+        "observed_excerpt",
+        "evidence_handle_ids",
+        "severity",
+        "invalidated_inputs",
+        "recommended_action_class",
+    }
+    if unknown:
+        raise ProtocolViolation("validation defect contains unsupported fields")
+    selector = value.get("scope_selector")
+    if not isinstance(selector, Mapping) or selector.get("anchor") != "mission":
+        raise ProtocolViolation("validation defect scope_selector is invalid")
+    path = selector.get("path")
+    if (
+        not isinstance(path, list)
+        or not 1 <= len(path) <= 4
+        or any(not isinstance(item, str) or not item.strip() for item in path)
+    ):
+        raise ProtocolViolation("validation defect scope_selector path is invalid")
+    handles = value.get("evidence_handle_ids", ())
+    if not isinstance(handles, list) or len(handles) > 64:
+        raise ProtocolViolation("validation defect evidence handles are invalid")
+    return {
+        **{
+            key: item
+            for key, item in value.items()
+            if key not in {"scope_selector", "evidence_handle_ids"}
+        },
+        "scope_ref": "script-build://scopes/" + "/".join(path),
+        "evidence_refs": [
+            _ref_dict(_required_evidence_handle(item, refs_by_handle)) for item in handles
+        ],
+    }
+
+
 def _normalize_defects(
     defects: Sequence[Mapping[str, Any]],
     allowed_refs: Mapping[tuple[Any, Any, Any, Any], ArtifactRef],

script_build_host/src/script_build_host/tools/registry.py

@@ -59,26 +59,29 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             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."""
-
-        return _json(await gateway.read_input_snapshot(context or {}))
-
-    _register(registry, read_input_snapshot, capabilities=["read"])
-
-    async def load_frozen_strategy(strategy_ref: str, context: dict[str, Any] | None = None) -> str:
-        """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 {}))
-
-    _register(registry, load_frozen_strategy, capabilities=["read"])
-
-    async def read_accepted_artifact(context: dict[str, Any] | None = None) -> str:
-        """Read only direct-child artifacts frozen into this attempt."""
+    async def load_frozen_strategy(
+        strategy_handle: str,
+        cursor: int = 0,
+        context: dict[str, Any] | None = None,
+    ) -> ToolResult:
+        """Load one on-demand strategy selected by its Workbench handle."""
 
-        return _json(await gateway.read_accepted_artifacts(context or {}))
+        value = _json(
+            await gateway.load_frozen_strategy(strategy_handle, context or {}, cursor=cursor)
+        )
+        return ToolResult(
+            title="Frozen strategy",
+            output=value,
+            long_term_memory=f"Loaded frozen strategy {strategy_handle}",
+            include_output_only_once=True,
+        )
 
-    _register(registry, read_accepted_artifact, capabilities=["read"])
+    _register(
+        registry,
+        load_frozen_strategy,
+        capabilities=["read"],
+        schema=_load_frozen_strategy_schema(),
+    )
 
     async def plan_script_tasks(
         contracts: list[dict[str, Any]],
@@ -164,14 +167,56 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         schema=_decide_script_task_schema(),
     )
 
-    async def inspect_script_plan(
+    async def read_workbench_detail(
+        view: str,
+        handle: str | None = None,
+        known_revision: str | None = None,
+        cursor: int = 0,
         context: dict[str, Any] | None = None,
-    ) -> str:
-        """Inspect Tasks; copy complete accepted_decision_ref objects into new contracts."""
+    ) -> ToolResult:
+        """Read one bounded, revision-aware Workbench detail page."""
 
-        return _json(await gateway.inspect_script_plan(context=context or {}))
+        value = await gateway.read_workbench_detail(
+            view=view,
+            handle=handle,
+            known_revision=known_revision,
+            cursor=cursor,
+            context=context or {},
+        )
+        output = _json(value)
+        return ToolResult(
+            title="Mission Workbench detail",
+            output=output,
+            long_term_memory=(
+                f"Workbench {view} revision={value.get('state_revision')} "
+                f"items={len(value.get('items', []))}"
+            ),
+            include_output_only_once=True,
+        )
 
-    _register(registry, inspect_script_plan, capabilities=["read"])
+    _register(
+        registry,
+        read_workbench_detail,
+        capabilities=["read"],
+        schema={
+            "type": "function",
+            "function": {
+                "name": "read_workbench_detail",
+                "description": "Read a bounded page omitted from the automatic Mission context.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "view": {"type": "string", "enum": ["task_tree", "source"]},
+                        "handle": {"type": ["string", "null"]},
+                        "known_revision": {"type": ["string", "null"]},
+                        "cursor": {"type": "integer", "minimum": 0, "default": 0},
+                    },
+                    "required": ["view"],
+                    "additionalProperties": False,
+                },
+            },
+        },
+    )
 
     async def dispatch_script_tasks(
         task_ids: list[str], context: dict[str, Any] | None = None
@@ -311,29 +356,48 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     )
 
     async def save_direction_candidate(
-        goals: list[dict[str, Any]],
-        constraints: list[dict[str, Any]],
-        preferences: list[dict[str, Any]],
-        topic_refs: list[str] | None = None,
-        persona_refs: list[str] | None = None,
-        strategy_refs: list[str] | None = None,
-        evidence_refs: list[str] | None = None,
+        goals: Any,
+        constraints: Any = None,
+        preferences: Any = None,
+        strategy_handles: list[str] | None = None,
+        evidence_decision_ids: list[str] | None = None,
         context: dict[str, Any] | None = None,
     ) -> str:
         """Freeze one direction candidate inside the protected Attempt."""
 
+        normalized = _decode_bounded_structured_value(goals, "goals")
+        if isinstance(normalized, dict) and "goals" in normalized:
+            envelope = normalized
+            goals = envelope["goals"]
+            constraints = envelope.get("constraints", constraints)
+            preferences = envelope.get("preferences", preferences)
+            strategy_handles = envelope.get("strategy_handles", strategy_handles)
+            evidence_decision_ids = envelope.get("evidence_decision_ids", evidence_decision_ids)
         goals = _structured_object_list(goals, "goals")
         constraints = _structured_object_list(constraints, "constraints")
         preferences = _structured_object_list(preferences, "preferences")
+        _require_object_fields(
+            goals,
+            ("client_key", "statement", "rationale", "success_criteria"),
+            "goals",
+        )
+        _require_object_fields(
+            constraints,
+            ("client_key", "statement", "rationale"),
+            "constraints",
+        )
+        _require_object_fields(
+            preferences,
+            ("client_key", "statement", "rationale"),
+            "preferences",
+        )
         return _json(
             await gateway.save_direction_candidate(
                 goals=goals,
                 constraints=constraints,
                 preferences=preferences,
-                topic_refs=topic_refs or [],
-                persona_refs=persona_refs or [],
-                strategy_refs=strategy_refs or [],
-                evidence_refs=evidence_refs or [],
+                strategy_handles=strategy_handles or [],
+                evidence_decision_ids=evidence_decision_ids or [],
                 context=context or {},
             )
         )
@@ -345,36 +409,13 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         schema=_save_direction_candidate_schema(),
     )
 
-    async def read_accepted_input_bundle(
-        context: dict[str, Any] | None = None,
-    ) -> str:
-        """Read the ordered, immutable direct and explicitly imported inputs."""
-
-        return _json(await gateway.read_accepted_input_bundle(context or {}))
-
-    _register(registry, read_accepted_input_bundle, capabilities=["read"])
-
-    async def read_active_frontier(context: dict[str, Any] | None = None) -> str:
-        """Read only adopted, non-superseded candidate inputs pinned by Compose."""
-
-        return _json(await gateway.read_active_frontier(context or {}))
-
-    _register(registry, read_active_frontier, capabilities=["read"])
-
-    async def read_pinned_candidates(context: dict[str, Any] | None = None) -> str:
-        """Read only immutable candidates explicitly pinned for Compare or Portfolio."""
-
-        return _json(await gateway.read_pinned_candidates(context or {}))
-
-    _register(registry, read_pinned_candidates, capabilities=["read"])
-
     async def view_frozen_images(
-        raw_artifact_refs: list[str],
+        image_handles: list[str],
         context: dict[str, Any] | None = None,
     ) -> ToolResult:
         """Load verified bytes from frozen raw artifacts, never from network URLs."""
 
-        values = await gateway.view_frozen_images(raw_artifact_refs, context or {})
+        values = await gateway.view_frozen_images(image_handles, context or {})
         images, manifest = _validated_images(values)
         output = _json({"images": manifest})
         return ToolResult(
@@ -387,210 +428,63 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, view_frozen_images, capabilities=["read"])
 
-    async def read_attempt_workspace(context: dict[str, Any] | None = None) -> str:
-        """Read the current Attempt's isolated Paragraph/Element/Link workspace."""
-
-        return _json(await gateway.read_attempt_workspace(context or {}))
-
-    _register(registry, read_attempt_workspace, capabilities=["read"])
-
-    async def create_script_paragraph(
-        paragraph_index: int,
-        name: str,
-        content_range: dict[str, Any],
-        level: int = 1,
-        parent_paragraph_id: int | None = None,
-        theme_elements: list[dict[str, Any]] | None = None,
-        form_elements: list[dict[str, Any]] | None = None,
-        function_elements: list[dict[str, Any]] | None = None,
-        feeling_elements: list[dict[str, Any]] | None = None,
-        context: dict[str, Any] | None = None,
-    ) -> 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",
-                {
-                    "paragraph_index": paragraph_index,
-                    "name": name,
-                    "content_range": content_range,
-                    "level": level,
-                    "parent_paragraph_id": parent_paragraph_id,
-                    "theme_elements": theme_elements or [],
-                    "form_elements": form_elements or [],
-                    "function_elements": function_elements or [],
-                    "feeling_elements": feeling_elements or [],
-                },
-                context or {},
-            )
-        )
-
-    _register(registry, create_script_paragraph, capabilities=["write"])
-
-    async def create_script_paragraphs(
+    async def save_script_paragraphs(
         paragraphs: list[dict[str, Any]],
+        expected_state_revision: str,
         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,
-        atoms: list[dict[str, Any]],
-        context: dict[str, Any] | None = None,
-    ) -> str:
-        """Append unique atoms to one allowlisted Paragraph column."""
+        """Atomically upsert Paragraphs using semantic client or target keys."""
 
-        atoms = _structured_object_list(atoms, "atoms")
         return _json(
             await gateway.candidate_command(
-                "append_paragraph_atoms",
-                {"paragraph_id": paragraph_id, "column": column, "atoms": atoms},
-                context or {},
-            )
-        )
-
-    _register(registry, append_paragraph_atoms, capabilities=["write"])
-
-    async def delete_paragraph_atom(
-        paragraph_id: int,
-        column: str,
-        dimension: str,
-        atom_value: str,
-        context: dict[str, Any] | None = None,
-    ) -> str:
-        """Delete one exact atom from the current Paragraph workspace."""
-
-        return _json(
-            await gateway.candidate_command(
-                "delete_paragraph_atom",
+                "save_script_paragraphs",
                 {
-                    "paragraph_id": paragraph_id,
-                    "column": column,
-                    "dimension": dimension,
-                    "atom_value": atom_value,
+                    "paragraphs": [_structured_object(item, "paragraph") for item in paragraphs],
+                    "expected_state_revision": expected_state_revision,
                 },
                 context or {},
             )
         )
 
-    _register(registry, delete_paragraph_atom, capabilities=["write"])
+    _register(
+        registry,
+        save_script_paragraphs,
+        capabilities=["write"],
+        schema=_save_script_paragraphs_schema(),
+    )
 
-    async def batch_update_script_paragraphs(
-        updates: list[dict[str, Any]],
+    async def save_script_elements(
+        elements: list[dict[str, Any]],
+        links: list[dict[str, Any]],
+        expected_state_revision: str,
         context: dict[str, Any] | None = None,
     ) -> 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 {}
-            )
-        )
-
-    _register(registry, batch_update_script_paragraphs, capabilities=["write"])
+        """Atomically save Elements and link them to semantic Paragraph targets."""
 
-    async def create_script_element(
-        name: str,
-        dimension_primary: str,
-        dimension_secondary: str,
-        commonality_analysis: dict[str, Any] | None = None,
-        topic_support: dict[str, Any] | None = None,
-        weight_score: dict[str, Any] | None = None,
-        support_elements: list[dict[str, Any]] | None = None,
-        context: dict[str, Any] | None = None,
-    ) -> 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")
+        elements = _structured_object_list(elements, "elements", max_items=100)
+        links = _structured_object_list(links, "links", max_items=100)
         return _json(
             await gateway.candidate_command(
-                "create_script_element",
+                "save_script_elements",
                 {
-                    "name": name,
-                    "dimension_primary": dimension_primary,
-                    "dimension_secondary": dimension_secondary,
-                    "commonality_analysis": commonality_analysis,
-                    "topic_support": topic_support,
-                    "weight_score": weight_score,
-                    "support_elements": support_elements or [],
+                    "elements": elements,
+                    "links": links,
+                    "expected_state_revision": expected_state_revision,
                 },
                 context or {},
             )
         )
 
-    _register(registry, create_script_element, capabilities=["write"])
-
-    async def update_script_element(
-        element_id: int,
-        updates: dict[str, Any],
-        context: dict[str, Any] | None = None,
-    ) -> str:
-        """Update one Element with an allowlisted, workspace-local patch."""
-
-        updates = _structured_object(updates, "updates")
-        return _json(
-            await gateway.candidate_command(
-                "update_script_element",
-                {"element_id": element_id, "updates": updates},
-                context or {},
-            )
-        )
-
-    _register(registry, update_script_element, capabilities=["write"])
-
-    async def batch_link_paragraph_elements(
-        links: list[dict[str, Any]],
-        context: dict[str, Any] | None = None,
-    ) -> 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 {})
-        )
-
-    _register(registry, batch_link_paragraph_elements, capabilities=["write"])
-
-    async def delete_paragraph_element_links(
-        links: list[dict[str, Any]],
-        context: dict[str, Any] | None = None,
-    ) -> 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 {})
-        )
-
-    _register(registry, delete_paragraph_element_links, capabilities=["write"])
+    _register(
+        registry,
+        save_script_elements,
+        capabilities=["write"],
+        schema=_save_script_elements_schema(),
+    )
 
     async def save_comparison_candidate(
         criterion_results: list[dict[str, Any]],
-        recommendation: str,
+        recommended_decision_id: str,
         conflicts: list[str] | None = None,
         context: dict[str, Any] | None = None,
     ) -> str:
@@ -603,7 +497,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
                 {
                     "criterion_results": criterion_results,
                     "conflicts": conflicts or [],
-                    "recommendation": recommendation,
+                    "recommended_decision_id": recommended_decision_id,
                 },
                 context or {},
             )
@@ -611,46 +505,6 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, save_comparison_candidate, capabilities=["write"])
 
-    async def save_structured_script_candidate(
-        acceptance_notes: list[str] | None = None,
-        context: dict[str, Any] | None = None,
-    ) -> str:
-        """Materialize and freeze exactly one StructuredScript from active frontier."""
-
-        return _json(
-            await gateway.save_structured_script_candidate(acceptance_notes or [], context or {})
-        )
-
-    _register(registry, save_structured_script_candidate, capabilities=["write"])
-
-    async def save_candidate_portfolio(
-        unresolved_defects: list[dict[str, Any]],
-        superseded_decision_ids: list[str] | None = None,
-        context: dict[str, Any] | None = None,
-    ) -> 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",
-                {
-                    "superseded_decision_ids": superseded_decision_ids or [],
-                    "unresolved_defects": unresolved_defects,
-                },
-                context or {},
-            )
-        )
-
-    _register(registry, save_candidate_portfolio, capabilities=["write"])
-
-    async def read_accepted_portfolio(context: dict[str, Any] | None = None) -> str:
-        """Read the Root Attempt's accepted Direction, Portfolio and adopted script."""
-
-        return _json(await gateway.read_accepted_portfolio(context or {}))
-
-    _register(registry, read_accepted_portfolio, capabilities=["read"])
-
     async def legacy_projection_dry_run(
         build_summary: str,
         context: dict[str, Any] | None = None,
@@ -810,9 +664,7 @@ def _strongest_cycle_failure(results: Any) -> FailureDetail | None:
     strongest: FailureDetail | None = None
     for result in results:
         value = (
-            result.get("failure")
-            if isinstance(result, dict)
-            else getattr(result, "failure", None)
+            result.get("failure") if isinstance(result, dict) else getattr(result, "failure", None)
         )
         if value is None:
             continue
@@ -833,6 +685,18 @@ def _strongest_cycle_failure(results: Any) -> FailureDetail | None:
             source_tool=failure.source_tool,
             details=details,
         )
+        if failure.code == "NO_PROGRESS" and failure.disposition is FailureDisposition.ABORT_RUN:
+            failure = FailureDetail(
+                code=failure.code,
+                message=failure.message,
+                disposition=FailureDisposition.REPLAN_TASK,
+                source_tool="dispatch_script_tasks",
+                details={
+                    **details,
+                    "child_disposition": FailureDisposition.ABORT_RUN.value,
+                    "child_source_tool": failure.source_tool,
+                },
+            )
         if strongest is None or priority[failure.disposition] > priority[strongest.disposition]:
             strongest = failure
     return strongest
@@ -976,9 +840,18 @@ def _planner_task_input_schema() -> dict[str, Any]:
         "type": "object",
         "properties": {
             "task_kind": {"type": "string", "enum": task_kinds},
-            "scope_ref": {
-                "type": "string",
-                "description": "Stable logical scope URI beginning with script-build://.",
+            "scope_selector": {
+                "type": "object",
+                "properties": {
+                    "anchor": {"type": "string", "enum": ["mission", "parent"]},
+                    "path": {
+                        "type": "array",
+                        "maxItems": 4,
+                        "items": {"type": "string", "minLength": 1, "maxLength": 128},
+                    },
+                },
+                "required": ["anchor", "path"],
+                "additionalProperties": False,
             },
             "intent_class": {
                 "type": "string",
@@ -1012,19 +885,19 @@ def _planner_task_input_schema() -> dict[str, Any]:
                 "uniqueItems": True,
                 "default": [],
             },
-            "gap_ref": {"type": ["string", "null"]},
+            "gap_key": {"type": ["string", "null"]},
             "criteria": {
                 "type": "array",
-                "minItems": 1,
+                "minItems": 0,
                 "maxItems": 16,
                 "items": {
                     "type": "object",
                     "properties": {
-                        "criterion_id": {"type": "string"},
+                        "client_key": {"type": "string"},
                         "description": {"type": "string"},
                         "hard": {"type": "boolean"},
                     },
-                    "required": ["criterion_id", "description", "hard"],
+                    "required": ["client_key", "description", "hard"],
                     "additionalProperties": False,
                 },
             },
@@ -1041,7 +914,7 @@ def _planner_task_input_schema() -> dict[str, Any]:
         },
         "required": [
             "task_kind",
-            "scope_ref",
+            "scope_selector",
             "intent_class",
             "objective",
             "criteria",
@@ -1057,10 +930,8 @@ def _root_replacement_schema() -> dict[str, Any]:
         "type": "object",
         "properties": {
             "objective": {"type": "string", "maxLength": 2000},
-            "acceptance_criteria": {"type": "array", "items": {"type": "object"}},
-            "context_refs": {"type": "array", "items": {"type": "string"}},
         },
-        "required": ["objective", "acceptance_criteria", "context_refs"],
+        "required": ["objective"],
         "additionalProperties": False,
     }
 
@@ -1069,22 +940,22 @@ def _save_direction_candidate_schema() -> dict[str, Any]:
     constraint = {
         "type": "object",
         "properties": {
-            "constraint_id": {"type": "string", "minLength": 1},
+            "client_key": {"type": "string", "minLength": 1},
             "statement": {"type": "string", "minLength": 1},
             "rationale": {"type": "string"},
         },
-        "required": ["constraint_id", "statement", "rationale"],
+        "required": ["client_key", "statement", "rationale"],
         "additionalProperties": False,
     }
     preference = {
         "type": "object",
         "properties": {
-            "preference_id": {"type": "string", "minLength": 1},
+            "client_key": {"type": "string", "minLength": 1},
             "statement": {"type": "string", "minLength": 1},
             "rationale": {"type": "string"},
             "priority": {"type": "integer", "minimum": 1, "maximum": 3},
         },
-        "required": ["preference_id", "statement", "rationale"],
+        "required": ["client_key", "statement", "rationale"],
         "additionalProperties": False,
     }
     return {
@@ -1092,8 +963,7 @@ def _save_direction_candidate_schema() -> dict[str, Any]:
         "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."
+                "Freeze one Direction using semantic client keys and accepted Evidence decisions."
             ),
             "parameters": {
                 "type": "object",
@@ -1131,24 +1001,203 @@ def _save_direction_candidate_schema() -> dict[str, Any]:
                     },
                     "constraints": {"type": "array", "items": constraint},
                     "preferences": {"type": "array", "items": preference},
-                    "topic_refs": {"type": "array", "items": {"type": "string"}, "default": []},
-                    "persona_refs": {
+                    "strategy_handles": {
                         "type": "array",
                         "items": {"type": "string"},
                         "default": [],
                     },
-                    "strategy_refs": {
+                    "evidence_decision_ids": {
                         "type": "array",
+                        "minItems": 1,
                         "items": {"type": "string"},
-                        "default": [],
                     },
-                    "evidence_refs": {
+                },
+                "required": [
+                    "goals",
+                    "constraints",
+                    "preferences",
+                    "evidence_decision_ids",
+                ],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
+def _load_frozen_strategy_schema() -> dict[str, Any]:
+    return {
+        "type": "function",
+        "function": {
+            "name": "load_frozen_strategy",
+            "description": (
+                "Load one on-demand strategy using the exact Workbench strategy handle."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "strategy_handle": {
+                        "type": "string",
+                        "pattern": "^strategy_[0-9a-f]{20}$",
+                    },
+                    "cursor": {"type": "integer", "minimum": 0, "default": 0},
+                },
+                "required": ["strategy_handle"],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
+def _save_script_paragraphs_schema() -> dict[str, Any]:
+    creative_atom = {
+        "type": "object",
+        "properties": {
+            "原子点": {"type": "string", "minLength": 1},
+            "维度": {"type": "string", "minLength": 1},
+            "维度类型": {"type": "string", "enum": ["主维度", "从维度"]},
+        },
+        "required": ["原子点", "维度", "维度类型"],
+        "additionalProperties": False,
+    }
+    feeling_atom = {
+        "type": "object",
+        "properties": {
+            "原子点": {"type": "string", "minLength": 1},
+            "维度": {"type": "string", "minLength": 1},
+        },
+        "required": ["原子点", "维度"],
+        "additionalProperties": False,
+    }
+    paragraph = {
+        "type": "object",
+        "properties": {
+            "client_key": {"type": "string", "minLength": 1, "maxLength": 128},
+            "paragraph_target_key": {
+                "type": "string",
+                "pattern": "^pt_[0-9a-f]{20}$",
+            },
+            "parent_client_key": {
+                "type": ["string", "null"],
+                "minLength": 1,
+                "maxLength": 128,
+            },
+            "parent_target_key": {
+                "type": ["string", "null"],
+                "pattern": "^pt_[0-9a-f]{20}$",
+            },
+            "paragraph_index": {"type": "integer", "minimum": 1},
+            "name": {"type": "string", "minLength": 1},
+            "content_range": {"type": "object"},
+            "level": {"type": "integer", "enum": [1, 2]},
+            "theme_elements": {"type": "array", "minItems": 1, "items": creative_atom},
+            "form_elements": {"type": "array", "minItems": 1, "items": creative_atom},
+            "function_elements": {
+                "type": "array",
+                "minItems": 1,
+                "items": creative_atom,
+            },
+            "feeling_elements": {"type": "array", "minItems": 1, "items": feeling_atom},
+            **{
+                field: {"type": "string", "minLength": 1}
+                for field in (
+                    "theme",
+                    "form",
+                    "function",
+                    "feeling",
+                    "description",
+                    "full_description",
+                )
+            },
+        },
+        "oneOf": [
+            {"required": ["client_key"], "not": {"required": ["paragraph_target_key"]}},
+            {"required": ["paragraph_target_key"], "not": {"required": ["client_key"]}},
+        ],
+        "additionalProperties": False,
+    }
+    return {
+        "type": "function",
+        "function": {
+            "name": "save_script_paragraphs",
+            "description": (
+                "Atomically create or update Paragraphs. New rows use client_key; existing "
+                "rows use paragraph_target_key from the protected role context."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "paragraphs": {
                         "type": "array",
                         "minItems": 1,
-                        "items": {"type": "string"},
+                        "items": paragraph,
                     },
+                    "expected_state_revision": {"type": "string", "minLength": 1},
+                },
+                "required": ["paragraphs", "expected_state_revision"],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
+def _save_script_elements_schema() -> dict[str, Any]:
+    element = {
+        "type": "object",
+        "properties": {
+            "client_key": {"type": "string", "minLength": 1, "maxLength": 128},
+            "name": {"type": "string", "minLength": 1},
+            "dimension_primary": {"type": "string", "enum": ["实质", "形式"]},
+            "dimension_secondary": {"type": "string", "minLength": 1},
+            "commonality_analysis": {"type": ["object", "null"]},
+            "topic_support": {"type": ["object", "null"]},
+            "weight_score": {"type": ["object", "null"]},
+            "support_elements": {
+                "type": "array",
+                "items": {"type": "object"},
+                "default": [],
+            },
+        },
+        "required": [
+            "client_key",
+            "name",
+            "dimension_primary",
+            "dimension_secondary",
+        ],
+        "additionalProperties": False,
+    }
+    link = {
+        "type": "object",
+        "properties": {
+            "paragraph_target_key": {
+                "type": "string",
+                "pattern": "^pt_[0-9a-f]{20}$",
+            },
+            "element_client_keys": {
+                "type": "array",
+                "minItems": 1,
+                "uniqueItems": True,
+                "items": {"type": "string", "minLength": 1, "maxLength": 128},
+            },
+        },
+        "required": ["paragraph_target_key", "element_client_keys"],
+        "additionalProperties": False,
+    }
+    return {
+        "type": "function",
+        "function": {
+            "name": "save_script_elements",
+            "description": (
+                "Atomically create every Element and all Paragraph links in one call. "
+                "Links use Paragraph target keys and batch-local Element client keys."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "elements": {"type": "array", "minItems": 1, "items": element},
+                    "links": {"type": "array", "minItems": 1, "items": link},
+                    "expected_state_revision": {"type": "string", "minLength": 1},
                 },
-                "required": ["goals", "constraints", "preferences", "evidence_refs"],
+                "required": ["elements", "links", "expected_state_revision"],
                 "additionalProperties": False,
             },
         },
@@ -1156,15 +1205,18 @@ def _save_direction_candidate_schema() -> dict[str, Any]:
 
 
 def _submit_validation_schema() -> dict[str, Any]:
-    artifact_ref = {
+    scope_selector = {
         "type": "object",
         "properties": {
-            "uri": {"type": "string"},
-            "kind": {"type": "string"},
-            "version": {"type": ["string", "null"]},
-            "digest": {"type": ["string", "null"]},
+            "anchor": {"type": "string", "enum": ["mission"]},
+            "path": {
+                "type": "array",
+                "minItems": 1,
+                "maxItems": 4,
+                "items": {"type": "string", "minLength": 1, "maxLength": 128},
+            },
         },
-        "required": ["uri", "kind"],
+        "required": ["anchor", "path"],
         "additionalProperties": False,
     }
     defect = {
@@ -1172,9 +1224,9 @@ def _submit_validation_schema() -> dict[str, Any]:
         "properties": {
             "defect_code": {"type": "string"},
             "criterion_id": {"type": "string"},
-            "scope_ref": {"type": "string"},
+            "scope_selector": scope_selector,
             "observed_excerpt": {"type": "string", "maxLength": 500},
-            "evidence_refs": {"type": "array", "items": artifact_ref},
+            "evidence_handle_ids": {"type": "array", "items": {"type": "string"}},
             "severity": {"type": "string", "enum": ["warning", "hard", "critical"]},
             "invalidated_inputs": {"type": "array", "items": {"type": "string"}},
             "recommended_action_class": {
@@ -1194,9 +1246,9 @@ def _submit_validation_schema() -> dict[str, Any]:
         "required": [
             "defect_code",
             "criterion_id",
-            "scope_ref",
+            "scope_selector",
             "observed_excerpt",
-            "evidence_refs",
+            "evidence_handle_ids",
             "severity",
             "invalidated_inputs",
             "recommended_action_class",
@@ -1229,9 +1281,9 @@ def _submit_validation_schema() -> dict[str, Any]:
                                     "enum": ["passed", "failed", "inconclusive"],
                                 },
                                 "reason": {"type": "string", "maxLength": 300},
-                                "evidence_refs": {
+                                "evidence_handle_ids": {
                                     "type": "array",
-                                    "items": artifact_ref,
+                                    "items": {"type": "string"},
                                     "maxItems": 64,
                                 },
                             },
@@ -1239,7 +1291,7 @@ def _submit_validation_schema() -> dict[str, Any]:
                                 "criterion_id",
                                 "verdict",
                                 "reason",
-                                "evidence_refs",
+                                "evidence_handle_ids",
                             ],
                             "additionalProperties": False,
                         },
@@ -1322,6 +1374,15 @@ def _structured_object_list(
     return [_structured_object(item, f"{label}[{index}]") for index, item in enumerate(normalized)]
 
 
+def _require_object_fields(
+    items: list[dict[str, Any]], required: tuple[str, ...], label: str
+) -> None:
+    for index, item in enumerate(items):
+        missing = [field for field in required if field not in item]
+        if missing:
+            raise ValueError(f"{label}[{index}] is missing required fields: {', '.join(missing)}")
+
+
 def _decode_bounded_structured_value(value: Any, label: str) -> Any:
     if isinstance(value, str):
         if len(value.encode("utf-8")) > _MAX_STRUCTURED_ARGUMENT_BYTES:
@@ -1369,21 +1430,16 @@ def _validated_images(
         expected_digest = item.get("sha256")
         if expected_digest is not None and expected_digest != digest:
             raise ValueError("frozen image digest does not match its bytes")
-        raw_artifact_ref = item.get("raw_artifact_ref")
-        if (
-            not isinstance(raw_artifact_ref, str)
-            or len(raw_artifact_ref) > 200
-            or not raw_artifact_ref.startswith("script-build://raw-artifacts/sha256/")
-        ):
-            raise ValueError("frozen image must identify one immutable raw artifact")
+        image_handle = item.get("image_handle")
+        if not isinstance(image_handle, str) or not image_handle.startswith("image_"):
+            raise ValueError("frozen image must identify one Workbench image handle")
         images.append({"type": "base64", "media_type": media_type, "data": data})
         manifest.append(
             {
                 "index": index,
                 "media_type": media_type,
                 "byte_size": len(content),
-                "sha256": digest,
-                "raw_artifact_ref": raw_artifact_ref,
+                "image_handle": image_handle,
             }
         )
     return images, manifest

script_build_host/tests/test_agent_surface.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from dataclasses import replace
 from datetime import UTC, datetime
 from hashlib import sha256
@@ -21,7 +22,8 @@ from script_build_host.agents.prompts import (
 from script_build_host.domain.canonical_json import canonical_sha256
 from script_build_host.domain.errors import ProtocolViolation
 from script_build_host.domain.input_snapshot import ScriptBuildInputSnapshotV1
-from script_build_host.tools.gateway import LegacyScriptToolGateway
+from script_build_host.domain.workbench import strategy_handle
+from script_build_host.tools.gateway import LegacyScriptToolGateway, _project_snapshot
 from script_build_host.tools.registry import register_script_tools
 
 
@@ -63,37 +65,33 @@ def test_script_presets_are_exact_and_never_expose_legacy_control_tools() -> Non
     assert set(planner.allowed_tools or []) == {
         "plan_script_tasks",
         "decide_script_task",
-        "inspect_script_plan",
+        "read_workbench_detail",
         "dispatch_script_tasks",
         "validate_attempt",
-        "read_input_snapshot",
     }
     compose = get_preset("script_compose_worker")
-    assert "read_active_frontier" in (compose.allowed_tools or [])
-    assert "read_accepted_artifact" not in (compose.allowed_tools or [])
+    assert set(compose.allowed_tools or []) == {"submit_attempt"}
     compare = get_preset("script_candidate_compare_worker")
-    assert "read_pinned_candidates" in (compare.allowed_tools or [])
+    assert "save_comparison_candidate" in (compare.allowed_tools or [])
     assert "create_script_paragraph" not in (compare.allowed_tools or [])
     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 [])
+    assert set(portfolio.allowed_tools or []) == {"submit_attempt"}
     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 "view_frozen_images" in (get_preset("script_candidate_validator").allowed_tools or [])
+    element_tools = set(get_preset("script_element_set_worker").allowed_tools or [])
+    assert "save_script_elements" in element_tools
+    assert "create_script_element" not in element_tools
+    assert "batch_link_paragraph_elements" not in element_tools
 
     assert set(get_preset("script_root_worker").allowed_tools or []) == {
-        "read_input_snapshot",
-        "read_accepted_portfolio",
+        "read_workbench_detail",
         "legacy_projection_dry_run",
         "save_root_delivery_manifest",
         "submit_attempt",
     }
     assert set(get_preset("script_root_validator").allowed_tools or []) == {
-        "read_input_snapshot",
-        "read_accepted_portfolio",
         "query_validation_evidence",
         "deterministic_precheck",
         "submit_validation",
@@ -150,7 +148,16 @@ def test_legacy_retrieval_tool_schemas_keep_old_argument_shapes() -> None:
     assert knowledge["properties"]["max_count"]["default"] == 3
     assert schemas["query_pattern_qa"]["required"] == ["message"]
     assert schemas["load_images"]["required"] == ["image_urls"]
-    assert schemas["load_frozen_strategy"]["required"] == ["strategy_ref"]
+    assert schemas["load_frozen_strategy"]["required"] == ["strategy_handle"]
+    assert schemas["save_script_elements"]["required"] == [
+        "elements",
+        "links",
+        "expected_state_revision",
+    ]
+    assert schemas["save_script_elements"]["properties"]["links"]["items"]["required"] == [
+        "paragraph_target_key",
+        "element_client_keys",
+    ]
     assert schemas["submit_attempt"].get("properties") == {}
     assert schemas["submit_validation"]["required"] == [
         "verdict",
@@ -166,6 +173,128 @@ def test_legacy_retrieval_tool_schemas_keep_old_argument_shapes() -> None:
     }
 
 
+def test_agent_visible_write_schemas_never_expose_mechanical_identity_fields() -> None:
+    registry = ToolRegistry()
+    register_script_tools(registry, _Gateway())  # type: ignore[arg-type]
+    schemas = {
+        item["function"]["name"]: item["function"]["parameters"] for item in registry.get_schemas()
+    }
+    write_tools = {
+        "plan_script_tasks",
+        "decide_script_task",
+        "save_direction_candidate",
+        "save_script_paragraphs",
+        "save_script_elements",
+        "save_comparison_candidate",
+        "save_root_delivery_manifest",
+        "submit_validation",
+    }
+    forbidden = {
+        "paragraph_id",
+        "element_id",
+        "branch_id",
+        "artifact_ref",
+        "evidence_refs",
+        "write_scope",
+        "output_schema",
+        "compose_order",
+        "context_refs",
+    }
+
+    def property_names(value: object) -> set[str]:
+        if isinstance(value, dict):
+            names = (
+                set(value.get("properties", {}))
+                if isinstance(value.get("properties"), dict)
+                else set()
+            )
+            return names | set().union(*(property_names(item) for item in value.values()))
+        if isinstance(value, list):
+            return set().union(*(property_names(item) for item in value))
+        return set()
+
+    for name in write_tools:
+        assert forbidden.isdisjoint(property_names(schemas[name])), name
+
+    register_script_presets()
+    forbidden_tools = {
+        "read_input_snapshot",
+        "read_attempt_workspace",
+        "read_active_frontier",
+        "read_pinned_candidates",
+        "read_accepted_artifact",
+        "create_script_paragraph",
+        "create_script_element",
+        "update_script_element",
+        "batch_link_paragraph_elements",
+        "append_paragraph_atoms",
+        "delete_paragraph_atom",
+    }
+    for preset_name in (
+        "script_direction_worker",
+        "script_structure_worker",
+        "script_paragraph_worker",
+        "script_element_set_worker",
+        "script_candidate_compare_worker",
+        "script_compose_worker",
+        "script_candidate_portfolio_worker",
+        "script_root_worker",
+    ):
+        assert forbidden_tools.isdisjoint(get_preset(preset_name).allowed_tools or []), preset_name
+
+
+def test_input_snapshot_projection_is_bounded_and_keeps_business_inputs() -> None:
+    persona = [
+        {
+            "列": ("实质", "形式", "作用", "感受")[index % 4],
+            "点名称": f"point-{index}",
+            "人设权重分": str(1 - index / 100),
+            "帖子覆盖率": "0.8",
+            "分类路径": ["/account/style"],
+        }
+        for index in range(87)
+    ]
+    payload = {
+        "snapshot_id": "50",
+        "script_build_id": 900049,
+        "canonical_sha256": "sha256:" + "a" * 64,
+        "account": {"account_name": "account"},
+        "topic": {
+            "topic": {"result": "gym story", "topic_direction": "satirical comic"},
+            "composition_items": [
+                {
+                    "id": index,
+                    "point_type": "关键点",
+                    "dimension": "形式",
+                    "element_name": f"item-{index}",
+                    "note": "useful",
+                    "created_at": "unused metadata",
+                    "is_active": True,
+                }
+                for index in range(30)
+            ],
+        },
+        "persona_points": persona,
+        "section_patterns": [{"模式名称": "parallel atlas"}],
+        "strategies": [],
+        "prompt_manifest": [{"content": "x" * 100_000}],
+        "model_manifest": {"secret": "x" * 100_000},
+        "datasource_manifest": {"endpoint": "x" * 100_000},
+    }
+    projected = _project_snapshot(payload, view="element-set")
+    encoded = json.dumps(projected, ensure_ascii=False)
+
+    assert projected["view"] == "element-set"
+    assert projected["persona_points_total"] == 87
+    assert len(projected["persona_points"]) == 12
+    assert projected["section_patterns"] == [{"模式名称": "parallel atlas"}]
+    assert "prompt_manifest" not in projected
+    assert "model_manifest" not in projected
+    assert "datasource_manifest" not in projected
+    assert "created_at" not in projected["topic"]["composition_items"][0]
+    assert len(encoded) < 20_000
+
+
 class _Bindings:
     async def get_by_root(self, root_trace_id: str) -> SimpleNamespace:
         assert root_trace_id == "root-a"
@@ -290,7 +419,7 @@ async def test_role_prompt_resolver_uses_task_pinned_snapshot_and_rechecks_diges
 
 @pytest.mark.asyncio
 async def test_snapshot_tool_hides_on_demand_body_until_explicit_digest_verified_load() -> None:
-    strategy_ref = "script-build://strategies/8/versions/2"
+    on_demand_body = "strategy detail " * 1_200
     snapshot = ScriptBuildInputSnapshotV1(
         snapshot_id="3",
         script_build_id=9,
@@ -314,8 +443,8 @@ async def test_snapshot_tool_hides_on_demand_body_until_explicit_digest_verified
                 "version": 2,
                 "mode": "on_demand",
                 "description": "optional",
-                "content": "on demand body",
-                "content_sha256": canonical_sha256("on demand body").wire,
+                "content": on_demand_body,
+                "content_sha256": canonical_sha256(on_demand_body).wire,
             },
         ),
         prompt_manifest=(),
@@ -344,8 +473,15 @@ async def test_snapshot_tool_hides_on_demand_body_until_explicit_digest_verified
     summary = await gateway.read_input_snapshot(context)
     assert summary["strategies"][0]["content"] == "always body"
     assert "content" not in summary["strategies"][1]
-    loaded = await gateway.load_frozen_strategy(strategy_ref, context)
-    assert loaded["content"] == "on demand body"
+    selected_handle = strategy_handle(snapshot.snapshot_id, 1, snapshot.strategies[1])
+    loaded = await gateway.load_frozen_strategy(selected_handle, context)
+    assert loaded["content_fragment"] == on_demand_body[:7_000]
+    assert loaded["next_cursor"] == 7_000
+    second = await gateway.load_frozen_strategy(
+        selected_handle, context, cursor=loaded["next_cursor"]
+    )
+    assert second["content_fragment"] == on_demand_body[7_000:14_000]
+    assert "content_sha256" not in loaded["metadata"]
 
     tampered = replace(
         snapshot,
@@ -362,7 +498,7 @@ async def test_snapshot_tool_hides_on_demand_body_until_explicit_digest_verified
 
     gateway.snapshots = TamperedSnapshots()
     with pytest.raises(ProtocolViolation, match="strategy digest"):
-        await gateway.load_frozen_strategy(strategy_ref, context)
+        await gateway.load_frozen_strategy(selected_handle, context)
 
 
 async def _async(value):

script_build_host/tests/test_phase_two_agent_tools.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
 import base64
+import hashlib
 from collections.abc import Mapping
-from dataclasses import asdict
 from types import SimpleNamespace
 from typing import Any
 
@@ -13,6 +13,7 @@ from agent.orchestration.evidence import EvidenceResponse
 
 from script_build_host.domain.artifacts import ArtifactState
 from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation
+from script_build_host.domain.workbench import semantic_handle
 from script_build_host.tools.contracts import AttemptManifest
 from script_build_host.tools.gateway import LegacyScriptToolGateway, RetrievalResult
 from script_build_host.tools.registry import register_script_tools
@@ -32,20 +33,27 @@ class _CandidateTools:
             scope_ref="script-build://scopes/body",
         )
 
-    async def create_script_paragraph(
-        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
-    ) -> dict[str, int]:
-        self.last_payload = dict(payload)
+    async def save_script_paragraphs(
+        self,
+        *,
+        paragraphs: list[Mapping[str, Any]],
+        expected_state_revision: str,
+        context: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        self.last_payload = {
+            "paragraphs": list(paragraphs),
+            "expected_state_revision": expected_state_revision,
+        }
         self.last_context = dict(context)
-        return {"paragraph_id": 7}
+        return {"created": 1}
 
     async def view_frozen_images(
         self,
         *,
-        raw_artifact_refs: list[str],
+        image_handles: list[str],
         context: Mapping[str, Any],
     ) -> list[dict[str, Any]]:
-        self.last_payload = {"raw_artifact_refs": raw_artifact_refs}
+        self.last_payload = {"image_handles": image_handles}
         self.last_context = dict(context)
         return self.images
 
@@ -194,7 +202,7 @@ def _gateway(ref: ArtifactRef) -> tuple[LegacyScriptToolGateway, _Coordinator, _
 
 
 @pytest.mark.asyncio
-async def test_validation_query_includes_exact_current_business_artifact() -> None:
+async def test_validation_query_returns_only_bounded_evidence_handles() -> None:
     ref = ArtifactRef(
         uri="script-build://artifact-versions/9",
         kind="paragraph",
@@ -233,23 +241,12 @@ async def test_validation_query_includes_exact_current_business_artifact() -> No
         },
     )
 
-    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",
-            },
-        }
-    ]
+    assert result == {
+        "items": [],
+        "queries_used": 1,
+        "queries_remaining": 4,
+        "evidence_handles": [],
+    }
 
 
 @pytest.mark.asyncio
@@ -356,15 +353,17 @@ async def test_candidate_write_receives_identity_only_from_protected_context() -
         "spec_version": 1,
     }
     await gateway.candidate_command(
-        "create_script_paragraph",
-        {"paragraph_index": 1, "name": "opening", "content_range": {}},
+        "save_script_paragraphs",
+        {
+            "paragraphs": [{"client_key": "opening", "name": "opening", "content_range": {}}],
+            "expected_state_revision": "revision-a",
+        },
         context,
     )
     assert candidates.last_context == context
     assert candidates.last_payload == {
-        "paragraph_index": 1,
-        "name": "opening",
-        "content_range": {},
+        "paragraphs": [{"client_key": "opening", "name": "opening", "content_range": {}}],
+        "expected_state_revision": "revision-a",
     }
 
 
@@ -392,23 +391,16 @@ async def test_structured_validation_keeps_excerpt_out_of_framework_summary() ->
                 "criterion_id": "criterion-a",
                 "verdict": "failed",
                 "reason": "not realized",
-                "evidence_refs": [],
+                "evidence_handle_ids": [],
             }
         ],
         defects=[
             {
                 "defect_code": "REALIZATION_PLACEHOLDER",
                 "criterion_id": "criterion-a",
-                "scope_ref": "script-build://scopes/body/end",
+                "scope_selector": {"anchor": "mission", "path": ["body", "end"]},
                 "observed_excerpt": excerpt,
-                "evidence_refs": [
-                    {
-                        "uri": ref.uri,
-                        "kind": ref.kind,
-                        "version": ref.version,
-                        "digest": ref.digest,
-                    }
-                ],
+                "evidence_handle_ids": [semantic_handle("src", ref.uri, ref.digest)],
                 "severity": "hard",
                 "invalidated_inputs": ["decision-a"],
                 "recommended_action_class": "split",
@@ -450,9 +442,9 @@ async def test_passed_validation_rejects_blocking_defect_and_unfrozen_evidence()
     defect: dict[str, Any] = {
         "defect_code": "WRITE_SCOPE_CONFLICT",
         "criterion_id": "criterion-a",
-        "scope_ref": "script-build://scopes/body",
+        "scope_selector": {"anchor": "mission", "path": ["body"]},
         "observed_excerpt": "overlapping body",
-        "evidence_refs": [],
+        "evidence_handle_ids": [],
         "severity": "hard",
         "invalidated_inputs": [],
         "recommended_action_class": "replace",
@@ -465,7 +457,7 @@ async def test_passed_validation_rejects_blocking_defect_and_unfrozen_evidence()
                     "criterion_id": "criterion-a",
                     "verdict": "passed",
                     "reason": "ok",
-                    "evidence_refs": [asdict(ref)],
+                    "evidence_handle_ids": [semantic_handle("src", ref.uri, ref.digest)],
                 }
             ],
             defects=[defect],
@@ -474,15 +466,8 @@ async def test_passed_validation_rejects_blocking_defect_and_unfrozen_evidence()
         )
 
     defect["severity"] = "warning"
-    defect["evidence_refs"] = [
-        {
-            "uri": "script-build://artifact-versions/8",
-            "kind": "paragraph",
-            "version": "8",
-            "digest": "sha256:" + "b" * 64,
-        }
-    ]
-    with pytest.raises(ProtocolViolation, match="protected validation closure"):
+    defect["evidence_handle_ids"] = ["src_" + "f" * 20]
+    with pytest.raises(ProtocolViolation, match="unknown evidence handle"):
         await gateway.submit_structured_validation(
             verdict="passed",
             criterion_results=[
@@ -490,7 +475,7 @@ async def test_passed_validation_rejects_blocking_defect_and_unfrozen_evidence()
                     "criterion_id": "criterion-a",
                     "verdict": "passed",
                     "reason": "ok",
-                    "evidence_refs": [asdict(ref)],
+                    "evidence_handle_ids": [semantic_handle("src", ref.uri, ref.digest)],
                 }
             ],
             defects=[defect],
@@ -509,19 +494,21 @@ async def test_view_frozen_images_returns_multimodal_without_url_or_data_in_text
     )
     gateway, _, candidates = _gateway(ref)
     encoded = base64.b64encode(b"safe-image-bytes").decode("ascii")
+    image_handle = "image_" + "a" * 20
     candidates.images = [
         {
             "type": "base64",
             "media_type": "image/png",
             "data": encoded,
-            "raw_artifact_ref": "script-build://raw-artifacts/sha256/abc",
+            "image_handle": image_handle,
+            "sha256": "sha256:" + hashlib.sha256(b"safe-image-bytes").hexdigest(),
         }
     ]
     registry = ToolRegistry()
     register_script_tools(registry, gateway)
     result = await registry.execute(
         "view_frozen_images",
-        {"raw_artifact_refs": ["script-build://raw-artifacts/sha256/abc"]},
+        {"image_handles": [image_handle]},
         context={"root_trace_id": "root-a", "task_id": "task-1"},
     )
     assert isinstance(result, dict)
@@ -539,7 +526,7 @@ async def test_view_frozen_images_returns_multimodal_without_url_or_data_in_text
     ]
     rejected = await registry.execute(
         "view_frozen_images",
-        {"raw_artifact_refs": ["script-build://raw-artifacts/sha256/abc"]},
+        {"image_handles": [image_handle]},
         context={"root_trace_id": "root-a", "task_id": "task-1"},
     )
     assert isinstance(rejected, dict)

script_build_host/tests/test_phase_two_sql_flow.py

@@ -95,7 +95,10 @@ def _contract(
         intent = "explore"
     payload: dict[str, Any] = {
         "task_kind": kind.value,
-        "scope_ref": scope,
+        "scope_selector": {
+            "anchor": "mission",
+            "path": scope.removeprefix("script-build://scopes/").split("/")[:4],
+        },
         "intent_class": intent,
         "objective": f"produce independently verifiable {kind.value} content",
         "input_decision_ids": [str(item["decision_id"]) for item in input_refs],
@@ -108,10 +111,10 @@ def _contract(
             ),
             None,
         ),
-        "gap_ref": None,
+        "gap_key": None,
         "criteria": [
             {
-                "criterion_id": "closed",
+                "client_key": "closed",
                 "description": "the frozen increment is concrete and independently verifiable",
                 "hard": True,
             }
@@ -231,9 +234,7 @@ class _ScriptedExecutor:
             if workspace["paragraphs"]:
                 paragraph_id = workspace["paragraphs"][0]["paragraph_id"]
             else:
-                atom = (
-                    {"原子点": "可观察的反转", "维度": "开场", "维度类型": "主维度"},
-                )
+                atom = ({"原子点": "可观察的反转", "维度": "开场", "维度类型": "主维度"},)
                 created = await self.candidates.create_script_paragraphs(
                     paragraphs=(
                         {
@@ -295,24 +296,32 @@ class _ScriptedExecutor:
                 )
         elif kind is ScriptTaskKind.COMPARE:
             pinned = await self.candidates.read_pinned_candidates(context=candidate_context)
-            candidate_refs = [
-                str(item["artifact_ref"]["uri"])
+            candidate_decision_ids = [
+                str(item["decision_id"])
                 for item in pinned
                 if item["artifact_ref"]["kind"] == ArtifactKind.PARAGRAPH.value
             ]
+            criteria = cast(
+                Sequence[Mapping[str, Any]],
+                cast(Mapping[str, Any], context["task_spec"])["acceptance_criteria"],
+            )
             await self.candidates.save_comparison_candidate(
                 payload={
                     "criterion_results": [
                         {
-                            "criterion_id": "closed",
+                            "criterion_id": str(criterion["criterion_id"]),
                             "candidate_results": [
-                                {"artifact_ref": ref, "reason": "immutable candidate reviewed"}
-                                for ref in candidate_refs
+                                {
+                                    "decision_id": decision_id,
+                                    "reason": "immutable candidate reviewed",
+                                }
+                                for decision_id in candidate_decision_ids
                             ],
                         }
+                        for criterion in criteria
                     ],
                     "conflicts": ["the replacement has the more concrete opening"],
-                    "recommendation": candidate_refs[-1],
+                    "recommended_decision_id": candidate_decision_ids[-1],
                 },
                 context=candidate_context,
             )

script_build_host/tests/test_tool_failure_bridge.py

@@ -8,19 +8,22 @@ from agent import (
     FailureDetail,
     FailureDisposition,
     RunConfig,
+    ToolExecutionError,
     ToolRegistry,
 )
-from agent.orchestration import CompletionPolicy
+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
 
@@ -51,6 +54,10 @@ def _knowledge_off():
             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,
@@ -77,64 +84,116 @@ def test_script_failure_classification_is_central_and_fail_closed(error, expecte
     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):
-    class Gateway:
-        async def save_structured_script_candidate(self, _notes, _context):
+    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",
             )
 
-    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",
+    worker = ScriptBuildDeterministicWorker(Candidates())
+    context = DeterministicWorkerContext(
+        root_trace_id="root",
+        task=type(
+            "Task",
+            (),
+            {
                 "task_id": "compose-task",
-                "attempt_id": "attempt-1",
+                "current_spec": type(
+                    "Spec", (), {"context_refs": ("script-build://task-kinds/compose",)}
+                )(),
+                "current_spec_version": 1,
             },
-            knowledge=_knowledge_off(),
-        ),
+        )(),
+        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 calls == 1
-    assert result["status"] == "failed"
-    assert result["failure"] == {
+    assert captured.value.failure.to_dict() == {
         "code": "INPUT_SCOPE_MISMATCH",
         "message": "Paragraph has no adopted covering Structure",
         "disposition": "replan_task",
-        "source_tool": "save_structured_script_candidate",
+        "source_tool": "deterministic_compose",
         "details": {"attempt_id": "attempt-1", "task_id": "compose-task"},
     }
 
@@ -146,7 +205,7 @@ async def test_workspace_error_can_be_repaired_in_same_attempt(tmp_path):
             self.write_calls = 0
 
         async def candidate_command(self, name, _payload, _context):
-            assert name == "create_script_paragraphs"
+            assert name == "save_script_paragraphs"
             self.write_calls += 1
             if self.write_calls == 1:
                 raise PhaseTwoCandidateError(
@@ -168,30 +227,49 @@ async def test_workspace_error_can_be_repaired_in_same_attempt(tmp_path):
         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"}],
-                }],
-            })
+            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},
-            }],
+            "tool_calls": [
+                {
+                    "id": f"call-{calls}",
+                    "type": "function",
+                    "function": {"name": tool_name, "arguments": arguments},
+                }
+            ],
             "finish_reason": "tool_calls",
         }
 
@@ -205,7 +283,7 @@ async def test_workspace_error_can_be_repaired_in_same_attempt(tmp_path):
         RunConfig(
             agent_type="script_paragraph_worker",
             completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
-            tools=["create_script_paragraphs", "submit_attempt"],
+            tools=["save_script_paragraphs", "submit_attempt"],
             tool_groups=[],
             context={"task_id": "paragraph-task", "attempt_id": "attempt-1"},
             knowledge=_knowledge_off(),
@@ -257,6 +335,41 @@ async def test_dispatch_preserves_batch_and_surfaces_child_failure():
     }
 
 
+@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(
@@ -268,12 +381,14 @@ async def test_planner_breaks_after_same_dispatched_failure_twice(tmp_path):
 
     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(),
-            }]
+            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()
@@ -285,14 +400,16 @@ async def test_planner_breaks_after_same_dispatched_failure_twice(tmp_path):
         calls += 1
         return {
             "content": "",
-            "tool_calls": [{
-                "id": f"dispatch-{calls}",
-                "type": "function",
-                "function": {
-                    "name": "dispatch_script_tasks",
-                    "arguments": json.dumps({"task_ids": ["compose-task"]}),
-                },
-            }],
+            "tool_calls": [
+                {
+                    "id": f"dispatch-{calls}",
+                    "type": "function",
+                    "function": {
+                        "name": "dispatch_script_tasks",
+                        "arguments": json.dumps({"task_ids": ["compose-task"]}),
+                    },
+                }
+            ],
             "finish_reason": "tool_calls",
         }
 
@@ -320,6 +437,4 @@ async def test_planner_breaks_after_same_dispatched_failure_twice(tmp_path):
     assert calls == 2
     assert result["status"] == "incomplete"
     assert result["failure"]["code"] == "NO_PROGRESS"
-    assert result["failure"]["details"]["last_failure"]["code"] == (
-        "INPUT_SCOPE_MISMATCH"
-    )
+    assert result["failure"]["details"]["last_failure"]["code"] == ("INPUT_SCOPE_MISMATCH")