代理：注册阶段一角色、验证器和最小工具面

新增 Planner、Direction、四类 Retrieval Worker 和两个独立 Validator preset，不注册旧 Round、Branch 或 multipath 角色。

按受保护上下文校验任务类型、直接子任务和 Artifact 归属，提供幂等派发、候选冻结、证据查询与 Validation 提交工具。

script_build_host/src/script_build_host/agents/__init__.py

@@ -0,0 +1,19 @@
+"""Script-build Agent presets and validation integration."""
+
+from .model_resolver import ScriptRoleRunConfigResolver
+from .presets import register_script_presets
+from .validation import (
+    ScriptBuildArtifactEvidenceReader,
+    ScriptBuildDeterministicValidator,
+    ScriptBuildEvidenceProvider,
+    ScriptBuildValidationPolicy,
+)
+
+__all__ = [
+    "ScriptBuildArtifactEvidenceReader",
+    "ScriptBuildDeterministicValidator",
+    "ScriptBuildEvidenceProvider",
+    "ScriptBuildValidationPolicy",
+    "ScriptRoleRunConfigResolver",
+    "register_script_presets",
+]

script_build_host/src/script_build_host/agents/model_resolver.py

@@ -0,0 +1,113 @@
+"""Frozen per-role model selection for script-build sub-traces."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+from agent import AgentRole, RoleRunConfigOverrides
+
+
+@dataclass(frozen=True)
+class ScriptRoleModelConfig:
+    model: str
+    temperature: float | None = None
+    max_iterations: int | None = None
+
+    def __post_init__(self) -> None:
+        if not self.model.strip():
+            raise ValueError("model must not be blank")
+
+
+class ModelManifestSource(Protocol):
+    async def load(self, root_trace_id: str) -> Mapping[str, Any]: ...
+
+
+class SnapshotModelManifestSource:
+    """Resolve the model manifest through root binding -> frozen input."""
+
+    def __init__(self, bindings: Any, snapshots: Any) -> None:
+        self._bindings = bindings
+        self._snapshots = snapshots
+
+    async def load(self, root_trace_id: str) -> Mapping[str, Any]:
+        binding = await self._bindings.get_by_root(root_trace_id)
+        snapshot = await self._snapshots.get(
+            str(binding.input_snapshot_id), script_build_id=binding.script_build_id
+        )
+        return dict(snapshot.model_manifest)
+
+
+class ScriptRoleRunConfigResolver:
+    """Resolve only the framework-approved per-role override fields.
+
+    A copy of the mapping is frozen at construction time so a running mission
+    cannot drift when process configuration is later mutated.
+    """
+
+    def __init__(
+        self,
+        configs: Mapping[str, ScriptRoleModelConfig | Mapping[str, Any]],
+        *,
+        default: ScriptRoleModelConfig | Mapping[str, Any] | None = None,
+        manifest_source: ModelManifestSource | None = None,
+    ) -> None:
+        self._configs = {key: self._coerce(value) for key, value in dict(configs).items()}
+        self._default = self._coerce(default) if default is not None else None
+        self._manifest_source = manifest_source
+
+    async def resolve(
+        self,
+        *,
+        role: AgentRole,
+        preset: str,
+        context: Mapping[str, Any],
+    ) -> RoleRunConfigOverrides:
+        del role
+        configs = self._configs
+        if self._manifest_source is not None:
+            root_trace_id = context.get("root_trace_id")
+            if not isinstance(root_trace_id, str) or not root_trace_id:
+                raise ValueError("root_trace_id is required to resolve the frozen model manifest")
+            manifest = dict(await self._manifest_source.load(root_trace_id))
+            raw_presets = manifest.get("presets", manifest)
+            if not isinstance(raw_presets, Mapping):
+                raise ValueError("frozen model_manifest.presets must be an object")
+            configs = {
+                str(key): self._coerce(value)
+                for key, value in raw_presets.items()
+                if isinstance(value, (ScriptRoleModelConfig, Mapping))
+            }
+        selected = configs.get(preset, self._default)
+        if selected is None:
+            return RoleRunConfigOverrides()
+        return RoleRunConfigOverrides(
+            model=selected.model,
+            temperature=selected.temperature,
+            max_iterations=selected.max_iterations,
+        )
+
+    @staticmethod
+    def _coerce(
+        value: ScriptRoleModelConfig | Mapping[str, Any],
+    ) -> ScriptRoleModelConfig:
+        if isinstance(value, ScriptRoleModelConfig):
+            return value
+        return ScriptRoleModelConfig(
+            model=str(value["model"]),
+            temperature=(
+                float(value["temperature"]) if value.get("temperature") is not None else None
+            ),
+            max_iterations=(
+                int(value["max_iterations"]) if value.get("max_iterations") is not None else None
+            ),
+        )
+
+
+__all__ = [
+    "ModelManifestSource",
+    "ScriptRoleModelConfig",
+    "ScriptRoleRunConfigResolver",
+    "SnapshotModelManifestSource",
+]

script_build_host/src/script_build_host/agents/presets.py

@@ -0,0 +1,133 @@
+"""Exact phase-one preset allowlists."""
+
+from __future__ import annotations
+
+from agent import AgentPreset, AgentRole
+from agent.core.presets import register_preset
+
+from .prompts import (
+    DIRECTION_WORKER_PROMPT,
+    PLANNER_PROMPT,
+    RETRIEVAL_VALIDATOR_PROMPT,
+    RETRIEVAL_WORKER_PROMPT,
+    SCRIPT_CANDIDATE_VALIDATOR_PROMPT,
+)
+
+_FORBIDDEN = [
+    "agent",
+    "evaluate",
+    "goal",
+    "read_file",
+    "write_file",
+    "edit_file",
+    "bash_command",
+    "dispatch_tasks",
+]
+
+
+def _worker(tools: list[str], *, direction: bool = False) -> AgentPreset:
+    return AgentPreset(
+        role=AgentRole.WORKER,
+        allowed_tools=[*tools, "submit_attempt"],
+        denied_tools=[
+            *_FORBIDDEN,
+            "task_plan",
+            "task_decide",
+            "validate_attempt",
+            "submit_validation",
+        ],
+        max_iterations=20,
+        skills=[],
+        system_prompt=DIRECTION_WORKER_PROMPT if direction else RETRIEVAL_WORKER_PROMPT,
+    )
+
+
+def register_script_presets() -> None:
+    """Register only the new Agentic script-build roles.
+
+    This function deliberately does not import or register any legacy Round,
+    branch, implementer, evaluator, or multipath role.
+    """
+
+    register_preset(
+        "script_planner",
+        AgentPreset(
+            role=AgentRole.PLANNER,
+            allowed_tools=[
+                "task_plan",
+                "dispatch_script_tasks",
+                "task_decide",
+                "validate_attempt",
+                "read_input_snapshot",
+            ],
+            denied_tools=[*_FORBIDDEN, "submit_attempt", "submit_validation"],
+            max_iterations=100,
+            skills=[],
+            system_prompt=PLANNER_PROMPT,
+        ),
+    )
+    register_preset(
+        "script_pattern_retrieval_worker",
+        _worker(["read_input_snapshot", "query_pattern_qa"]),
+    )
+    register_preset(
+        "script_decode_retrieval_worker",
+        _worker(["read_input_snapshot", "search_script_decode_case"]),
+    )
+    register_preset(
+        "script_external_retrieval_worker",
+        _worker(["read_input_snapshot", "external_search_case", "load_images"]),
+    )
+    register_preset(
+        "script_knowledge_retrieval_worker",
+        _worker(["read_input_snapshot", "search_knowledge"]),
+    )
+    register_preset(
+        "script_direction_worker",
+        _worker(
+            ["read_input_snapshot", "read_accepted_artifact", "save_direction_candidate"],
+            direction=True,
+        ),
+    )
+    validator_tools = [
+        "read_input_snapshot",
+        "query_validation_evidence",
+        "deterministic_precheck",
+        "submit_validation",
+    ]
+    validator_denied = [
+        *_FORBIDDEN,
+        "task_plan",
+        "dispatch_script_tasks",
+        "task_decide",
+        "validate_attempt",
+        "submit_attempt",
+        "save_direction_candidate",
+    ]
+    register_preset(
+        "script_retrieval_validator",
+        AgentPreset(
+            role=AgentRole.VALIDATOR,
+            allowed_tools=validator_tools,
+            denied_tools=validator_denied,
+            max_iterations=20,
+            temperature=0.0,
+            skills=[],
+            system_prompt=RETRIEVAL_VALIDATOR_PROMPT,
+        ),
+    )
+    register_preset(
+        "script_candidate_validator",
+        AgentPreset(
+            role=AgentRole.VALIDATOR,
+            allowed_tools=validator_tools,
+            denied_tools=validator_denied,
+            max_iterations=25,
+            temperature=0.0,
+            skills=[],
+            system_prompt=SCRIPT_CANDIDATE_VALIDATOR_PROMPT,
+        ),
+    )
+
+
+__all__ = ["register_script_presets"]

script_build_host/src/script_build_host/agents/prompts/__init__.py

@@ -0,0 +1,19 @@
+"""Immutable prompt contracts used by script-build presets."""
+
+from .contracts import (
+    DIRECTION_WORKER_PROMPT,
+    PLANNER_PROMPT,
+    RETRIEVAL_VALIDATOR_PROMPT,
+    RETRIEVAL_WORKER_PROMPT,
+    SCRIPT_CANDIDATE_VALIDATOR_PROMPT,
+    phase_one_prompt_manifest,
+)
+
+__all__ = [
+    "DIRECTION_WORKER_PROMPT",
+    "PLANNER_PROMPT",
+    "RETRIEVAL_VALIDATOR_PROMPT",
+    "RETRIEVAL_WORKER_PROMPT",
+    "SCRIPT_CANDIDATE_VALIDATOR_PROMPT",
+    "phase_one_prompt_manifest",
+]

script_build_host/src/script_build_host/agents/prompts/candidate_validator.md

@@ -0,0 +1,4 @@
+Independently validate the frozen direction candidate. Check schema and digest,
+topic grounding, persona and strategy grounding, accepted evidence closure,
+and whether each direction goal is actionable and testable. Use only protected
+read tools and submit exactly one validation report. Never publish or mutate.

script_build_host/src/script_build_host/agents/prompts/contracts.py

@@ -0,0 +1,72 @@
+"""Role contracts for the phase-one script-build mission.
+
+The prompts describe capabilities rather than a fixed workflow.  Identity and
+ownership are enforced by protected tool context, not by prompt text.
+"""
+
+from pathlib import Path
+
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+
+_ROOT = Path(__file__).resolve().parent
+
+
+def _read(filename: str) -> str:
+    return (_ROOT / filename).read_text(encoding="utf-8")
+
+
+PLANNER_PROMPT = _read("script_planner.md")
+RETRIEVAL_WORKER_PROMPT = _read("retrieval_worker.md")
+DIRECTION_WORKER_PROMPT = _read("direction_worker.md")
+RETRIEVAL_VALIDATOR_PROMPT = _read("retrieval_validator.md")
+SCRIPT_CANDIDATE_VALIDATOR_PROMPT = _read("candidate_validator.md")
+
+
+def phase_one_prompt_manifest() -> tuple[dict[str, object], ...]:
+    values = {
+        "script_planner": ("planner", "script_planner.md", PLANNER_PROMPT),
+        "script_direction_worker": ("worker", "direction_worker.md", DIRECTION_WORKER_PROMPT),
+        "script_pattern_retrieval_worker": (
+            "worker",
+            "retrieval_worker.md",
+            RETRIEVAL_WORKER_PROMPT,
+        ),
+        "script_decode_retrieval_worker": (
+            "worker",
+            "retrieval_worker.md",
+            RETRIEVAL_WORKER_PROMPT,
+        ),
+        "script_external_retrieval_worker": (
+            "worker",
+            "retrieval_worker.md",
+            RETRIEVAL_WORKER_PROMPT,
+        ),
+        "script_knowledge_retrieval_worker": (
+            "worker",
+            "retrieval_worker.md",
+            RETRIEVAL_WORKER_PROMPT,
+        ),
+        "script_retrieval_validator": (
+            "validator",
+            "retrieval_validator.md",
+            RETRIEVAL_VALIDATOR_PROMPT,
+        ),
+        "script_candidate_validator": (
+            "validator",
+            "candidate_validator.md",
+            SCRIPT_CANDIDATE_VALIDATOR_PROMPT,
+        ),
+    }
+    return tuple(
+        {
+            "biz_type": preset,
+            "preset": preset,
+            "role": role,
+            "source": "file",
+            "source_id": filename,
+            "version": 1,
+            "content": content,
+            "content_sha256": canonical_sha256(content).wire,
+        }
+        for preset, (role, filename, content) in values.items()
+    )

script_build_host/src/script_build_host/agents/prompts/direction_worker.md

@@ -0,0 +1,5 @@
+Create ScriptDirectionArtifactV1 for the current TaskSpec from the frozen input
+and only the direct-child ACCEPT decisions frozen into this attempt. Produce
+one to three concrete, testable goals, with topic/persona/strategy/evidence
+reasons. Save the immutable direction candidate and submit exactly one attempt.
+Never read a latest/unaccepted candidate and never publish legacy state.

script_build_host/src/script_build_host/agents/prompts/retrieval_validator.md

@@ -0,0 +1,4 @@
+Independently validate the frozen retrieval attempt against every acceptance
+criterion. You may read the frozen input, run deterministic prechecks, and use
+query_validation_evidence; validation identity is supplied by protected
+context. Submit exactly one structured validation report. Never mutate data.

script_build_host/src/script_build_host/agents/prompts/retrieval_worker.md

@@ -0,0 +1,5 @@
+Execute exactly one retrieval TaskSpec. Read only the frozen input and call the
+single retrieval capability exposed by this preset. Freeze the normalized
+EvidenceRecordV1 returned by the gateway, then submit exactly one attempt with
+that immutable evidence reference. Do not create tasks, agents, files, shell
+commands, database queries, or arbitrary network requests.

script_build_host/src/script_build_host/agents/prompts/script_planner.md

@@ -0,0 +1,8 @@
+You are the sole planner for one script-build mission. Inspect the task ledger,
+identify the next missing capability, and create or revise tasks dynamically.
+Retrieval tasks that a direction attempt consumes must be direct children of
+the direction task. Dispatch only through dispatch_script_tasks. Read every
+independent validation report before accepting, repairing, retrying, revising,
+splitting, or blocking a task. Never create a fixed Round, Branch, multipath,
+or precomputed workflow. Phase one ends after an accepted direction: block the
+Root with reason PHASE_ONE_CAPABILITY_BOUNDARY; never claim final script success.

script_build_host/src/script_build_host/agents/validation.py

@@ -0,0 +1,259 @@
+"""Validation planning, deterministic checks, and guarded evidence reads."""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict
+from datetime import datetime
+from enum import Enum
+from typing import Any, Protocol
+
+from agent.orchestration import (
+    ArtifactRef,
+    EvidenceProviderResult,
+    EvidenceQuery,
+    ValidationMode,
+    ValidationPlan,
+)
+from agent.orchestration.models import ValidationVerdict
+from agent.orchestration.validation_policy import (
+    DeterministicRuleResult,
+    DeterministicValidationResult,
+    ValidationContext,
+)
+
+from script_build_host.domain.artifacts import ArtifactKind
+from script_build_host.domain.ports import (
+    MissionBindingRepository,
+    ScriptBusinessArtifactRepository,
+)
+
+TASK_KIND_PREFIX = "script-build://task-kinds/"
+RETRIEVAL_KINDS = frozenset(
+    {
+        "pattern-retrieval",
+        "decode-retrieval",
+        "external-retrieval",
+        "knowledge-retrieval",
+    }
+)
+_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
+
+
+def task_kind(context_refs: Sequence[str]) -> str | None:
+    values = [
+        item.removeprefix(TASK_KIND_PREFIX)
+        for item in context_refs
+        if item.startswith(TASK_KIND_PREFIX)
+    ]
+    if len(values) > 1:
+        raise ValueError("TaskSpec contains multiple script-build task kinds")
+    return values[0] if values else None
+
+
+class ScriptBuildValidationPolicy:
+    """Choose the Validator from the controlled task-kind reference."""
+
+    def plan(self, context: ValidationContext) -> ValidationPlan:
+        kind = task_kind(context.task.current_spec.context_refs)
+        if kind in RETRIEVAL_KINDS:
+            preset = "script_retrieval_validator"
+        elif kind == "direction":
+            preset = "script_candidate_validator"
+        else:
+            raise ValueError(f"Unsupported phase-one task kind: {kind!r}")
+        return ValidationPlan(
+            mode=ValidationMode.AGENT,
+            validator_preset=preset,
+            max_evidence_queries=8,
+            max_evidence_items_per_query=25,
+            evidence_timeout_seconds=8.0,
+        )
+
+
+def precheck_snapshot(context: ValidationContext) -> list[DeterministicRuleResult]:
+    """Validate immutable reference shape without interpreting creative quality."""
+
+    refs = [*context.snapshot.artifact_refs, *context.snapshot.evidence_refs]
+    results: list[DeterministicRuleResult] = []
+    results.append(
+        DeterministicRuleResult(
+            rule_id="immutable-reference-present",
+            verdict=ValidationVerdict.PASSED if refs else ValidationVerdict.FAILED,
+            reason="Attempt contains immutable references"
+            if refs
+            else "Attempt contains no immutable references",
+        )
+    )
+    bad = [ref.uri for ref in refs if ref.digest is None or not _DIGEST.fullmatch(ref.digest)]
+    results.append(
+        DeterministicRuleResult(
+            rule_id="sha256-digest-shape",
+            verdict=ValidationVerdict.FAILED if bad else ValidationVerdict.PASSED,
+            reason=(
+                f"Invalid digest on {len(bad)} reference(s)"
+                if bad
+                else "All references carry canonical sha256 digests"
+            ),
+        )
+    )
+    kind = task_kind(context.task.current_spec.context_refs)
+    if kind in RETRIEVAL_KINDS:
+        shape_ok = (
+            not context.snapshot.artifact_refs
+            and len(context.snapshot.evidence_refs) == 1
+            and context.snapshot.evidence_refs[0].kind == ArtifactKind.EVIDENCE.value
+        )
+        expected = "one evidence reference"
+    elif kind == "direction":
+        shape_ok = (
+            len(context.snapshot.artifact_refs) == 1
+            and not context.snapshot.evidence_refs
+            and context.snapshot.artifact_refs[0].kind == ArtifactKind.DIRECTION.value
+        )
+        expected = "one direction artifact reference"
+    else:
+        shape_ok = False
+        expected = "a supported phase-one task kind"
+    results.append(
+        DeterministicRuleResult(
+            rule_id="phase-one-artifact-shape",
+            verdict=ValidationVerdict.PASSED if shape_ok else ValidationVerdict.FAILED,
+            reason=(
+                f"Attempt contains {expected}" if shape_ok else f"Attempt must contain {expected}"
+            ),
+        )
+    )
+    return results
+
+
+class ScriptBuildDeterministicValidator:
+    """Optional deterministic validator for explicitly selected rule plans."""
+
+    async def validate(
+        self,
+        context: ValidationContext,
+        plan: ValidationPlan,
+    ) -> DeterministicValidationResult:
+        if plan.mode != ValidationMode.DETERMINISTIC:
+            raise ValueError("deterministic validator requires deterministic mode")
+        available = {item.rule_id: item for item in precheck_snapshot(context)}
+        selected = [
+            available.get(
+                rule_id,
+                DeterministicRuleResult(
+                    rule_id=rule_id,
+                    verdict=ValidationVerdict.INCONCLUSIVE,
+                    reason="Rule is not registered",
+                    error="unknown_rule",
+                ),
+            )
+            for rule_id in plan.rule_ids
+        ]
+        verdicts = {item.verdict for item in selected}
+        if ValidationVerdict.FAILED in verdicts:
+            verdict = ValidationVerdict.FAILED
+        elif selected and verdicts == {ValidationVerdict.PASSED}:
+            verdict = ValidationVerdict.PASSED
+        else:
+            verdict = ValidationVerdict.INCONCLUSIVE
+        return DeterministicValidationResult(verdict=verdict, rule_results=selected)
+
+
+class EvidenceContentReader(Protocol):
+    async def read(
+        self,
+        ref: ArtifactRef,
+        request: EvidenceQuery,
+    ) -> Mapping[str, Any]: ...
+
+
+class ScriptBuildArtifactEvidenceReader:
+    """Resolve an immutable business ArtifactRef using its protected build identity."""
+
+    def __init__(
+        self,
+        artifacts: ScriptBusinessArtifactRepository,
+        bindings: MissionBindingRepository,
+    ) -> None:
+        self._artifacts = artifacts
+        self._bindings = bindings
+
+    async def read(
+        self,
+        ref: ArtifactRef,
+        request: EvidenceQuery,
+    ) -> Mapping[str, Any]:
+        binding = await self._bindings.get_by_root(request.root_trace_id)
+        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,
+        )
+        return {
+            "uri": ref.uri,
+            "kind": ref.kind,
+            "version": ref.version,
+            "digest": ref.digest,
+            "artifact": _json_value(asdict(version.artifact)),
+        }
+
+
+def _json_value(value: Any) -> Any:
+    if isinstance(value, datetime):
+        return value.isoformat()
+    if isinstance(value, Enum):
+        return value.value
+    if isinstance(value, Mapping):
+        return {str(key): _json_value(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_json_value(item) for item in value]
+    return value
+
+
+class ScriptBuildEvidenceProvider:
+    """Read evidence only from the Validator's already-frozen snapshot."""
+
+    def __init__(self, artifact_store: Any, reader: EvidenceContentReader | None = None) -> None:
+        self._artifact_store = artifact_store
+        self._reader = reader
+
+    async def query(self, request: EvidenceQuery) -> EvidenceProviderResult:
+        snapshot = await self._artifact_store.get(request.root_trace_id, request.snapshot_id)
+        candidates = [*snapshot.evidence_refs, *snapshot.artifact_refs]
+        needle = request.query.casefold().strip()
+        refs = [
+            ref
+            for ref in candidates
+            if not needle
+            or needle in " ".join([ref.kind, ref.summary or "", str(ref.metadata)]).casefold()
+        ][: request.limit]
+        items: list[dict[str, Any]] = []
+        for ref in refs:
+            if self._reader is None:
+                items.append(
+                    {
+                        "uri": ref.uri,
+                        "kind": ref.kind,
+                        "version": ref.version,
+                        "digest": ref.digest,
+                        "summary": ref.summary,
+                    }
+                )
+            else:
+                items.append(dict(await self._reader.read(ref, request)))
+        return EvidenceProviderResult(items=items, evidence_refs=refs)
+
+
+__all__ = [
+    "RETRIEVAL_KINDS",
+    "TASK_KIND_PREFIX",
+    "ScriptBuildArtifactEvidenceReader",
+    "ScriptBuildDeterministicValidator",
+    "ScriptBuildEvidenceProvider",
+    "ScriptBuildValidationPolicy",
+    "precheck_snapshot",
+    "task_kind",
+]

