工具协议：以统一搜索和详情入口替换旧工作台读取面

注册 search_mission_context 与 read_mission_context，并由 Gateway 从受保护 Root、Role、Task 和 Attempt 注入身份。删除模型可见的旧快照、前沿、Strategy 和图片读取入口；强化分页工具说明、Validator 必读页检查及 CONTEXT_NOT_EXHAUSTED 失败分类。

script_build_host/src/script_build_host/tools/failures.py

@@ -40,6 +40,7 @@ _RETRY_CODES = frozenset(
         "STALE_BASE_REVISION",
         "STALE_WORKBENCH_STATE",
         "VALIDATION_EVIDENCE_INVALID",
+        "CONTEXT_NOT_EXHAUSTED",
     }
 )
 _ABORT_MARKERS = (

script_build_host/src/script_build_host/tools/gateway.py

@@ -40,8 +40,7 @@ from script_build_host.domain.artifacts import (
     DirectionPreference,
     EvidenceRecordV1,
 )
-from script_build_host.domain.canonical_json import canonical_sha256
-from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation
+from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation, ScriptBuildError
 from script_build_host.domain.ports import (
     InputSnapshotRepository,
     MissionBindingRepository,
@@ -86,10 +85,6 @@ class RetrievalAdapter(Protocol):
     ) -> 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: ...
 
@@ -113,7 +108,6 @@ class LegacyScriptToolGateway:
         artifacts: ScriptBusinessArtifactRepository,
         coordinator: Any,
         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,
@@ -126,7 +120,6 @@ class LegacyScriptToolGateway:
         self.artifacts = artifacts
         self.coordinator = coordinator
         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
@@ -199,25 +192,58 @@ class LegacyScriptToolGateway:
     async def inspect_script_plan(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
         return await self._planner().inspect_script_plan(context=context)
 
-    async def read_workbench_detail(
+    async def search_mission_context(
         self,
         *,
-        view: str,
-        handle: str | None,
+        collection: str,
+        query: str,
+        goal_ids: Sequence[str],
+        task_kinds: Sequence[str],
+        source_types: Sequence[str],
         known_revision: str | None,
-        cursor: int,
+        cursor: str | None,
+        page_size: int,
         context: Mapping[str, Any],
     ) -> Mapping[str, Any]:
         if self.workbench is None:
-            raise ProtocolViolation("Mission Workbench is not configured")
+            raise ProtocolViolation("Context Broker is not configured")
         return cast(
             Mapping[str, Any],
-            await self.workbench.detail(
+            await self.workbench.search_mission_context(
                 root_trace_id=_required(context, "root_trace_id"),
-                view=view,
-                handle=handle,
+                role=str(context.get("role") or "agent"),
+                task_id=cast(str | None, context.get("task_id")),
+                collection=collection,
+                query=query,
+                goal_ids=goal_ids,
+                task_kinds=task_kinds,
+                source_types=source_types,
                 known_revision=known_revision,
                 cursor=cursor,
+                page_size=page_size,
+            ),
+        )
+
+    async def read_mission_context(
+        self,
+        *,
+        handle: str,
+        section: str,
+        cursor: str | None,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        if self.workbench is None:
+            raise ProtocolViolation("Context Broker is not configured")
+        return cast(
+            Mapping[str, Any],
+            await self.workbench.read_mission_context(
+                root_trace_id=_required(context, "root_trace_id"),
+                role=str(context.get("role") or "agent"),
+                task_id=cast(str | None, context.get("task_id")),
+                attempt_id=cast(str | None, context.get("attempt_id")),
+                handle=handle,
+                section=section,
+                cursor=cursor,
             ),
         )
 
@@ -231,35 +257,6 @@ class LegacyScriptToolGateway:
         # immediately before it reserves the Operation.
         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", [])
-        ]
-        view = await self._snapshot_view(context)
-        return cast(dict[str, Any], _json_values(_project_snapshot(payload, view=view)))
-
-    async def _snapshot_view(self, context: Mapping[str, Any]) -> str:
-        task_id = context.get("task_id")
-        if not isinstance(task_id, str) or not task_id.strip():
-            return "planner"
-        ledger = await self.coordinator.task_store.load(_required(context, "root_trace_id"))
-        task = ledger.tasks.get(task_id)
-        if task is None:
-            raise ProtocolViolation("protected Task is not in the mission ledger")
-        return task_kind(task.current_spec.context_refs) or "unknown"
-
     async def read_accepted_portfolio(self, context: Mapping[str, Any]) -> Mapping[str, Any]:
         return await self._root_tools().read_accepted_portfolio(context=context)
 
@@ -289,60 +286,6 @@ class LegacyScriptToolGateway:
             ),
         )
 
-    async def load_frozen_strategy(
-        self,
-        selected_handle: str,
-        context: Mapping[str, Any],
-        *,
-        cursor: int = 0,
-    ) -> dict[str, Any]:
-        """Explicitly load one digest-verified on-demand strategy from the bound snapshot."""
-
-        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 index, item in enumerate(snapshot.strategies):
-            if (
-                strategy_handle(snapshot.snapshot_id, index, item) == selected_handle
-                and item.get("mode") == "on_demand"
-            ):
-                matches.append(item)
-        if len(matches) != 1:
-            raise ValueError("strategy_handle must select one frozen on-demand strategy")
-        item = matches[0]
-        content = item.get("content")
-        digest = item.get("content_sha256")
-        if not isinstance(content, str) or canonical_sha256(content).wire != digest:
-            raise ProtocolViolation("frozen on-demand strategy digest does not match")
-        start = max(cursor, 0)
-        fragment = content[start : start + 7_000]
-        metadata = {
-            key: value
-            for key, value in item.items()
-            if key not in {"content", "content_sha256", "sha256", "source_uri"}
-        }
-        return cast(
-            dict[str, Any],
-            _json_values(
-                {
-                    "strategy_handle": selected_handle,
-                    "metadata": metadata,
-                    "content_fragment": fragment,
-                    "next_cursor": (
-                        start + len(fragment) if start + len(fragment) < len(content) else None
-                    ),
-                }
-            ),
-        )
-
     async def read_accepted_artifacts(self, context: Mapping[str, Any]) -> list[dict[str, Any]]:
         binding, _ = await self._scope(context)
         ledger = await self.coordinator.task_store.load(binding.root_trace_id)
@@ -496,20 +439,19 @@ class LegacyScriptToolGateway:
             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 {}))),
