工具：接入受控规划、候选读写与验证提交网关

script_build_host/src/script_build_host/tools/__init__.py

@@ -1,11 +1,15 @@
 """Ownership-aware tools exposed to script-build Agents."""
 
+from .contracts import AttemptManifest, ScriptCandidateToolPort, ScriptPlannerToolPort
 from .gateway import LegacyScriptToolGateway, RetrievalResult
 from .registry import TASK_PRESET_BY_KIND, register_script_tools
 
 __all__ = [
     "TASK_PRESET_BY_KIND",
+    "AttemptManifest",
     "LegacyScriptToolGateway",
     "RetrievalResult",
+    "ScriptCandidateToolPort",
+    "ScriptPlannerToolPort",
     "register_script_tools",
 ]

script_build_host/src/script_build_host/tools/gateway.py

@@ -7,15 +7,28 @@ framework's protected context and then checked against durable bindings.
 
 from __future__ import annotations
 
+import re
 from collections.abc import Mapping, Sequence
 from dataclasses import asdict, dataclass
 from datetime import UTC, datetime
 from typing import Any, Protocol, cast
 from uuid import uuid4
 
-from agent.orchestration import ArtifactRef, EvidenceQuery
+from agent.orchestration import (
+    ArtifactRef,
+    AttemptSubmission,
+    CriterionResult,
+    EvidenceQuery,
+    ValidationVerdict,
+)
 