script_build_host/src/script_build_host/tools/__init__.py

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

script_build_host/src/script_build_host/tools/gateway.py

@@ -0,0 +1,384 @@
+"""Application gateway used by all script-build Agent tools.
+
+Agent-visible functions never receive a build, root, task, attempt, or
+validation identity from model arguments.  Those values are resolved from the
+framework's protected context and then checked against durable bindings.
+"""
+
+from __future__ import annotations
+
+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 script_build_host.agents.validation import RETRIEVAL_KINDS, task_kind
+from script_build_host.domain.artifacts import (
+    ArtifactKind,
+    Criterion,
+    DirectionGoal,
+    EvidenceRecordV1,
+    ScriptDirectionArtifactV1,
+)
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.ports import (
+    InputSnapshotRepository,
+    MissionBindingRepository,
+    ScriptBusinessArtifactRepository,
+)
+from script_build_host.infrastructure.redaction import redact, redact_text
+
+
+@dataclass(frozen=True, slots=True)
+class RetrievalResult:
+    source_refs: tuple[str, ...]
+    summary: str
+    supports: tuple[str, ...] = ()
+    confidence: str = "unknown"
+    limitations: tuple[str, ...] = ()
+    raw_artifact_ref: str | None = None
+    metadata: Mapping[str, Any] | None = None
+
+
+class RetrievalAdapter(Protocol):
+    async def retrieve(
+        self,
+        *,
+        query: Mapping[str, Any],
+        snapshot: Any,
+    ) -> RetrievalResult: ...
+
+
+class ImageAdapter(Protocol):
+    async def load(self, *, urls: Sequence[str], snapshot: Any) -> tuple[dict[str, Any], ...]: ...
+
+
+class DispatchGuard(Protocol):
+    async def ensure_dispatch_allowed(self, script_build_id: int) -> None: ...
+
+
+class LegacyScriptToolGateway:
+    """Translate protected Agent tool calls to typed domain operations."""
+
+    def __init__(
+        self,
+        *,
+        bindings: MissionBindingRepository,
+        snapshots: InputSnapshotRepository,
+        artifacts: ScriptBusinessArtifactRepository,
+        coordinator: Any,
+        retrieval_adapters: Mapping[str, RetrievalAdapter] | None = None,
+        image_adapter: ImageAdapter | None = None,
+        dispatch_guard: DispatchGuard | None = None,
+    ) -> None:
+        self.bindings = bindings
+        self.snapshots = snapshots
+        self.artifacts = artifacts
+        self.coordinator = coordinator
+        self.retrieval_adapters = dict(retrieval_adapters or {})
+        self.image_adapter = image_adapter
+        self.dispatch_guard = dispatch_guard
+
+    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 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
+        return cast(dict[str, Any], _json_values(payload))
+
+    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)
+        attempt_id = _required(context, "attempt_id")
+        attempt = ledger.attempts.get(attempt_id)
+        if attempt is None or attempt.task_id != _required(context, "task_id"):
+            raise ProtocolViolation("attempt is outside the protected task context")
+        decision_ids = attempt.accepted_child_decision_ids
+        if decision_ids is None:
+            raise ProtocolViolation("attempt has no frozen child decision binding")
+        values: list[dict[str, Any]] = []
+        for decision_id in decision_ids:
+            decision = ledger.decisions.get(decision_id)
+            if (
+                decision is None
+                or decision.action.value != "accept"
+                or not decision.attempt_id
+                or not decision.validation_id
+            ):
+                raise ProtocolViolation("accepted child decision binding is invalid")
+            child_attempt = ledger.attempts[decision.attempt_id]
+            child_task = ledger.tasks[child_attempt.task_id]
+            validation = ledger.validations.get(decision.validation_id)
+            child_kind = task_kind(child_task.current_spec.context_refs)
+            if (
+                child_task.parent_task_id != attempt.task_id
+                or child_kind not in RETRIEVAL_KINDS
+                or decision.task_id != child_task.task_id
+                or validation is None
+                or validation.attempt_id != child_attempt.attempt_id
+                or validation.verdict is None
+                or validation.verdict.value != "passed"
+                or child_attempt.snapshot_id != validation.snapshot_id
+            ):
+                raise ProtocolViolation(
+                    "accepted decision is not closed over a passed direct-child snapshot"
+                )
+            if child_attempt.submission is None:
+                raise ProtocolViolation("accepted child attempt has no submission")
+            refs = [
+                *child_attempt.submission.artifact_refs,
+                *child_attempt.submission.evidence_refs,
+            ]
+            for ref in refs:
+                if ref.kind != ArtifactKind.EVIDENCE.value:
+                    raise ProtocolViolation(
+                        "direction child decisions may expose only evidence artifacts"
+                    )
+                version = await self.artifacts.read_by_ref(
+                    ref,
+                    script_build_id=binding.script_build_id,
+                    task_id=child_attempt.task_id,
+                    attempt_id=child_attempt.attempt_id,
+                )
+                values.append(
+                    {
+                        "decision_id": decision_id,
+                        "task_id": child_attempt.task_id,
+                        "attempt_id": child_attempt.attempt_id,
+                        "artifact_ref": _ref_dict(ref),
+                        "artifact": _json_values(asdict(version.artifact)),
+                    }
+                )
+        return values
+
+    async def retrieve(
+        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"):
+            frozen_accounts = {
+                str(snapshot.account.get(key) or "")
+                for key in ("account_name", "resolved_account_name")
+            }
+            if query["account_name"] not in frozen_accounts:
+                raise ProtocolViolation("decode account_name is outside the frozen account scope")
+        adapter = self.retrieval_adapters.get(source_type)
+        if adapter is None:
+            raise ProtocolViolation(f"retrieval adapter is not configured: {source_type}")
+        result = await adapter.retrieve(query=dict(query), snapshot=snapshot)
+        frozen_query = redact(dict(query))
+        if not isinstance(frozen_query, dict):
+            raise ProtocolViolation("redacted retrieval query must remain an object")
+        if result.metadata:
+            safe_metadata = redact(dict(result.metadata))
+            if not isinstance(safe_metadata, dict):
+                raise ProtocolViolation("redacted retrieval metadata must remain an object")
+            frozen_query["_retrieval_metadata"] = safe_metadata
+        record = EvidenceRecordV1(
+            evidence_id=str(uuid4()),
+            source_type=source_type,
+            tool_name=tool_name,
+            query=frozen_query,
+            source_refs=_safe_text_tuple(result.source_refs),
+            raw_artifact_ref=(
+                redact_text(result.raw_artifact_ref) if result.raw_artifact_ref else None
+            ),
+            summary=redact_text(result.summary),
+            supports=_safe_text_tuple(result.supports),
+            confidence=redact_text(result.confidence),
+            limitations=_safe_text_tuple(result.limitations),
+            content_sha256="",
+            created_at=datetime.now(UTC),
+        )
+        version, ref = await self.artifacts.freeze(
+            script_build_id=binding.script_build_id,
+            task_id=_required(context, "task_id"),
+            attempt_id=_required(context, "attempt_id"),
+            spec_version=_required_int(context, "spec_version"),
+            artifact=record,
+        )
+        return {
+            "artifact_ref": _ref_dict(ref),
+            "artifact_version_id": version.artifact_version_id,
+            "evidence": _json_values(asdict(version.artifact)),
+            "metadata": _json_values(redact(dict(result.metadata or {}))),
+        }
+
+    async def load_images(
+        self, urls: Sequence[str], context: Mapping[str, Any]
+    ) -> tuple[dict[str, Any], ...]:
+        if self.image_adapter is None:
+            raise ProtocolViolation("image adapter is not configured")
+        _, snapshot = await self._scope(context)
+        return await self.image_adapter.load(urls=urls, snapshot=snapshot)
+
+    async def save_direction_candidate(
+        self,
+        *,
+        goals: Sequence[Mapping[str, Any]],
+        criteria: Sequence[Mapping[str, Any]],
+        domain_criteria: Sequence[Mapping[str, Any]],
+        topic_refs: Sequence[str],
+        persona_refs: Sequence[str],
+        strategy_refs: Sequence[str],
+        evidence_refs: Sequence[str],
+        legacy_markdown: str,
+        context: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        binding, _ = 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):
+            raise ProtocolViolation(
+                "direction references evidence outside frozen child ACCEPT decisions"
+            )
+        artifact = ScriptDirectionArtifactV1(
+            goals=tuple(
+                DirectionGoal(
+                    goal_id=redact_text(str(item["goal_id"])),
+                    statement=redact_text(str(item["statement"])),
+                    rationale=redact_text(str(item.get("rationale", ""))),
+                )
+                for item in goals
+            ),
+            criteria=tuple(
+                Criterion(
+                    redact_text(str(item["criterion_id"])),
+                    redact_text(str(item["description"])),
+                )
+                for item in criteria
+            ),
+            domain_criteria=tuple(
+                Criterion(
+                    redact_text(str(item["criterion_id"])),
+                    redact_text(str(item["description"])),
+                )
+                for item in domain_criteria
+            ),
+            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),
+            legacy_markdown=redact_text(legacy_markdown),
+        )
+        version, ref = await self.artifacts.freeze(
+            script_build_id=binding.script_build_id,
+            task_id=_required(context, "task_id"),
+            attempt_id=_required(context, "attempt_id"),
+            spec_version=_required_int(context, "spec_version"),
+            artifact=artifact,
+        )
+        return {
+            "artifact_ref": _ref_dict(ref),
+            "artifact_version_id": version.artifact_version_id,
+            "direction": _json_values(version.artifact.content_payload()),
+        }
+
+    async def query_validation_evidence(
+        self,
+        query: str,
+        limit: int,
+        context: Mapping[str, Any],
+    ) -> dict[str, Any]:
+        request = EvidenceQuery(
+            root_trace_id=_required(context, "root_trace_id"),
+            task_id=_required(context, "task_id"),
+            attempt_id=_required(context, "attempt_id"),
+            snapshot_id=_required(context, "snapshot_id"),
+            validation_id=_required(context, "validation_id"),
+            query=query,
+            limit=limit,
+        )
+        response = await self.coordinator.query_evidence(dict(context), request)
+        return cast(dict[str, Any], _json_values(asdict(response)))
+
+    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
+
+        root_trace_id = _required(context, "root_trace_id")
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        task = ledger.tasks[_required(context, "task_id")]
+        attempt = ledger.attempts[_required(context, "attempt_id")]
+        snapshot = await self.coordinator.artifact_store.get(
+            root_trace_id, _required(context, "snapshot_id")
+        )
+        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),
+        )
+        return {"rule_results": [item.to_dict() for item in precheck_snapshot(validation_context)]}
+
+    async def _scope(self, context: Mapping[str, Any]) -> tuple[Any, Any]:
+        root_trace_id = _required(context, "root_trace_id")
+        binding = await self.bindings.get_by_root(root_trace_id)
+        snapshot = await self.snapshots.get(
+            str(binding.input_snapshot_id),
+            script_build_id=binding.script_build_id,
+        )
+        return binding, snapshot
+
+
+def _required(context: Mapping[str, Any], key: str) -> str:
+    value = context.get(key)
+    if not isinstance(value, str) or not value.strip():
+        raise ProtocolViolation(f"protected context is missing {key}")
+    return value
+
+
+def _required_int(context: Mapping[str, Any], key: str) -> int:
+    value = context.get(key)
+    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
+        raise ProtocolViolation(f"protected context is missing {key}")
+    return value
+
+
+def _ref_dict(ref: ArtifactRef) -> dict[str, Any]:
+    return {
+        "uri": ref.uri,
+        "kind": ref.kind,
+        "version": ref.version,
+        "digest": ref.digest,
+        "summary": ref.summary,
+        "metadata": ref.metadata,
+    }
+
+
+def _json_values(value: Any) -> Any:
+    if isinstance(value, datetime):
+        return value.isoformat()
+    if isinstance(value, Mapping):
+        return {str(key): _json_values(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_json_values(item) for item in value]
+    if hasattr(value, "value"):
+        return value.value
+    return value
+
+
+def _safe_text_tuple(values: Sequence[str]) -> tuple[str, ...]:
+    return tuple(redact_text(str(value)) for value in values)
+
+
+__all__ = [
+    "DispatchGuard",
+    "ImageAdapter",
+    "LegacyScriptToolGateway",
+    "RetrievalAdapter",
+    "RetrievalResult",
+]

script_build_host/src/script_build_host/tools/registry.py

@@ -0,0 +1,355 @@
+"""Register the small, capability-labelled script-build tool surface."""
+
+from __future__ import annotations
+
+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 .gateway import LegacyScriptToolGateway
+
+TASK_PRESET_BY_KIND = {
+    "direction": "script_direction_worker",
+    "pattern-retrieval": "script_pattern_retrieval_worker",
+    "decode-retrieval": "script_decode_retrieval_worker",
+    "external-retrieval": "script_external_retrieval_worker",
+    "knowledge-retrieval": "script_knowledge_retrieval_worker",
+}
+
+
+def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGateway) -> None:
+    """Install Host wrappers into the supplied Runner registry.
+
+    Re-registering intentionally replaces a previous Host instance, which keeps
+    isolated tests deterministic while production creates only one composition.
+    """
+
+    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 read_accepted_artifact(context: dict[str, Any] | None = None) -> str:
+        """Read only direct-child artifacts frozen into this attempt."""
+
+        return _json(await gateway.read_accepted_artifacts(context or {}))
+
+    _register(registry, read_accepted_artifact, 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"),
+        )
+        return _json([item.to_dict() for item in results])
+
+    _register(registry, dispatch_script_tasks, capabilities=["task_control"])
+
+    async def query_pattern_qa(message: str, context: dict[str, Any] | None = None) -> str:
+        """Query the configured Pattern service and freeze its evidence."""
+
+        return _json(
+            await gateway.retrieve(
+                "pattern",
+                "query_pattern_qa",
+                {"message": message},
+                context or {},
+            )
+        )
+
+    _register(registry, query_pattern_qa, capabilities=["external_send", "write"])
+
+    async def search_script_decode_case(
+        return_field: str,
+        keyword: str = "",
+        account_name: str | None = None,
+        match_fields: dict[str, str] | None = None,
+        top_k: int = 3,
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Search the frozen script-decode index and freeze ranked evidence."""
+
+        if return_field not in {"主脉络", "元素", "主题", "形式", "作用", "感受"}:
+            raise ValueError("return_field is not supported")
+        if not keyword.strip() and not account_name:
+            raise ValueError("keyword and account_name cannot both be empty")
+        if not 1 <= top_k <= 5:
+            raise ValueError("top_k must be between 1 and 5")
+        valid_columns = {"主题", "形式", "作用", "感受"}
+        valid_subfields = {"维度名", "维度值"}
+        if match_fields and (
+            set(match_fields) - valid_columns or set(match_fields.values()) - valid_subfields
+        ):
+            raise ValueError("match_fields contains an unsupported column or subfield")
+        return _json(
+            await gateway.retrieve(
+                "decode",
+                "search_script_decode_case",
+                {
+                    "return_field": return_field,
+                    "keyword": keyword,
+                    "account_name": account_name,
+                    "match_fields": match_fields or {},
+                    "top_k": top_k,
+                },
+                context or {},
+            )
+        )
+
+    _register(
+        registry,
+        search_script_decode_case,
+        capabilities=["read", "write"],
+        schema=_decode_schema(),
+    )
+
+    async def external_search_case(
+        keyword: str,
+        platform_channel: str = "xhs",
+        max_count: int = 5,
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Search the allowlisted external service without legacy log writes."""
+
+        if platform_channel not in {"xhs", "zhihu"}:
+            raise ValueError("platform_channel must be xhs or zhihu")
+        if not 1 <= max_count <= 20:
+            raise ValueError("max_count must be between 1 and 20")
+        return _json(
+            await gateway.retrieve(
+                "external",
+                "external_search_case",
+                {
+                    "keyword": keyword,
+                    "platform_channel": platform_channel,
+                    "max_count": max_count,
+                },
+                context or {},
+            )
+        )
+
+    _register(
+        registry,
+        external_search_case,
+        capabilities=["external_send", "write"],
+        schema=_external_schema(),
+    )
+
+    async def load_images(image_urls: list[str], context: dict[str, Any] | None = None) -> str:
+        """Load images through the Host outbound and size policy."""
+
+        return _json(await gateway.load_images(image_urls, context or {}))
+
+    _register(registry, load_images, capabilities=["external_send", "read"])
+
+    async def search_knowledge(
+        keyword: str,
+        max_count: int = 3,
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Search the frozen internal knowledge corpus and freeze evidence."""
+
+        if not 1 <= max_count <= 20:
+            raise ValueError("max_count must be between 1 and 20")
+        return _json(
+            await gateway.retrieve(
+                "knowledge",
+                "search_knowledge",
+                {"keyword": keyword, "max_count": max_count},
+                context or {},
+            )
+        )
+
+    _register(
+        registry,
+        search_knowledge,
+        capabilities=["read", "write"],
+        schema=_knowledge_schema(),
+    )
+
+    async def save_direction_candidate(
+        goals: list[dict[str, Any]],
+        criteria: list[dict[str, Any]],
+        legacy_markdown: str,
+        domain_criteria: list[dict[str, Any]] | None = None,
+        topic_refs: list[str] | None = None,
+        persona_refs: list[str] | None = None,
+        strategy_refs: list[str] | None = None,
+        evidence_refs: list[str] | None = None,
+        context: dict[str, Any] | None = None,
+    ) -> str:
+        """Freeze one direction candidate inside the protected Attempt."""
+
+        return _json(
+            await gateway.save_direction_candidate(
+                goals=goals,
+                criteria=criteria,
+                domain_criteria=domain_criteria or [],
+                topic_refs=topic_refs or [],
+                persona_refs=persona_refs or [],
+                strategy_refs=strategy_refs or [],
+                evidence_refs=evidence_refs or [],
+                legacy_markdown=legacy_markdown,
+                context=context or {},
+            )
+        )
+
+    _register(registry, save_direction_candidate, capabilities=["write"])
+
+    async def query_validation_evidence(
+        query: str, limit: int = 10, context: dict[str, Any] | None = None
+    ) -> str:
+        """Query evidence within this Validator's protected frozen scope."""
+
+        return _json(await gateway.query_validation_evidence(query, limit, context or {}))
+
+    _register(registry, query_validation_evidence, capabilities=["read"])
+
+    async def deterministic_precheck(context: dict[str, Any] | None = None) -> str:
+        """Run schema and immutable digest-shape checks on this validation."""
+
+        return _json(await gateway.deterministic_precheck(context or {}))
+
+    _register(registry, deterministic_precheck, capabilities=["read"])
+
+
+def _register(
+    registry: ToolRegistry,
+    func: Any,
+    *,
+    capabilities: list[str],
+    schema: dict[str, Any] | None = None,
+) -> None:
+    registry.register(
+        func,
+        schema=schema,
+        hidden_params=["context"],
+        groups=["script_build"],
+        capabilities=capabilities,
+    )
+
+
+def _decode_schema() -> dict[str, Any]:
+    return {
+        "type": "function",
+        "function": {
+            "name": "search_script_decode_case",
+            "description": "Search frozen script-decode cases.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "return_field": {
+                        "type": "string",
+                        "enum": ["主脉络", "元素", "主题", "形式", "作用", "感受"],
+                    },
+                    "keyword": {"type": "string", "default": ""},
+                    "account_name": {"type": ["string", "null"]},
+                    "match_fields": {
+                        "type": "object",
+                        "properties": {
+                            key: {"type": "string", "enum": ["维度名", "维度值"]}
+                            for key in ("主题", "形式", "作用", "感受")
+                        },
+                        "additionalProperties": False,
+                    },
+                    "top_k": {
+                        "type": "integer",
+                        "default": 3,
+                        "minimum": 1,
+                        "maximum": 5,
+                    },
+                },
+                "required": ["return_field"],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
+def _external_schema() -> dict[str, Any]:
+    return {
+        "type": "function",
+        "function": {
+            "name": "external_search_case",
+            "description": "Search allowlisted external content cases.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "keyword": {"type": "string"},
+                    "platform_channel": {
+                        "type": "string",
+                        "enum": ["xhs", "zhihu"],
+                        "default": "xhs",
+                    },
+                    "max_count": {"type": "integer", "default": 5},
+                },
+                "required": ["keyword"],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
+def _knowledge_schema() -> dict[str, Any]:
+    return {
+        "type": "function",
+        "function": {
+            "name": "search_knowledge",
+            "description": "Search the frozen creation knowledge corpus.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "keyword": {"type": "string"},
+                    "max_count": {"type": "integer", "default": 3},
+                },
+                "required": ["keyword"],
+                "additionalProperties": False,
+            },
+        },
+    }
+
+
+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)
+
+
+__all__ = ["TASK_PRESET_BY_KIND", "register_script_tools"]