+            "evidence_handle": semantic_handle(
+                "evidence",
+                _required(context, "root_trace_id"),
+                _required(context, "task_id"),
+                ref.digest or ref.version or "",
+            ),
+            "evidence": protect_model_value(
+                _json_values(asdict(version.artifact)),
+                namespace=_required(context, "attempt_id"),
+            ),
+            "source_count": len(result.source_refs),
         }
 
-    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 view_frozen_images(
         self, image_handles: Sequence[str], context: Mapping[str, Any]
     ) -> Sequence[Mapping[str, Any]]:
@@ -518,20 +460,6 @@ class LegacyScriptToolGateway:
             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,
@@ -719,6 +647,17 @@ class LegacyScriptToolGateway:
         if set(result_by_id) != expected_ids:
             raise ProtocolViolation("validation must report every Task criterion exactly once")
         overall = ValidationVerdict(verdict)
+        if overall is ValidationVerdict.PASSED and self.workbench is not None:
+            try:
+                await self.workbench.verify_context_exhausted(
+                    root_trace_id=root_trace_id,
+                    task_id=task_id,
+                    attempt_id=_required(context, "attempt_id"),
+                )
+            except Exception as exc:
+                if getattr(exc, "code", None) != "CONTEXT_NOT_EXHAUSTED":
+                    raise
+                raise ScriptBuildError("CONTEXT_NOT_EXHAUSTED", str(exc)) from exc
         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")