-from script_build_host.agents.validation import RETRIEVAL_KINDS, task_kind
+from script_build_host.agents.validation import (
+    PHASE_TWO_KINDS,
+    RETRIEVAL_KINDS,
+    precheck_business_artifact,
+    precheck_snapshot,
+    task_kind,
+)
 from script_build_host.domain.artifacts import (
     ArtifactKind,
     Criterion,
@@ -23,7 +36,8 @@ from script_build_host.domain.artifacts import (
     EvidenceRecordV1,
     ScriptDirectionArtifactV1,
 )
-from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.canonical_json import canonical_sha256
+from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation
 from script_build_host.domain.ports import (
     InputSnapshotRepository,
     MissionBindingRepository,
@@ -31,6 +45,15 @@ from script_build_host.domain.ports import (
 )
 from script_build_host.infrastructure.redaction import redact, redact_text
 
+from .contracts import ScriptCandidateToolPort, ScriptPlannerToolPort
+
+_DEFECT_SEVERITIES = frozenset({"warning", "hard", "critical"})
+_DEFECT_ACTIONS = frozenset(
+    {"repair", "retry", "revise", "split", "retrieve", "replace", "compare", "block"}
+)
+_BUSINESS_ARTIFACT_KINDS = frozenset(item.value for item in ArtifactKind)
+_DEFECT_CODE = re.compile(r"^[A-Z][A-Z0-9_]{0,63}$")
+
 
 @dataclass(frozen=True, slots=True)
 class RetrievalResult:
@@ -60,6 +83,10 @@ class DispatchGuard(Protocol):
     async def ensure_dispatch_allowed(self, script_build_id: int) -> None: ...
 
 
+class PhaseTwoBudgetGuard(Protocol):
+    async def ensure_tool_budget(self, *, root_trace_id: str, task_id: str, entry: str) -> None: ...
+
+
 class LegacyScriptToolGateway:
     """Translate protected Agent tool calls to typed domain operations."""
 
@@ -73,6 +100,9 @@ class LegacyScriptToolGateway:
         retrieval_adapters: Mapping[str, RetrievalAdapter] | None = None,
         image_adapter: ImageAdapter | None = None,
         dispatch_guard: DispatchGuard | None = None,
+        planner_tools: ScriptPlannerToolPort | None = None,
+        candidate_tools: ScriptCandidateToolPort | None = None,
+        budget_guard: PhaseTwoBudgetGuard | None = None,
     ) -> None:
         self.bindings = bindings
         self.snapshots = snapshots
@@ -81,18 +111,109 @@ class LegacyScriptToolGateway:
         self.retrieval_adapters = dict(retrieval_adapters or {})
         self.image_adapter = image_adapter
         self.dispatch_guard = dispatch_guard
+        self.planner_tools = planner_tools
+        self.candidate_tools = candidate_tools
+        self.budget_guard = budget_guard
+        self._active_retrieval_attempts: set[tuple[str, str, str]] = set()
 
     async def ensure_dispatch_allowed(self, context: Mapping[str, Any]) -> None:
         binding = await self.bindings.get_by_root(_required(context, "root_trace_id"))
         if self.dispatch_guard is not None:
             await self.dispatch_guard.ensure_dispatch_allowed(binding.script_build_id)
 
+    async def plan_script_tasks(
+        self,
+        *,
+        contracts: Sequence[Mapping[str, Any]],
+        parent_task_id: str | None,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        return await self._planner().plan_script_tasks(
+            contract_payloads=contracts,
+            parent_task_id=parent_task_id,
+            context=context,
+        )
+
+    async def decide_script_task(
+        self,
+        *,
+        task_id: str,
+        action: str,
+        reason: str,
+        validation_id: str | None,
+        replacement_contract: Mapping[str, Any] | None,
+        child_contracts: Sequence[Mapping[str, Any]],
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        return await self._planner().decide_script_task(
+            task_id=task_id,
+            action=action,
+            reason=reason,
+            validation_id=validation_id,
+            replacement_contract=replacement_contract,
+            child_contracts=child_contracts,
+            context=context,
+        )
+
+    async def inspect_script_plan(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        return await self._planner().inspect_script_plan(context=context)
+
+    async def dispatch_script_tasks(
+        self, *, task_ids: Sequence[str], context: Mapping[str, Any]
+    ) -> Sequence[Mapping[str, Any]]:
+        await self.ensure_dispatch_allowed(context)
+        return await self._planner().dispatch_script_tasks(task_ids=task_ids, context=context)
+
     async def read_input_snapshot(self, context: Mapping[str, Any]) -> dict[str, Any]:
         binding, snapshot = await self._scope(context)
         payload = asdict(snapshot)
         payload["script_build_id"] = binding.script_build_id
+        payload["strategies"] = [
+            (
+                item
+                if item.get("mode") == "always_on"
+                else {key: value for key, value in item.items() if key != "content"}
+            )
+            for item in payload.get("strategies", [])
+        ]
+        payload["prompt_manifest"] = [
+            {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))
 
+    async def load_frozen_strategy(
+        self, strategy_ref: str, context: Mapping[str, Any]
+    ) -> dict[str, Any]:
+        """Explicitly load one digest-verified on-demand strategy from the bound snapshot."""
+
+        binding, snapshot = await self._scope(context)
+        ledger = await self.coordinator.task_store.load(_required(context, "root_trace_id"))
+        task = ledger.tasks.get(_required(context, "task_id"))
+        expected_input_ref = f"script-build://inputs/{binding.input_snapshot_id}"
+        if task is None or {
+            item
+            for item in task.current_spec.context_refs
+            if item.startswith("script-build://inputs/")
+        } != {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":
+                matches.append(item)
+        if len(matches) != 1:
+            raise ProtocolViolation("strategy_ref is outside the frozen on-demand strategies")
+        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))
+
     async def read_accepted_artifacts(self, context: Mapping[str, Any]) -> list[dict[str, Any]]:
         binding, _ = await self._scope(context)
         ledger = await self.coordinator.task_store.load(binding.root_trace_id)
@@ -164,6 +285,43 @@ class LegacyScriptToolGateway:
         tool_name: str,
         query: Mapping[str, Any],
         context: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        await self.ensure_dispatch_allowed(context)
+        if self.budget_guard is not None:
+            await self.budget_guard.ensure_tool_budget(
+                root_trace_id=_required(context, "root_trace_id"),
+                task_id=_required(context, "task_id"),
+                entry="retrieval",
+            )
+        root_trace_id = _required(context, "root_trace_id")
+        task_id = _required(context, "task_id")
+        attempt_id = _required(context, "attempt_id")
+        retrieval_key = (root_trace_id, task_id, attempt_id)
+        if retrieval_key in self._active_retrieval_attempts:
+            raise ProtocolViolation("one Retrieval Attempt may execute only one external query")
+        binding = await self.bindings.get_by_root(root_trace_id)
+        try:
+            await self.artifacts.get_by_attempt(
+                script_build_id=binding.script_build_id,
+                task_id=task_id,
+                attempt_id=attempt_id,
+            )
+        except ArtifactNotFound:
+            pass
+        else:
+            raise ProtocolViolation("one Retrieval Attempt may freeze only one Evidence artifact")
+        self._active_retrieval_attempts.add(retrieval_key)
+        try:
+            return await self._retrieve_once(source_type, tool_name, query, context)
+        finally:
+            self._active_retrieval_attempts.discard(retrieval_key)
+
+    async def _retrieve_once(
+        self,
+        source_type: str,
+        tool_name: str,
+        query: Mapping[str, Any],
+        context: Mapping[str, Any],
     ) -> dict[str, Any]:
         binding, snapshot = await self._scope(context)
         if source_type == "decode" and query.get("account_name"):
@@ -223,6 +381,267 @@ class LegacyScriptToolGateway:
         _, snapshot = await self._scope(context)
         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]
+    ) -> Sequence[Mapping[str, Any]]:
+        return await self._candidates().view_frozen_images(
+            raw_artifact_refs=raw_artifact_refs,
+            context=context,
+        )
+
+    async def read_accepted_input_bundle(self, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        return await self._candidates().read_accepted_input_bundle(context=context)
+
+    async def read_active_frontier(self, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        return await self._candidates().read_active_frontier(context=context)
+
+    async def read_pinned_candidates(
+        self, context: Mapping[str, Any]
+    ) -> Sequence[Mapping[str, Any]]:
+        return await self._candidates().read_pinned_candidates(context=context)
+
+    async def read_attempt_workspace(self, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        return await self._candidates().read_attempt_workspace(context=context)
+
+    async def candidate_command(
+        self,
+        operation: str,
+        payload: Mapping[str, Any] | Sequence[Mapping[str, Any]],
+        context: Mapping[str, Any],
+    ) -> 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 == "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_comparison_candidate":
+            return await tools.save_comparison_candidate(
+                payload=_mapping_payload(payload), context=context
+            )
+        if operation == "save_candidate_portfolio":
+            return await tools.save_candidate_portfolio(
+                payload=_mapping_payload(payload), context=context
+            )
+        raise ProtocolViolation(f"unsupported candidate operation: {operation}")
+
+    async def save_structured_script_candidate(
+        self, acceptance_notes: Sequence[str], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        await self.ensure_dispatch_allowed(context)
+        notes = tuple(_bounded_text(item, "acceptance note", 500) for item in acceptance_notes)
+        if len(notes) > 16:
+            raise ProtocolViolation("a StructuredScript may contain at most 16 acceptance notes")
+        return await self._candidates().save_structured_script_candidate(
+            acceptance_notes=notes,
+            context=context,
+        )
+
+    async def submit_current_attempt(self, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        await self.ensure_dispatch_allowed(context)
+        if self.budget_guard is not None:
+            await self.budget_guard.ensure_tool_budget(
+                root_trace_id=_required(context, "root_trace_id"),
+                task_id=_required(context, "task_id"),
+                entry="submit",
+            )
+        manifest = await self._candidates().resolve_attempt_manifest(context=context)
+        ref = manifest.artifact_ref
+        if ref.kind not in _BUSINESS_ARTIFACT_KINDS or not ref.digest:
+            raise ProtocolViolation("attempt manifest must contain one frozen business artifact")
+        binding = await self.bindings.get_by_root(_required(context, "root_trace_id"))
+        version = await self.artifacts.read_by_ref(
+            ref,
+            script_build_id=binding.script_build_id,
+            task_id=_required(context, "task_id"),
+            attempt_id=_required(context, "attempt_id"),
+        )
+        if (
+            version.artifact_type.value != ref.kind
+            or version.canonical_sha256 != ref.digest
+            or version.state.value != "frozen"
+        ):
+            raise ProtocolViolation("attempt artifact ownership or frozen digest is invalid")
+        safe_ref = ArtifactRef(
+            uri=ref.uri,
+            kind=ref.kind,
+            version=ref.version,
+            digest=ref.digest,
+        )
+        summary = _attempt_summary(safe_ref, manifest.scope_ref)
+        submission = AttemptSubmission(
+            summary=summary,
+            artifact_refs=([] if safe_ref.kind == ArtifactKind.EVIDENCE.value else [safe_ref]),
+            evidence_refs=([safe_ref] if safe_ref.kind == ArtifactKind.EVIDENCE.value else []),
+        )
+        return cast(
+            dict[str, Any],
+            await self.coordinator.submit_attempt(dict(context), submission),
+        )
+
+    async def submit_structured_validation(
+        self,
+        *,
+        verdict: str,
+        criterion_results: Sequence[Mapping[str, Any]],
+        defects: Sequence[Mapping[str, Any]],
+        recommendation: str,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        root_trace_id = _required(context, "root_trace_id")
+        task_id = _required(context, "task_id")
+        snapshot_id = _required(context, "snapshot_id")
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        task = ledger.tasks.get(task_id)
+        if task is None:
+            raise ProtocolViolation("validation task is outside the protected mission")
+        snapshot = await self.coordinator.artifact_store.get(root_trace_id, snapshot_id)
+        allowed_refs = {
+            (ref.uri, ref.version, ref.digest, ref.kind): ref
+            for ref in (*snapshot.artifact_refs, *snapshot.evidence_refs)
+        }
+        normalized_defects = _normalize_defects(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")
+        result_by_id: dict[str, CriterionResult] = {}
+        for raw in criterion_results:
+            criterion_id = _bounded_text(raw.get("criterion_id"), "criterion_id", 128)
+            if criterion_id not in expected_ids or criterion_id in result_by_id:
+                raise ProtocolViolation("validation criterion identity is unknown or duplicated")
+            criterion_verdict = ValidationVerdict(str(raw.get("verdict", "")))
+            reason = _bounded_text(raw.get("reason"), "criterion reason", 300)
+            defect_codes = tuple(
+                item["defect_code"]
+                for item in normalized_defects
+                if item["criterion_id"] == criterion_id
+            )
+            if defect_codes:
+                reason = f"{reason}; defect_codes={','.join(defect_codes)}"[:500]
+            evidence = _dedupe_refs(
+                ref
+                for item in normalized_defects
+                if item["criterion_id"] == criterion_id
+                for ref in item["evidence_refs"]
+            )
+            result_by_id[criterion_id] = CriterionResult(
+                criterion_id=criterion_id,
+                verdict=criterion_verdict,
+                reason=reason,
+                evidence_refs=list(evidence),
+            )
+        if set(result_by_id) != expected_ids:
+            raise ProtocolViolation("validation must report every Task criterion exactly once")
+        overall = ValidationVerdict(verdict)
+        blocking = [item for item in normalized_defects if item["severity"] in {"hard", "critical"}]
+        if overall is ValidationVerdict.PASSED and blocking:
+            raise ProtocolViolation("passed validation cannot contain hard or critical defects")
+        from agent.orchestration.validation_policy import ValidationContext
+
+        attempt = ledger.attempts.get(_required(context, "attempt_id"))
+        if attempt is None or attempt.task_id != task_id:
+            raise ProtocolViolation("validation Attempt is outside the protected Task")
+        validation_context = ValidationContext(
+            root_trace_id=root_trace_id,
+            task=task,
+            attempt=attempt,
+            snapshot=snapshot,
+            prior_validation_count=max(0, len(task.validation_ids) - 1),
+        )
+        deterministic = precheck_snapshot(validation_context)
+        if overall is ValidationVerdict.PASSED and any(
+            item.verdict is not ValidationVerdict.PASSED for item in deterministic
+        ):
+            raise ProtocolViolation("passed validation conflicts with deterministic precheck")
+        if (
+            overall is ValidationVerdict.PASSED
+            and task_kind(task.current_spec.context_refs) in PHASE_TWO_KINDS
+        ):
+            binding = await self.bindings.get_by_root(root_trace_id)
+            for ref in snapshot.artifact_refs:
+                version = await self.artifacts.read_by_ref(
+                    ref,
+                    script_build_id=binding.script_build_id,
+                    task_id=task_id,
+                    attempt_id=attempt.attempt_id,
+                )
+                if precheck_business_artifact(version.artifact):
+                    raise ProtocolViolation(
+                        "passed validation conflicts with unrealized placeholder content"
+                    )
+            business_rules = await self._candidates().deterministic_precheck(context=context)
+            if any(item.get("verdict") != "passed" for item in business_rules):
+                raise ProtocolViolation(
+                    "passed validation conflicts with phase-two deterministic precheck"
+                )
+        if overall is not ValidationVerdict.PASSED and not normalized_defects:
+            raise ProtocolViolation("failed or inconclusive validation must report a defect")
+        if any(
+            result_by_id[item["criterion_id"]].verdict is ValidationVerdict.PASSED
+            for item in blocking
+        ):
+            raise ProtocolViolation("blocking defect conflicts with a passed criterion")
+        codes = tuple(dict.fromkeys(item["defect_code"] for item in normalized_defects))
+        scopes = tuple(dict.fromkeys(item["scope_ref"] for item in normalized_defects))
+        action_classes = tuple(
+            dict.fromkeys(item["recommended_action_class"] for item in normalized_defects)
+        )
+        summary = (
+            f"validation={overall.value};criteria={len(result_by_id)};"
+            f"defects={len(normalized_defects)};codes={','.join(codes) or 'none'};"
+            f"scopes={','.join(scopes) or 'none'};"
+            f"actions={','.join(action_classes) or 'none'}"
+        )[:500]
+        evidence_refs = _dedupe_refs(
+            ref for item in normalized_defects for ref in item["evidence_refs"]
+        )
+        bounded_recommendation = _bounded_text(
+            recommendation or (action_classes[0] if action_classes else "accept"),
+            "recommendation",
+            200,
+        )
+        return cast(
+            dict[str, Any],
+            await self.coordinator.submit_validation(
+                dict(context),
+                overall,
+                tuple(result_by_id.values()),
+                summary,
+                evidence_refs,
+                (),
+                codes[:50],
+                bounded_recommendation,
+            ),
+        )
+
     async def save_direction_candidate(
         self,
         *,
@@ -236,6 +655,7 @@ class LegacyScriptToolGateway:
         legacy_markdown: str,
         context: Mapping[str, Any],
     ) -> dict[str, Any]:
+        await self.ensure_dispatch_allowed(context)
         binding, _ = await self._scope(context)
         accepted = await self.read_accepted_artifacts(context)
         allowed_evidence = {item["artifact_ref"]["uri"] for item in accepted}
@@ -306,7 +726,11 @@ class LegacyScriptToolGateway:
     async def deterministic_precheck(self, context: Mapping[str, Any]) -> dict[str, Any]:
         from agent.orchestration.validation_policy import ValidationContext
 
-        from script_build_host.agents.validation import precheck_snapshot
+        from script_build_host.agents.validation import (
+            PHASE_TWO_KINDS,
+            precheck_snapshot,
+            task_kind,
+        )
 
         root_trace_id = _required(context, "root_trace_id")
         ledger = await self.coordinator.task_store.load(root_trace_id)
@@ -322,7 +746,11 @@ class LegacyScriptToolGateway:
             snapshot=snapshot,
             prior_validation_count=max(0, len(task.validation_ids) - 1),
         )
-        return {"rule_results": [item.to_dict() for item in precheck_snapshot(validation_context)]}
+        rules = [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))
+        return {"rule_results": rules}
 
     async def _scope(self, context: Mapping[str, Any]) -> tuple[Any, Any]:
         root_trace_id = _required(context, "root_trace_id")
@@ -333,6 +761,16 @@ class LegacyScriptToolGateway:
         )
         return binding, snapshot
 
+    def _planner(self) -> ScriptPlannerToolPort:
+        if self.planner_tools is None:
+            raise ProtocolViolation("script planner tools are not configured")
+        return self.planner_tools
+
+    def _candidates(self) -> ScriptCandidateToolPort:
+        if self.candidate_tools is None:
+            raise ProtocolViolation("script candidate tools are not configured")
+        return self.candidate_tools
+
 
 def _required(context: Mapping[str, Any], key: str) -> str:
     value = context.get(key)
@@ -375,10 +813,124 @@ def _safe_text_tuple(values: Sequence[str]) -> tuple[str, ...]:
     return tuple(redact_text(str(value)) for value in values)
 
 
+def _mapping_payload(
+    value: Mapping[str, Any] | Sequence[Mapping[str, Any]],
+) -> Mapping[str, Any]:
+    if not isinstance(value, Mapping):
+        raise ProtocolViolation("candidate command expects an object payload")
+    return value
+
+
+def _sequence_payload(
+    value: Mapping[str, Any] | Sequence[Mapping[str, Any]],
+) -> Sequence[Mapping[str, Any]]:
+    if isinstance(value, Mapping) or isinstance(value, (str, bytes)):
+        raise ProtocolViolation("candidate batch command expects an array payload")
+    if any(not isinstance(item, Mapping) for item in value):
+        raise ProtocolViolation("candidate batch entries must be objects")
+    return value
+
+
+def _bounded_text(value: Any, label: str, maximum: int) -> str:
+    if not isinstance(value, str) or not value.strip() or len(value) > maximum:
+        raise ProtocolViolation(f"{label} must contain between 1 and {maximum} characters")
+    return redact_text(value.strip())
+
+
+def _attempt_summary(ref: ArtifactRef, scope_ref: str) -> str:
+    scope = _bounded_text(scope_ref, "scope_ref", 300)
+    if not scope.startswith("script-build://"):
+        raise ProtocolViolation("scope_ref must use the script-build namespace")
+    digest = _bounded_text(ref.digest, "artifact digest", 71)
+    return f"kind={ref.kind};scope={scope};digest={digest};status=frozen"[:500]
+
+
+def _normalize_defects(
+    defects: Sequence[Mapping[str, Any]],
+    allowed_refs: Mapping[tuple[Any, Any, Any, Any], ArtifactRef],
+) -> tuple[dict[str, Any], ...]:
+    if len(defects) > 50:
+        raise ProtocolViolation("a validation may report at most 50 defects")
+    values: list[dict[str, Any]] = []
+    for raw in defects:
+        if not isinstance(raw, Mapping):
+            raise ProtocolViolation("each validation defect must be an object")
+        code = _bounded_text(raw.get("defect_code"), "defect_code", 64)
+        if not _DEFECT_CODE.fullmatch(code):
+            raise ProtocolViolation("defect_code must be an uppercase stable identifier")
+        criterion_id = _bounded_text(raw.get("criterion_id"), "criterion_id", 128)
+        scope_ref = _bounded_text(raw.get("scope_ref"), "scope_ref", 500)
+        if not scope_ref.startswith("script-build://"):
+            raise ProtocolViolation("defect scope_ref must use script-build://")
+        excerpt = _bounded_text(raw.get("observed_excerpt"), "observed_excerpt", 500)
+        severity = str(raw.get("severity", ""))
+        if severity not in _DEFECT_SEVERITIES:
+            raise ProtocolViolation("defect severity must be warning, hard, or critical")
+        action = str(raw.get("recommended_action_class", ""))
+        if action not in _DEFECT_ACTIONS:
+            raise ProtocolViolation("defect recommended action class is unsupported")
+        invalidated = raw.get("invalidated_inputs", [])
+        if (
+            not isinstance(invalidated, list)
+            or len(invalidated) > 64
+            or any(not isinstance(item, str) or not item.strip() for item in invalidated)
+        ):
+            raise ProtocolViolation("invalidated_inputs must be a bounded string array")
+        raw_refs = raw.get("evidence_refs", [])
+        if not isinstance(raw_refs, list) or len(raw_refs) > 64:
+            raise ProtocolViolation("defect evidence_refs must be a bounded array")
+        refs: list[ArtifactRef] = []
+        for item in raw_refs:
+            if not isinstance(item, Mapping):
+                raise ProtocolViolation("defect evidence reference must be an object")
+            candidate = ArtifactRef.from_dict(dict(item))
+            key = (candidate.uri, candidate.version, candidate.digest, candidate.kind)
+            frozen = allowed_refs.get(key)
+            if frozen is None:
+                raise ProtocolViolation(
+                    "defect evidence must come from the fixed validation snapshot"
+                )
+            refs.append(frozen)
+        values.append(
+            {
+                "defect_code": code,
+                "criterion_id": criterion_id,
+                "scope_ref": scope_ref,
+                "observed_excerpt": excerpt,
+                "evidence_refs": tuple(refs),
+                "severity": severity,
+                "invalidated_inputs": tuple(str(item) for item in invalidated),
+                "recommended_action_class": action,
+            }
+        )
+    return tuple(values)
+
+
+def _dedupe_refs(values: Any) -> tuple[ArtifactRef, ...]:
+    result: list[ArtifactRef] = []
+    seen: set[tuple[Any, Any, Any, Any]] = set()
+    for ref in values:
+        key = (ref.uri, ref.version, ref.digest, ref.kind)
+        if key in seen:
+            continue
+        seen.add(key)
+        result.append(
+            ArtifactRef(
+                uri=ref.uri,
+                kind=ref.kind,
+                version=ref.version,
+                digest=ref.digest,
+            )
+        )
+    return tuple(result)
+
+
 __all__ = [
     "DispatchGuard",
     "ImageAdapter",
     "LegacyScriptToolGateway",
     "RetrievalAdapter",
     "RetrievalResult",
+    "ScriptCandidateToolPort",
+    "ScriptPlannerToolPort",
 ]

script_build_host/src/script_build_host/tools/registry.py

@@ -2,13 +2,13 @@
 
 from __future__ import annotations
 
+import base64
+import hashlib
 import json
 from typing import Any
 
 from agent import ToolRegistry
-
-from script_build_host.agents.validation import RETRIEVAL_KINDS, task_kind
-from script_build_host.domain.errors import ProtocolViolation
+from agent.tools.models import ToolResult
 
 from .gateway import LegacyScriptToolGateway
 
@@ -18,6 +18,12 @@ TASK_PRESET_BY_KIND = {
     "decode-retrieval": "script_decode_retrieval_worker",
     "external-retrieval": "script_external_retrieval_worker",
     "knowledge-retrieval": "script_knowledge_retrieval_worker",
+    "structure": "script_structure_worker",
+    "paragraph": "script_paragraph_worker",
+    "element-set": "script_element_set_worker",
+    "compare": "script_candidate_compare_worker",
+    "compose": "script_compose_worker",
+    "candidate-portfolio": "script_candidate_portfolio_worker",
 }
 
 
@@ -35,6 +41,13 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, read_input_snapshot, capabilities=["read"])
 
+    async def load_frozen_strategy(strategy_ref: str, context: dict[str, Any] | None = None) -> str:
+        """Load one on-demand strategy explicitly pinned by this Task."""
+
+        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."""
 
@@ -42,38 +55,91 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, read_accepted_artifact, capabilities=["read"])
 
+    async def plan_script_tasks(
+        contracts: list[dict[str, Any]],
+        parent_task_id: str | None = None,
+        context: dict[str, Any] | None = None,
+    ) -> str | ToolResult:
+        """Validate and atomically create Tasks from complete business contracts."""
+
+        return _json(
+            await gateway.plan_script_tasks(
+                contracts=contracts,
+                parent_task_id=parent_task_id,
+                context=context or {},
+            )
+        )
+
+    _register(registry, plan_script_tasks, capabilities=["task_control"])
+
+    async def decide_script_task(
+        task_id: str,
+        action: str,
+        reason: str,
+        validation_id: str | None = None,
+        replacement_contract: dict[str, Any] | None = None,
+        child_contracts: list[dict[str, Any]] | None = None,
+        context: dict[str, Any] | None = None,
+    ) -> str | ToolResult:
+        """Apply one phase-aware Planner decision after contract and boundary checks."""
+
+        if action not in {
+            "accept",
+            "repair",
+            "retry",
+            "revise",
+            "split",
+            "block",
+            "supersede",
+            "cancel",
+        }:
+            raise ValueError("action is not supported by decide_script_task")
+        result = await gateway.decide_script_task(
+            task_id=task_id,
+            action=action,
+            reason=reason,
+            validation_id=validation_id,
+            replacement_contract=replacement_contract,
+            child_contracts=child_contracts or [],
+            context=context or {},
+        )
+        result_json = _json(result)
+        if result.get("terminal_boundary") is True:
+            phase_two = result.get("phase_boundary") == "PHASE_TWO_CANDIDATE_PORTFOLIO_READY"
+            return ToolResult(
+                title="Phase two checkpoint ready" if phase_two else "Phase one checkpoint ready",
+                output=result_json,
+                long_term_memory=(
+                    "Phase two candidate portfolio checkpoint is ready"
+                    if phase_two
+                    else "Phase one accepted direction checkpoint is ready"
+                ),
+                terminate_run=True,
+                result_summary=result_json,
+            )
+        return result_json
+
+    _register(registry, decide_script_task, capabilities=["task_control"])
+
+    async def inspect_script_plan(
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Inspect a bounded business view of the current Task tree and contracts."""
+
+        return _json(await gateway.inspect_script_plan(context=context or {}))
+
+    _register(registry, inspect_script_plan, capabilities=["read"])
+
     async def dispatch_script_tasks(
         task_ids: list[str], context: dict[str, Any] | None = None
     ) -> str:
         """Dispatch allowlisted script Tasks through durable operations."""
 
-        ctx = context or {}
-        root_trace_id = _context_id(ctx, "root_trace_id")
-        await gateway.ensure_dispatch_allowed(ctx)
-        ledger = await gateway.coordinator.task_store.load(root_trace_id)
-        presets: list[str] = []
-        for task_id in task_ids:
-            task = ledger.tasks.get(task_id)
-            if task is None:
-                raise ProtocolViolation("task is outside this mission")
-            kind = task_kind(task.current_spec.context_refs)
-            preset = TASK_PRESET_BY_KIND.get(kind or "")
-            if preset is None:
-                raise ProtocolViolation(f"task kind is not dispatchable in phase one: {kind!r}")
-            if kind in RETRIEVAL_KINDS:
-                parent = ledger.tasks.get(task.parent_task_id or "")
-                if parent is None or task_kind(parent.current_spec.context_refs) != "direction":
-                    raise ProtocolViolation(
-                        "retrieval tasks must be direct children of the direction task"
-                    )
-            presets.append(preset)
-        results = await gateway.coordinator.dispatch_tasks(
-            root_trace_id,
-            task_ids,
-            worker_presets=presets,
-            idempotency_key=ctx.get("tool_call_id"),
+        results = await gateway.dispatch_script_tasks(
+            task_ids=task_ids,
+            context=context or {},
         )
-        return _json([item.to_dict() for item in results])
+        return _json(results)
 
     _register(registry, dispatch_script_tasks, capabilities=["task_control"])
 
@@ -131,7 +197,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
     _register(
         registry,
         search_script_decode_case,
-        capabilities=["read", "write"],
+        capabilities=["external_send", "read", "write"],
         schema=_decode_schema(),
     )
 
@@ -172,7 +238,7 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
         return _json(await gateway.load_images(image_urls, context or {}))
 
-    _register(registry, load_images, capabilities=["external_send", "read"])
+    _register(registry, load_images, capabilities=["external_send", "read", "write"])
 
     async def search_knowledge(
         keyword: str,
@@ -228,6 +294,270 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, save_direction_candidate, capabilities=["write"])
 
+    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],
+        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 {})
+        images, manifest = _validated_images(values)
+        output = _json({"images": manifest})
+        return ToolResult(
+            title="Frozen images",
+            output=output,
+            long_term_memory=f"Loaded {len(images)} frozen image(s)",
+            include_output_only_once=True,
+            images=images,
+        )
+
+    _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."""
+
+        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 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."""
+
+        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",
+                {
+                    "paragraph_id": paragraph_id,
+                    "column": column,
+                    "dimension": dimension,
+                    "atom_value": atom_value,
+                },
+                context or {},
+            )
+        )
+
+    _register(registry, delete_paragraph_atom, capabilities=["write"])
+
+    async def batch_update_script_paragraphs(
+        updates: list[dict[str, Any]],
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Atomically validate and update a Paragraph batch in this workspace."""
+
+        return _json(
+            await gateway.candidate_command(
+                "batch_update_script_paragraphs", updates, context or {}
+            )
+        )
+
+    _register(registry, batch_update_script_paragraphs, capabilities=["write"])
+
+    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."""
+
+        return _json(
+            await gateway.candidate_command(
+                "create_script_element",
+                {
+                    "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 [],
+                },
+                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."""
+
+        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."""
+
+        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."""
+
+        return _json(
+            await gateway.candidate_command("delete_paragraph_element_links", links, context or {})
+        )
+
+    _register(registry, delete_paragraph_element_links, capabilities=["write"])
+
+    async def save_comparison_candidate(
+        criterion_results: list[dict[str, Any]],
+        recommendation: str,
+        conflicts: list[str] | None = None,
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Freeze one comparison over the candidates pinned by the Task contract."""
+
+        return _json(
+            await gateway.candidate_command(
+                "save_comparison_candidate",
+                {
+                    "criterion_results": criterion_results,
+                    "conflicts": conflicts or [],
+                    "recommendation": recommendation,
+                },
+                context or {},
+            )
+        )
+
+    _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."""
+
+        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 query_validation_evidence(
         query: str, limit: int = 10, context: dict[str, Any] | None = None
     ) -> str:
@@ -244,6 +574,50 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
 
     _register(registry, deterministic_precheck, capabilities=["read"])
 
+    async def submit_attempt(context: dict[str, Any] | None = None) -> ToolResult:
+        """Submit the one frozen business output owned by this protected Attempt."""
+
+        result = await gateway.submit_current_attempt(context or {})
+        result_json = _json(result)
+        return ToolResult(
+            title="Attempt submitted",
+            output=result_json,
+            long_term_memory=f"Attempt {result['attempt_id']} submitted",
+            terminate_run=True,
+            result_summary=result_json,
+        )
+
+    _register(registry, submit_attempt, capabilities=["attempt_submit"])
+
+    async def submit_validation(
+        verdict: str,
+        criterion_results: list[dict[str, Any]],
+        defects: list[dict[str, Any]],
+        recommendation: str = "",
+        context: dict[str, Any] | None = None,
+    ) -> ToolResult:
+        """Submit bounded criterion results plus structured, evidence-bound defects."""
+
+        result = await gateway.submit_structured_validation(
+            verdict=verdict,
+            criterion_results=criterion_results,
+            defects=defects,
+            recommendation=recommendation,
+            context=context or {},
+        )
+        result_json = _json(result)
+        return ToolResult(
+            title="Validation submitted",
+            output=result_json,
+            long_term_memory=(
+                f"Validation {result['validation_id']} submitted: {result['verdict']}"
+            ),
+            terminate_run=True,
+            result_summary=result_json,
+        )
+
+    _register(registry, submit_validation, capabilities=["validation_submit"])
+
 
 def _register(
     registry: ToolRegistry,
@@ -341,15 +715,58 @@ def _knowledge_schema() -> dict[str, Any]:
     }
 
 
-def _context_id(context: dict[str, Any], key: str) -> str:
-    value = context.get(key)
-    if not isinstance(value, str) or not value:
-        raise ProtocolViolation(f"protected context is missing {key}")
-    return value
-
-
 def _json(value: Any) -> str:
     return json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
 
 
+def _validated_images(
+    values: Any,
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+    if not isinstance(values, (list, tuple)) or len(values) > 20:
+        raise ValueError("view_frozen_images returned an invalid image collection")
+    images: list[dict[str, Any]] = []
+    manifest: list[dict[str, Any]] = []
+    total_bytes = 0
+    for index, item in enumerate(values):
+        if not isinstance(item, dict) or item.get("type") != "base64":
+            raise ValueError("frozen images must be base64 image objects")
+        if "url" in item or "image_url" in item:
+            raise ValueError("frozen images must not contain network URLs")
+        media_type = item.get("media_type")
+        if media_type not in {"image/png", "image/jpeg", "image/webp"}:
+            raise ValueError("frozen image MIME type is unsupported")
+        data = item.get("data")
+        if not isinstance(data, str):
+            raise ValueError("frozen image data must be base64 text")
+        try:
+            content = base64.b64decode(data, validate=True)
+        except ValueError as exc:
+            raise ValueError("frozen image data is not valid base64") from exc
+        total_bytes += len(content)
+        if len(content) > 20 * 1024 * 1024 or total_bytes > 50 * 1024 * 1024:
+            raise ValueError("frozen image bytes exceed the defensive tool limit")
+        digest = f"sha256:{hashlib.sha256(content).hexdigest()}"
+        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")
+        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,
+            }
+        )
+    return images, manifest
+
+
 __all__ = ["TASK_PRESET_BY_KIND", "register_script_tools"]