script_build_host/tests/test_agent_surface.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from types import SimpleNamespace
+
+import pytest
+from agent import AgentRole, ToolRegistry, get_preset
+
+from script_build_host.agents.model_resolver import (
+    ScriptRoleRunConfigResolver,
+    SnapshotModelManifestSource,
+)
+from script_build_host.agents.presets import register_script_presets
+from script_build_host.domain.input_snapshot import ScriptBuildInputSnapshotV1
+from script_build_host.tools.registry import register_script_tools
+
+
+def test_script_presets_are_exact_and_never_expose_legacy_control_tools() -> None:
+    register_script_presets()
+    expected = {
+        "script_planner",
+        "script_direction_worker",
+        "script_pattern_retrieval_worker",
+        "script_decode_retrieval_worker",
+        "script_external_retrieval_worker",
+        "script_knowledge_retrieval_worker",
+        "script_retrieval_validator",
+        "script_candidate_validator",
+    }
+    forbidden = {
+        "agent",
+        "evaluate",
+        "bash_command",
+        "write_file",
+        "dispatch_tasks",
+        "implement_paths",
+        "begin_round",
+        "multipath",
+    }
+    for name in expected:
+        preset = get_preset(name)
+        assert preset.role in {AgentRole.PLANNER, AgentRole.WORKER, AgentRole.VALIDATOR}
+        assert forbidden.isdisjoint(set(preset.allowed_tools or []))
+        assert set(preset.allowed_tools or []).isdisjoint(set(preset.denied_tools or []))
+
+
+class _Gateway:
+    coordinator = SimpleNamespace(task_store=None)
+
+
+def test_legacy_retrieval_tool_schemas_keep_old_argument_shapes() -> 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()
+    }
+    decode = schemas["search_script_decode_case"]
+    assert decode["required"] == ["return_field"]
+    assert decode["properties"]["match_fields"]["type"] == "object"
+    assert decode["properties"]["top_k"]["default"] == 3
+    external = schemas["external_search_case"]
+    assert external["required"] == ["keyword"]
+    assert external["properties"]["platform_channel"]["default"] == "xhs"
+    knowledge = schemas["search_knowledge"]
+    assert knowledge["required"] == ["keyword"]
+    assert knowledge["properties"]["max_count"]["default"] == 3
+    assert schemas["query_pattern_qa"]["required"] == ["message"]
+    assert schemas["load_images"]["required"] == ["image_urls"]
+
+
+class _Bindings:
+    async def get_by_root(self, root_trace_id: str) -> SimpleNamespace:
+        assert root_trace_id == "root-a"
+        return SimpleNamespace(script_build_id=9, input_snapshot_id=3)
+
+
+class _Snapshots:
+    async def get(self, snapshot_id: str, *, script_build_id: int) -> ScriptBuildInputSnapshotV1:
+        assert (snapshot_id, script_build_id) == ("3", 9)
+        return ScriptBuildInputSnapshotV1(
+            snapshot_id="3",
+            script_build_id=9,
+            execution_id=1,
+            topic_build_id=2,
+            topic_id=3,
+            topic={},
+            account={},
+            persona_points=(),
+            section_patterns=(),
+            strategies=(),
+            prompt_manifest=(),
+            datasource_manifest={},
+            model_manifest={
+                "presets": {
+                    "script_direction_worker": {
+                        "model": "frozen-direction-model",
+                        "temperature": 0.2,
+                        "max_iterations": 18,
+                    }
+                }
+            },
+            canonical_sha256="sha256:" + "a" * 64,
+            created_at=datetime.now(UTC),
+        )
+
+
+@pytest.mark.asyncio
+async def test_role_model_resolver_reads_each_roots_frozen_manifest() -> None:
+    resolver = ScriptRoleRunConfigResolver(
+        {}, manifest_source=SnapshotModelManifestSource(_Bindings(), _Snapshots())
+    )
+    value = await resolver.resolve(
+        role=AgentRole.WORKER,
+        preset="script_direction_worker",
+        context={"root_trace_id": "root-a"},
+    )
+    assert value.model == "frozen-direction-model"
+    assert value.temperature == 0.2
+    assert value.max_iterations == 18