@@ -890,9 +829,16 @@ class LegacyScriptToolGateway:
             artifact=artifact,
         )
         return {
-            "artifact_ref": _ref_dict(ref),
-            "artifact_version_id": version.artifact_version_id,
-            "direction": _json_values(version.artifact.content_payload()),
+            "direction_handle": semantic_handle(
+                "direction",
+                _required(context, "root_trace_id"),
+                _required(context, "task_id"),
+                ref.digest or ref.version or "",
+            ),
+            "direction": protect_model_value(
+                _json_values(version.artifact.content_payload()),
+                namespace=_required(context, "attempt_id"),
+            ),
             "goal_ids_by_client_key": goal_ids_by_client_key,
         }
 
@@ -987,150 +933,6 @@ class LegacyScriptToolGateway:
         return self.root_tools
 
 
-_SNAPSHOT_IDENTITY_FIELDS = (
-    "snapshot_id",
-    "script_build_id",
-    "execution_id",
-    "topic_build_id",
-    "topic_id",
-    "canonical_sha256",
-    "business_input_sha256",
-    "parent_snapshot_id",
-    "parent_snapshot_sha256",
-    "created_at",
-)
-_TOPIC_ITEM_FIELDS = (
-    "point_type",
-    "dimension",
-    "element_name",
-    "note",
-    "category_path",
-    "item_level",
-    "derivation_type",
-    "content",
-    "result",
-    "name",
-)
-
-
-def _project_snapshot(payload: Mapping[str, Any], *, view: str) -> dict[str, Any]:
-    """Project immutable inputs into a role-sized business view."""
-
-    persona_limits = {
-        "planner": 24,
-        "direction": 32,
-        "pattern-retrieval": 16,
-        "structure": 24,
-        "paragraph": 12,
-        "element-set": 12,
-        "root-delivery": 16,
-    }
-    persona = list(payload.get("persona_points") or ())
-    section_patterns = list(payload.get("section_patterns") or ())
-    output = {key: payload.get(key) for key in _SNAPSHOT_IDENTITY_FIELDS if key in payload}
-    output.update(
-        {
-            "view": view,
-            "account": payload.get("account") or {},
-            "topic": _project_topic(payload.get("topic")),
-            "strategies": list(payload.get("strategies") or ()),
-            "persona_points_total": len(persona),
-            "section_patterns_total": len(section_patterns),
-        }
-    )
-    limit = persona_limits.get(view, 16)
-    if limit:
-        output["persona_points"] = _select_persona_points(persona, limit=limit)
-    if view in {
-        "planner",
-        "direction",
-        "pattern-retrieval",
-        "structure",
-        "paragraph",
-        "element-set",
-        "root-delivery",
-    }:
-        output["section_patterns"] = section_patterns
-    return output
-
-
-def _project_topic(raw: Any) -> dict[str, Any]:
-    if not isinstance(raw, Mapping):
-        return {}
-    topic = raw.get("topic")
-    topic_row = dict(topic) if isinstance(topic, Mapping) else {}
-    projected: dict[str, Any] = {
-        "topic": {
-            key: topic_row[key]
-            for key in (
-                "result",
-                "topic_direction",
-                "topic_understanding",
-                "inspiration_evaluation_config",
-            )
-            if key in topic_row
-        }
-    }
-    for collection in ("composition_items", "points", "relations", "sources"):
-        values = raw.get(collection)
-        if not isinstance(values, Sequence) or isinstance(values, (str, bytes)):
-            continue
-        rows = []
-        for value in values:
-            if not isinstance(value, Mapping) or value.get("is_active") is False:
-                continue
-            rows.append({key: value[key] for key in _TOPIC_ITEM_FIELDS if key in value})
-        if rows:
-            projected[collection] = rows
-    return projected
-
-
-def _select_persona_points(values: Sequence[Any], *, limit: int) -> list[dict[str, Any]]:
-    groups: dict[str, list[Mapping[str, Any]]] = {}
-    for value in values:
-        if isinstance(value, Mapping):
-            groups.setdefault(str(value.get("列") or "其他"), []).append(value)
-    for rows in groups.values():
-        rows.sort(
-            key=lambda item: (
-                -_numeric(item.get("人设权重分")),
-                -_numeric(item.get("帖子覆盖率")),
-                str(item.get("点名称") or ""),
-            )
-        )
-    selected: list[dict[str, Any]] = []
-    group_names = sorted(groups)
-    while len(selected) < limit and group_names:
-        remaining = []
-        for name in group_names:
-            rows = groups[name]
-            if rows and len(selected) < limit:
-                selected.append(_project_persona_point(rows.pop(0)))
-            if rows:
-                remaining.append(name)
-        group_names = remaining
-    return selected
-
-
-def _project_persona_point(value: Mapping[str, Any]) -> dict[str, Any]:
-    output = {
-        key: value[key]
-        for key in ("列", "点名称", "点类型", "人设权重分", "帖子覆盖率")
-        if key in value
-    }
-    paths = value.get("分类路径")
-    if isinstance(paths, Sequence) and not isinstance(paths, (str, bytes)):
-        output["分类路径"] = [str(item)[:160] for item in paths[:2]]
-    return output
-
-
-def _numeric(value: Any) -> float:
-    try:
-        return float(value)
-    except (TypeError, ValueError):
-        return 0.0
-
-
 def _required(context: Mapping[str, Any], key: str) -> str:
     value = context.get(key)
     if not isinstance(value, str) or not value.strip():
@@ -1416,7 +1218,6 @@ def _dedupe_refs(values: Any) -> tuple[ArtifactRef, ...]:
 
 __all__ = [
     "DispatchGuard",
-    "ImageAdapter",
     "LegacyScriptToolGateway",
     "RetrievalAdapter",
     "RetrievalResult",

script_build_host/src/script_build_host/tools/registry.py

@@ -12,6 +12,7 @@ from agent import FailureDetail, FailureDisposition, ToolExecutionError, ToolReg
 from agent.tools.models import ToolResult
 
 from script_build_host.domain.errors import ScriptBuildError
+from script_build_host.domain.workbench import protect_model_value
 
 from .failures import classify_script_tool_failure
 from .gateway import LegacyScriptToolGateway
@@ -49,8 +50,15 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         context: dict[str, Any],
     ) -> ToolResult:
         evidence = await gateway.retrieve(source_type, tool_name, query, context)
-        submission = await gateway.submit_current_attempt(context)
-        result_json = _json({"evidence": evidence, "submission": submission})
+        await gateway.submit_current_attempt(context)
+        result_json = _json(
+            {
+                "evidence": protect_model_value(
+                    evidence, namespace=str(context.get("attempt_id") or tool_name)
+                ),
+                "submitted": True,
+            }
+        )
         return ToolResult(
             title="Evidence frozen and attempt submitted",
             output=result_json,
@@ -59,30 +67,6 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             result_summary=result_json,
         )
 
-    async def load_frozen_strategy(
-        strategy_handle: str,
-        cursor: int = 0,
-        context: dict[str, Any] | None = None,
-    ) -> ToolResult:
-        """Load one on-demand strategy selected by its Workbench handle."""
-
-        value = _json(
-            await gateway.load_frozen_strategy(strategy_handle, context or {}, cursor=cursor)
-        )
-        return ToolResult(
-            title="Frozen strategy",
-            output=value,
-            long_term_memory=f"Loaded frozen strategy {strategy_handle}",
-            include_output_only_once=True,
-        )
-
-    _register(
-        registry,
-        load_frozen_strategy,
-        capabilities=["read"],
-        schema=_load_frozen_strategy_schema(),
-    )
-
     async def plan_script_tasks(
         contracts: list[dict[str, Any]],
         parent_task_id: str | None = None,
@@ -167,51 +151,124 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         schema=_decide_script_task_schema(),
     )
 
-    async def read_workbench_detail(
-        view: str,
-        handle: str | None = None,
+    async def search_mission_context(
+        collection: str,
+        query: str = "",
+        goal_ids: list[str] | None = None,
+        task_kinds: list[str] | None = None,
+        source_types: list[str] | None = None,
         known_revision: str | None = None,
-        cursor: int = 0,
+        cursor: str | None = None,
+        page_size: int = 5,
         context: dict[str, Any] | None = None,
     ) -> ToolResult:
-        """Read one bounded, revision-aware Workbench detail page."""
-
-        value = await gateway.read_workbench_detail(
-            view=view,
-            handle=handle,
+        """Search one bounded, revision-aware Mission context collection."""
+
+        value = await gateway.search_mission_context(
+            collection=collection,
+            query=query,
+            goal_ids=goal_ids or [],
+            task_kinds=task_kinds or [],
+            source_types=source_types or [],
             known_revision=known_revision,
             cursor=cursor,
+            page_size=page_size,
             context=context or {},
         )
         output = _json(value)
         return ToolResult(
-            title="Mission Workbench detail",
+            title="Mission context search",
             output=output,
-            long_term_memory=(
-                f"Workbench {view} revision={value.get('state_revision')} "
-                f"items={len(value.get('items', []))}"
-            ),
+            long_term_memory=f"Searched Mission context collection {collection}",
             include_output_only_once=True,
         )
 
     _register(
         registry,
-        read_workbench_detail,
+        search_mission_context,
         capabilities=["read"],
         schema={
             "type": "function",
             "function": {
-                "name": "read_workbench_detail",
-                "description": "Read a bounded page omitted from the automatic Mission context.",
+                "name": "search_mission_context",
+                "description": (
+                    "Search the current role's bounded Mission context. Inspect sufficiency and "
+                    "has_more; when more results are actually needed, continue with the exact "
+                    "returned next_cursor without changing filters."
+                ),
                 "parameters": {
                     "type": "object",
                     "properties": {
-                        "view": {"type": "string", "enum": ["task_tree", "source"]},
-                        "handle": {"type": ["string", "null"]},
+                        "collection": {
+                            "type": "string",
+                            "enum": ["tasks", "accepted_decisions", "inputs", "patterns"],
+                        },
+                        "query": {"type": "string", "default": ""},
+                        "goal_ids": {"type": "array", "items": {"type": "string"}},
+                        "task_kinds": {"type": "array", "items": {"type": "string"}},
+                        "source_types": {"type": "array", "items": {"type": "string"}},
                         "known_revision": {"type": ["string", "null"]},
-                        "cursor": {"type": "integer", "minimum": 0, "default": 0},
+                        "cursor": {"type": ["string", "null"]},
+                        "page_size": {
+                            "type": "integer",
+                            "enum": [5, 8, 10, 20],
+                            "default": 5,
+                        },
+                    },
+                    "required": ["collection"],
+                    "additionalProperties": False,
+                },
+            },
+        },
+    )
+
+    async def read_mission_context(
+        handle: str,
+        section: str = "content",
+        cursor: str | None = None,
+        context: dict[str, Any] | None = None,
+    ) -> ToolResult:
+        """Read a bounded page of one Broker-authorized context source."""
+
+        value = await gateway.read_mission_context(
+            handle=handle,
+            section=section,
+            cursor=cursor,
+            context=context or {},
+        )
+        output = _json(value)
+        return ToolResult(
+            title="Mission context detail",
+            output=output,
+            long_term_memory=f"Read Mission context handle {handle}",
+            include_output_only_once=True,
+        )
+
+    _register(
+        registry,
+        read_mission_context,
+        capabilities=["read"],
+        schema={
+            "type": "function",
+            "function": {
+                "name": "read_mission_context",
+                "description": (
+                    "Read one authorized context handle. Use content for evidence; outline only "
+                    "for navigation. When next_cursor is returned, continue in order until "
+                    "exhausted=true if the source is required."
+                ),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "handle": {"type": "string", "minLength": 1},
+                        "section": {
+                            "type": "string",
+                            "enum": ["content", "outline"],
+                            "default": "content",
+                        },
+                        "cursor": {"type": ["string", "null"]},
                     },
-                    "required": ["view"],
+                    "required": ["handle"],
                     "additionalProperties": False,
                 },
             },
@@ -325,13 +382,6 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
         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", "write"])
-
     async def search_knowledge(
         keyword: str,
         max_count: int = 3,
@@ -391,14 +441,17 @@ def register_script_tools(registry: ToolRegistry, gateway: LegacyScriptToolGatew
             ("client_key", "statement", "rationale"),
             "preferences",
         )
+        value = await gateway.save_direction_candidate(
+            goals=goals,
+            constraints=constraints,
+            preferences=preferences,
+            strategy_handles=strategy_handles or [],
+            evidence_decision_ids=evidence_decision_ids or [],
+            context=context or {},
+        )
         return _json(
-            await gateway.save_direction_candidate(
-                goals=goals,
-                constraints=constraints,
-                preferences=preferences,
-                strategy_handles=strategy_handles or [],
-                evidence_decision_ids=evidence_decision_ids or [],
-                context=context or {},
+            protect_model_value(
+                value, namespace=str((context or {}).get("attempt_id") or "direction")
             )
         )
 
@@ -1024,30 +1077,6 @@ def _save_direction_candidate_schema() -> dict[str, Any]:
     }
 
 
-def _load_frozen_strategy_schema() -> dict[str, Any]:
-    return {
-        "type": "function",
-        "function": {
-            "name": "load_frozen_strategy",
-            "description": (
-                "Load one on-demand strategy using the exact Workbench strategy handle."
-            ),
-            "parameters": {
-                "type": "object",
-                "properties": {
-                    "strategy_handle": {
-                        "type": "string",
-                        "pattern": "^strategy_[0-9a-f]{20}$",
-                    },
-                    "cursor": {"type": "integer", "minimum": 0, "default": 0},
-                },
-                "required": ["strategy_handle"],
-                "additionalProperties": False,
-            },
-        },
-    }
-
-
 def _save_script_paragraphs_schema() -> dict[str, Any]:
     creative_atom = {
         "type": "object",

script_build_host/tests/test_tool_argument_normalization.py

@@ -98,4 +98,7 @@ async def test_retrieval_tool_freezes_and_submits_in_one_terminal_call() -> None
 
     assert result.terminate_run is True
     assert [name for name, _ in gateway.calls] == ["retrieve", "submit"]
-    assert "attempt-1" in result.output
+    assert "attempt-1" not in result.output
+    assert "script-build://" not in result.output
+    assert "artifact_handle" in result.output
+    assert '"submitted": true' in result.output

script_build_host/tests/test_tool_failure_bridge.py

@@ -124,13 +124,24 @@ def test_paragraph_batch_tool_exposes_exact_nested_contract():
     assert item["additionalProperties"] is False
 
 
-def test_strategy_tool_accepts_only_workbench_handles():
+def test_context_read_tool_accepts_only_broker_handles_and_opaque_cursor():
     registry = ToolRegistry()
     register_script_tools(registry, object())  # type: ignore[arg-type]
-    schema = registry._tools["load_frozen_strategy"]["schema"]
-    strategy_ref = schema["function"]["parameters"]["properties"]["strategy_handle"]
+    schema = registry._tools["read_mission_context"]["schema"]
+    properties = schema["function"]["parameters"]["properties"]
 
-    assert strategy_ref["pattern"] == "^strategy_[0-9a-f]{20}$"
+    assert schema["function"]["parameters"]["required"] == ["handle"]
+    assert properties["cursor"]["type"] == ["string", "null"]
+
+
+def test_validator_unread_required_context_is_retryable_without_replanning():
+    detail = classify_script_tool_failure(
+        ScriptBuildError("CONTEXT_NOT_EXHAUSTED", "read remaining pages"),
+        source_tool="submit_validation",
+    )
+
+    assert detail.code == "CONTEXT_NOT_EXHAUSTED"
+    assert detail.disposition is FailureDisposition.RETRY_CALL
 
 
 def test_element_tool_exposes_only_semantic_targets_and_dimension_contracts():