高维治理：统一任务契约与检索恢复

agent/agent/orchestration/coordinator.py

@@ -376,7 +376,7 @@ class TaskCoordinator:
             if item.failure is not None
         ]
         last_failure = failures[-1].fingerprint() if failures else None
-        return {
+        snapshot = {
             "root": {
                 "spec": spec_value(root),
                 "status": root.status.value,
@@ -394,6 +394,10 @@ class TaskCoordinator:
             ),
             "last_failure": last_failure,
         }
+        observer = getattr(self.task_context_provider, "progress_snapshot", None)
+        if callable(observer):
+            snapshot["observation_progress"] = await observer(root_trace_id)
+        return snapshot
 
     async def _mutate(
         self,

script_build_host/migrations/CHECKSUMS.sha256

@@ -1,3 +1,4 @@
 0d524c7647d567ec3fba3c6d747ec09c2c96c9805506857adf14996a2539bff7  migrations/versions/0001_phase_one_business_tables.py
 c570caabd328ca4ff0c7aaa6212f44af3bae916bcdb5f2b028e40a1f16da2ff7  migrations/versions/0002_phase_two_artifact_types.py
 1e4f94fd826ca8dfe8b0b0c1dd33470f052d371221d9d47c7d4f5638cc741f80  migrations/versions/0003_phase_three_publication_fencing.py
+ee3b6c2af035616b17fe8e0c2430f50be0a6de31fed23b076837120d07ddaa0e  migrations/versions/0004_candidate_source_identity.py

script_build_host/migrations/versions/0004_candidate_source_identity.py

@@ -0,0 +1,64 @@
+"""Persist source-qualified logical identities for candidate workspace seeds.
+
+Revision ID: 0004_candidate_source_identity
+Revises: 0003_phase_three_publication
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import mysql
+
+revision: str = "0004_candidate_source_identity"
+down_revision: str | None = "0003_phase_three_publication"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    timestamp = sa.DateTime(timezone=True).with_variant(mysql.DATETIME(fsp=6), "mysql")
+    op.create_table(
+        "script_build_candidate_source_identity",
+        sa.Column("id", sa.BigInteger().with_variant(sa.Integer, "sqlite"), primary_key=True),
+        sa.Column("workspace_artifact_version_id", sa.BigInteger(), nullable=False),
+        sa.Column("entity_kind", sa.String(32), nullable=False),
+        sa.Column("target_local_id", sa.BigInteger(), nullable=False),
+        sa.Column("source_artifact_version_id", sa.BigInteger(), nullable=False),
+        sa.Column("source_artifact_digest", sa.String(64), nullable=False),
+        sa.Column("source_local_id", sa.BigInteger(), nullable=False),
+        sa.Column("created_at", timestamp, nullable=False),
+        sa.UniqueConstraint(
+            "workspace_artifact_version_id",
+            "entity_kind",
+            "target_local_id",
+            name="uq_candidate_source_target",
+        ),
+        sa.UniqueConstraint(
+            "workspace_artifact_version_id",
+            "source_artifact_version_id",
+            "entity_kind",
+            "source_local_id",
+            name="uq_candidate_source_origin",
+        ),
+        sa.CheckConstraint(
+            "entity_kind IN ('paragraph', 'element')", name="ck_candidate_source_kind"
+        ),
+        sa.CheckConstraint(
+            "length(source_artifact_digest) = 64", name="ck_candidate_source_digest"
+        ),
+    )
+    op.create_index(
+        "ix_candidate_source_workspace",
+        "script_build_candidate_source_identity",
+        ["workspace_artifact_version_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_candidate_source_workspace", table_name="script_build_candidate_source_identity"
+    )
+    op.drop_table("script_build_candidate_source_identity")

script_build_host/src/script_build_host/adapters/legacy_retrieval.py

@@ -357,6 +357,7 @@ class LegacyExternalRetrievalAdapter:
         http: SafeHttpClient,
         openrouter: LegacyOpenRouterClient | None = None,
         image_model: str = "google/gemini-3-flash-preview",
+        detail_concurrency: int = 5,
     ) -> None:
         self.xhs_search_endpoint = xhs_search_endpoint
         self.xhs_detail_endpoint = xhs_detail_endpoint
@@ -364,17 +365,18 @@ class LegacyExternalRetrievalAdapter:
         self.http = http
         self.openrouter = openrouter
         self.image_model = image_model
+        self._detail_semaphore = asyncio.Semaphore(max(1, detail_concurrency))
 
     async def retrieve(self, *, query: Mapping[str, Any], snapshot: Any) -> RetrievalResult:
         del snapshot
         keyword = str(query.get("keyword") or "")
         max_count = int(query.get("max_count", 5))
         platform = query.get("platform_channel", "xhs")
-        rows = (
-            await self._search_xhs(keyword, max_count)
-            if platform == "xhs"
-            else await self._search_zhihu(keyword, max_count)
-        )
+        if platform == "xhs":
+            rows, detail_failures = await self._search_xhs(keyword, max_count)
+        else:
+            rows = await self._search_zhihu(keyword, max_count)
+            detail_failures = 0
         await asyncio.gather(*(self._add_image_understanding(row) for row in rows))
         safe_rows = redact(rows)
         if not isinstance(safe_rows, list):
@@ -388,13 +390,32 @@ class LegacyExternalRetrievalAdapter:
             summary=json.dumps(safe_rows, ensure_ascii=False, sort_keys=True),
             supports=refs,
             confidence="upstream",
-            limitations=("image understanding unavailable",) if self.openrouter is None else (),
-            metadata={"response_sha256": canonical_sha256(rows).wire},
+            limitations=tuple(
+                item
+                for item in (
+                    "image understanding unavailable" if self.openrouter is None else None,
+                    (
+                        f"{detail_failures} upstream detail request(s) failed; "
+                        "search cards retained"
+                        if detail_failures
+                        else None
+                    ),
+                )
+                if item is not None
+            ),
+            metadata={
+                "response_sha256": canonical_sha256(rows).wire,
+                "detail_failures": detail_failures,
+                "partial_result": detail_failures > 0,
+            },
         )
 
-    async def _search_xhs(self, keyword: str, max_count: int) -> list[dict[str, Any]]:
+    async def _search_xhs(
+        self, keyword: str, max_count: int
+    ) -> tuple[list[dict[str, Any]], int]:
         cursor: str | int = ""
         output: list[dict[str, Any]] = []
+        detail_failures = 0
         for _ in range(2):
             body = await self.http.json(
                 "POST",
@@ -406,41 +427,53 @@ class LegacyExternalRetrievalAdapter:
                     "publish_time": "",
                     "cursor": cursor,
                 },
+                retryable=True,
+                provider="xhs",
+                operation="search",
             )
             if not isinstance(body, dict) or body.get("code") not in (0, "0"):
                 raise ProtocolViolation("XHS search returned a non-zero code")
             block = body.get("data") or {}
             raw_items = block.get("data") if isinstance(block, dict) else []
             items = [item for item in raw_items or [] if isinstance(item, dict)]
+            remaining = max_count - len(output)
+            items = items[:remaining]
             ids = [str(item["id"]).strip() for item in items if str(item.get("id") or "").strip()]
             detail_values = await asyncio.gather(
-                *(self._xhs_detail(identifier) for identifier in ids)
+                *(self._xhs_detail(identifier) for identifier in ids),
+                return_exceptions=True,
             )
             detail_map = {
                 identifier: value
                 for identifier, value in zip(ids, detail_values, strict=True)
-                if value is not None
+                if isinstance(value, dict)
             }
+            detail_failures += sum(
+                isinstance(value, BaseException) or value is None for value in detail_values
+            )
             for item in items:
                 identifier = str(item.get("id") or "").strip()
                 if identifier and identifier in detail_map:
                     output.append(detail_map[identifier])
-                elif not identifier:
+                else:
                     output.append(_xhs_card(item))
                 if len(output) >= max_count:
-                    return output[:max_count]
+                    return output[:max_count], detail_failures
             if not block.get("has_more") or block.get("next_cursor") in (None, ""):
                 break
             cursor = block["next_cursor"]
-        return output[:max_count]
+        return output[:max_count], detail_failures
 
     async def _xhs_detail(self, identifier: str) -> dict[str, Any] | None:
-        try:
+        async with self._detail_semaphore:
             body = await self.http.json(
-                "POST", self.xhs_detail_endpoint, json_body={"content_id": identifier}
+                "POST",
+                self.xhs_detail_endpoint,
+                json_body={"content_id": identifier},
+                retryable=True,
+                provider="xhs",
+                operation="detail",
             )
-        except ProtocolViolation:
-            return None
         inner = (
             body.get("data") if isinstance(body, dict) and body.get("code") in (0, "0") else None
         )

script_build_host/src/script_build_host/adapters/retrieval.py

@@ -15,7 +15,7 @@ from urllib.parse import urljoin
 
 import httpx
 
-from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.errors import ProtocolViolation, UpstreamFailure
 from script_build_host.infrastructure.canonical_json import canonical_json_bytes, canonical_sha256
 from script_build_host.infrastructure.outbound import OutboundPolicy
 from script_build_host.infrastructure.redaction import redact, redact_text
@@ -33,12 +33,16 @@ class SafeHttpClient:
         timeout_seconds: float = 30.0,
         max_response_bytes: int = 5_000_000,
         max_redirects: int = 3,
+        max_attempts: int = 3,
+        retry_backoff_seconds: float = 0.05,
     ) -> None:
         self.client = client
         self.policy = policy
         self.timeout_seconds = timeout_seconds
         self.max_response_bytes = max_response_bytes
         self.max_redirects = max_redirects
+        self.max_attempts = max_attempts
+        self.retry_backoff_seconds = retry_backoff_seconds
 
     async def request(
         self,
@@ -47,6 +51,42 @@ class SafeHttpClient:
         *,
         json_body: Mapping[str, Any] | None = None,
         headers: Mapping[str, str] | None = None,
+        retryable: bool = False,
+        provider: str = "http",
+        operation: str = "request",
+    ) -> httpx.Response:
+        attempts = self.max_attempts if retryable else 1
+        for attempt in range(attempts):
+            try:
+                return await self._request_attempt(
+                    method,
+                    url,
+                    json_body=json_body,
+                    headers=headers,
+                    provider=provider,
+                    operation=operation,
+                )
+            except UpstreamFailure as exc:
+                if not exc.retryable or attempt + 1 >= attempts:
+                    raise
+                delay = (
+                    exc.retry_after_seconds
+                    if exc.retry_after_seconds is not None
+                    else self.retry_backoff_seconds * (2**attempt)
+                )
+                if delay > 0:
+                    await asyncio.sleep(min(delay, 2.0))
+        raise AssertionError("unreachable retry loop")
+
+    async def _request_attempt(
+        self,
+        method: str,
+        url: str,
+        *,
+        json_body: Mapping[str, Any] | None,
+        headers: Mapping[str, str] | None,
+        provider: str,
+        operation: str,
     ) -> httpx.Response:
         current = await self.policy.validate_url(url)
         request_headers = {"Accept-Encoding": "identity", **dict(headers or {})}
@@ -58,7 +98,24 @@ class SafeHttpClient:
                 headers=request_headers,
                 timeout=self.timeout_seconds,
             )
-            response = await self.client.send(request, stream=True, follow_redirects=False)
+            try:
+                response = await self.client.send(request, stream=True, follow_redirects=False)
+            except httpx.TimeoutException as exc:
+                raise UpstreamFailure(
+                    "TRANSIENT_TIMEOUT",
+                    "upstream request timed out",
+                    provider=provider,
+                    operation=operation,
+                    retryable=True,
+                ) from exc
+            except httpx.TransportError as exc:
+                raise UpstreamFailure(
+                    "TRANSIENT_CONNECTION_LOST",
+                    "upstream connection was interrupted",
+                    provider=provider,
+                    operation=operation,
+                    retryable=True,
+                ) from exc
             if 300 <= response.status_code < 400:
                 location = response.headers.get("location")
                 await response.aclose()
@@ -69,14 +126,67 @@ class SafeHttpClient:
                 json_body = None if method == "GET" else json_body
                 continue
             content = bytearray()
-            async for chunk in response.aiter_bytes():
-                content.extend(chunk)
-                if len(content) > self.max_response_bytes:
-                    await response.aclose()
-                    raise ProtocolViolation("upstream response exceeds the configured byte limit")
+            try:
+                async for chunk in response.aiter_bytes():
+                    content.extend(chunk)
+                    if len(content) > self.max_response_bytes:
+                        await response.aclose()
+                        raise ProtocolViolation(
+                            "upstream response exceeds the configured byte limit"
+                        )
+            except httpx.TimeoutException as exc:
+                await response.aclose()
+                raise UpstreamFailure(
+                    "TRANSIENT_TIMEOUT",
+                    "upstream response timed out",
+                    provider=provider,
+                    operation=operation,
+                    retryable=True,
+                ) from exc
+            except httpx.TransportError as exc:
+                await response.aclose()
+                raise UpstreamFailure(
+                    "TRANSIENT_CONNECTION_LOST",
+                    "upstream response was interrupted",
+                    provider=provider,
+                    operation=operation,
+                    retryable=True,
+                ) from exc
             await response.aclose()
             if response.is_error:
-                raise ProtocolViolation(f"upstream returned HTTP {response.status_code}")
+                retry_after = _retry_after_seconds(response.headers.get("retry-after"))
+                if response.status_code in {401, 403}:
+                    raise UpstreamFailure(
+                        "UPSTREAM_AUTH_REJECTED",
+                        f"upstream rejected authentication with HTTP {response.status_code}",
+                        provider=provider,
+                        operation=operation,
+                        retryable=False,
+                    )
+                if response.status_code == 429:
+                    raise UpstreamFailure(
+                        "RATE_LIMITED",
+                        "upstream rate limit was reached",
+                        provider=provider,
+                        operation=operation,
+                        retryable=True,
+                        retry_after_seconds=retry_after,
+                    )
+                if response.status_code >= 500:
+                    raise UpstreamFailure(
+                        "UPSTREAM_UNAVAILABLE",
+                        f"upstream returned HTTP {response.status_code}",
+                        provider=provider,
+                        operation=operation,
+                        retryable=True,
+                    )
+                raise UpstreamFailure(
+                    "UPSTREAM_CONTRACT_MISMATCH",
+                    f"upstream returned HTTP {response.status_code}",
+                    provider=provider,
+                    operation=operation,
+                    retryable=False,
+                )
             return httpx.Response(
                 status_code=response.status_code,
                 headers=response.headers,
@@ -92,12 +202,38 @@ class SafeHttpClient:
         *,
         json_body: Mapping[str, Any] | None = None,
         headers: Mapping[str, str] | None = None,
+        retryable: bool = False,
+        provider: str = "http",
+        operation: str = "json",
     ) -> Any:
-        response = await self.request(method, url, json_body=json_body, headers=headers)
+        response = await self.request(
+            method,
+            url,
+            json_body=json_body,
+            headers=headers,
+            retryable=retryable,
+            provider=provider,
+            operation=operation,
+        )
         try:
             return response.json()
         except ValueError as exc:
-            raise ProtocolViolation("upstream response is not valid JSON") from exc
+            raise UpstreamFailure(
+                "UPSTREAM_CONTRACT_MISMATCH",
+                "upstream response is not valid JSON",
+                provider=provider,
+                operation=operation,
+                retryable=False,
+            ) from exc
+
+
+def _retry_after_seconds(value: str | None) -> float | None:
+    if value is None:
+        return None
+    try:
+        return max(0.0, min(float(value), 60.0))
+    except ValueError:
+        return None
 
 
 class PatternRetrievalAdapter:

script_build_host/src/script_build_host/agents/model_manifest.py

@@ -9,6 +9,7 @@ from script_build_host.agents.prompts import (
     phase_three_prompt_manifest,
     script_build_prompt_manifest,
 )
+from script_build_host.domain.model_contracts import validate_model_prompt_closure
 
 _VALIDATOR_CAPS = {
     "script_retrieval_validator": 20,
@@ -16,6 +17,18 @@ _VALIDATOR_CAPS = {
     "script_root_validator": 25,
 }
 
+_LEGACY_MODEL_KEYS = {
+    "script_pattern_retrieval_worker": "model_name_means_pattern_relation",
+    "script_decode_retrieval_worker": "model_name_means_case",
+    "script_external_retrieval_worker": "model_name_means_data",
+    "script_knowledge_retrieval_worker": "model_name_means_knowledge",
+    "script_retrieval_validator": "model_name_script_evaluator",
+    "script_candidate_validator": "model_name_script_evaluator",
+    "script_candidate_compare_worker": "model_name_script_evaluator",
+    "script_candidate_portfolio_worker": "model_name_script_evaluator",
+    "script_root_validator": "model_name_script_evaluator",
+}
+
 
 def resolve_script_role_model_manifest(
     configured: Mapping[str, object],
@@ -23,7 +36,7 @@ def resolve_script_role_model_manifest(
     model: str | None,
     max_iterations: int,
 ) -> dict[str, object]:
-    """Merge explicit presets over complete defaults for all sixteen roles."""
+    """Merge explicit presets over complete defaults for every frozen role."""
 
     output = dict(configured)
     raw_presets = output.get("presets")
@@ -60,4 +73,70 @@ def resolve_script_role_model_manifest(
     return output
 
 
-__all__ = ["resolve_script_role_model_manifest"]
+def resolve_request_role_model_manifest(
+    raw_config: Mapping[str, object],
+    defaults: Mapping[str, object],
+) -> dict[str, object]:
+    """Resolve API legacy model keys through the same complete role catalog."""
+
+    default_presets = defaults.get("presets", defaults)
+    if not isinstance(default_presets, Mapping):
+        default_presets = {}
+    planner_default = default_presets.get("script_planner", {})
+    if not isinstance(planner_default, Mapping):
+        planner_default = {}
+    model = str(raw_config.get("model_name", planner_default.get("model", "gpt-4o")))
+    max_iterations = _as_int(raw_config.get("max_iterations", 120), "max_iterations")
+    temperature = _as_float(raw_config.get("temperature", 0.3), "temperature")
+    explicit: dict[str, object] = {}
+    for item in (*script_build_prompt_manifest(), *phase_three_prompt_manifest()):
+        preset = str(item["preset"])
+        role = str(item["role"])
+        preset_default = default_presets.get(preset, {})
+        if not isinstance(preset_default, Mapping):
+            preset_default = {}
+        model_key = _LEGACY_MODEL_KEYS.get(preset, "model_name")
+        explicit[preset] = {
+            "model": str(raw_config.get(model_key, preset_default.get("model", model))),
+            "temperature": _as_float(
+                raw_config.get(
+                    "temperature",
+                    preset_default.get("temperature", 0.0 if role == "validator" else temperature),
+                ),
+                "temperature",
+            ),
+            "max_iterations": _as_int(
+                raw_config.get(
+                    "max_iterations", preset_default.get("max_iterations", max_iterations)
+                ),
+                "max_iterations",
+            ),
+        }
+    return resolve_script_role_model_manifest(
+        {"presets": explicit}, model=model, max_iterations=max_iterations
+    )
+
+
+def _as_int(value: object, label: str) -> int:
+    if isinstance(value, bool) or not isinstance(value, (int, float, str)):
+        raise ValueError(f"{label} must be numeric")
+    try:
+        return int(value)
+    except ValueError as exc:
+        raise ValueError(f"{label} must be numeric") from exc
+
+
+def _as_float(value: object, label: str) -> float:
+    if isinstance(value, bool) or not isinstance(value, (int, float, str)):
+        raise ValueError(f"{label} must be numeric")
+    try:
+        return float(value)
+    except ValueError as exc:
+        raise ValueError(f"{label} must be numeric") from exc
+
+
+__all__ = [
+    "resolve_request_role_model_manifest",
+    "resolve_script_role_model_manifest",
+    "validate_model_prompt_closure",
+]

script_build_host/src/script_build_host/agents/presets.py

@@ -5,6 +5,9 @@ from __future__ import annotations
 from agent import AgentPreset, AgentRole
 from agent.core.presets import register_preset
 
+from script_build_host.domain.task_capabilities import task_capability
+from script_build_host.domain.task_contracts import ScriptTaskKind
+
 from .prompts import (
     COMPARISON_WORKER_PROMPT,
     COMPOSE_WORKER_PROMPT,
@@ -63,6 +66,14 @@ def _worker(tools: list[str], *, prompt: str) -> AgentPreset:
     )
 
 
+def _task_worker(kind: ScriptTaskKind, *, prompt: str) -> AgentPreset:
+    capability = task_capability(kind)
+    return _worker(
+        sorted(capability.allowed_tools - {"submit_attempt"}),
+        prompt=prompt,
+    )
+
+
 def register_script_presets() -> None:
     """Register only the new Agentic script-build roles.
 
@@ -91,98 +102,85 @@ def register_script_presets() -> None:
     )
     register_preset(
         "script_pattern_retrieval_worker",
-        _worker(
-            ["search_mission_context", "read_mission_context", "query_pattern_qa"],
+        _task_worker(
+            ScriptTaskKind.PATTERN_RETRIEVAL,
             prompt=PATTERN_RETRIEVAL_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_decode_retrieval_worker",
-        _worker(
-            ["search_mission_context", "read_mission_context", "search_script_decode_case"],
+        _task_worker(
+            ScriptTaskKind.DECODE_RETRIEVAL,
             prompt=DECODE_RETRIEVAL_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_external_retrieval_worker",
-        _worker(
-            ["search_mission_context", "read_mission_context", "external_search_case"],
+        _task_worker(
+            ScriptTaskKind.EXTERNAL_RETRIEVAL,
             prompt=EXTERNAL_RETRIEVAL_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_knowledge_retrieval_worker",
-        _worker(
-            ["search_mission_context", "read_mission_context", "search_knowledge"],
+        _task_worker(
+            ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
             prompt=KNOWLEDGE_RETRIEVAL_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_direction_worker",
-        _worker(
-            ["search_mission_context", "read_mission_context", "save_direction_candidate"],
+        _task_worker(
+            ScriptTaskKind.DIRECTION,
             prompt=DIRECTION_WORKER_PROMPT,
         ),
     )
-    structure_writes = ["save_script_paragraphs"]
-    paragraph_writes = ["save_script_paragraphs"]
-    element_writes = ["save_script_elements"]
-    common_candidate_reads = [
-        "search_mission_context",
-        "read_mission_context",
-        "view_frozen_images",
-    ]
     register_preset(
         "script_structure_worker",
-        _worker(
-            [*common_candidate_reads, *structure_writes],
+        _task_worker(
+            ScriptTaskKind.STRUCTURE,
             prompt=STRUCTURE_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_paragraph_worker",
-        _worker(
-            [*common_candidate_reads, *paragraph_writes],
+        _task_worker(
+            ScriptTaskKind.PARAGRAPH,
             prompt=PARAGRAPH_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_element_set_worker",
-        _worker(
-            [*common_candidate_reads, *element_writes],
+        _task_worker(
+            ScriptTaskKind.ELEMENT_SET,
             prompt=ELEMENT_SET_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_candidate_compare_worker",
-        _worker(
-            ["search_mission_context", "read_mission_context", "save_comparison_candidate"],
+        _task_worker(
+            ScriptTaskKind.COMPARE,
             prompt=COMPARISON_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_compose_worker",
-        _worker(
-            [],
+        _task_worker(
+            ScriptTaskKind.COMPOSE,
             prompt=COMPOSE_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_candidate_portfolio_worker",
-        _worker(
-            [],
+        _task_worker(
+            ScriptTaskKind.CANDIDATE_PORTFOLIO,
             prompt=PORTFOLIO_WORKER_PROMPT,
         ),
     )
     register_preset(
         "script_root_worker",
-        _worker(
-            [
-                "search_mission_context",
-                "read_mission_context",
-                "legacy_projection_dry_run",
-                "save_root_delivery_manifest",
-            ],
+        _task_worker(
+            ScriptTaskKind.ROOT_DELIVERY,
             prompt=ROOT_WORKER_PROMPT,
         ),
     )

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

@@ -9,8 +9,11 @@
   `dispatch_script_tasks`；每次决策前读取对应 Validation。
 - PlannerTaskInput 只提交 `task_kind`、语义 `scope_selector`、`intent_class`、`objective`、
   `goal_ids`、`criteria` 和 `input_decision_ids`，真实需要时才提交 `base_decision_id`、
-  `comparison_decision_ids`、`supersedes_decision_ids` 或 `gap_key`。`criteria` 只用
-  `client_key`、`description`、`hard` 补充业务标准，通用硬标准由 Host 注入。
+  `comparison_decision_ids`、`supersedes_decision_ids` 或 `gap_key`。`criteria` 使用
+  `client_key`、`criterion_type`、`params`、`hard`；只有非 Retrieval 的 `custom` 标准才写
+  `description`。Retrieval 只允许 `evidence_nonempty`、`source_attributable`、
+  `semantic_relevance`、`concept_coverage`、`freshness`，不能要求映射、人设分析、写作或
+  结构化合成文档。通用硬标准由 Host 注入。
 - 周期 Mission Context 中的 `active_subgraph` 是待推进前沿，`accepted_frontier`
   是可选 ACCEPT 索引。Decision ID 只能来自该索引。不要提交物理引用、
   内容摘要值、producer/write scope、schema、budget、candidate closure 或 compose order；
@@ -21,6 +24,9 @@
 - Phase 1 使用空 `goal_ids`。Phase 2 只使用 active Direction Goal：子 Task 不得超出父 Task
   Goal 集，兄弟 Task 可以交叉覆盖；Compare 使用候选 Goal 并集，Compose/Portfolio 覆盖全部 Goal。
 - 先创建父 Task 再创建子 Task。阶段层级、Root 所有权和 BLOCK reason 只服从受保护阶段策略。
+- `phase_protocol` 是当前阶段唯一的机器可读协议。只从其中的 `allowed_child_kinds`、
+  `valid_actions` 和 `selectable_decisions` 选择动作与 Decision，不从全局 accepted 列表猜测
+  当前 Compose/Portfolio 的采用闭包。
 - `scope_selector` 只写 `{anchor: mission|parent, path: [语义段...]}`；Host 负责拼接、
   包含关系和写范围，不得构造 URI。
 - Phase 2 初始创作 Task 不要把 Phase 1 Retrieval Decision 放入 `input_decision_ids`；active

script_build_host/src/script_build_host/agents/validation.py

@@ -30,6 +30,8 @@ from script_build_host.domain.ports import (
     MissionBindingRepository,
     ScriptBusinessArtifactRepository,
 )
+from script_build_host.domain.task_capabilities import task_capability
+from script_build_host.domain.task_contracts import ScriptTaskKind
 
 TASK_KIND_PREFIX = "script-build://task-kinds/"
 RETRIEVAL_KINDS = frozenset(
@@ -80,14 +82,10 @@ class ScriptBuildValidationPolicy:
 
     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 == ROOT_DELIVERY_KIND:
-            preset = "script_root_validator"
-        elif kind == "direction" or kind in PHASE_TWO_KINDS:
-            preset = "script_candidate_validator"
-        else:
-            raise ValueError(f"Unsupported script-build task kind: {kind!r}")
+        try:
+            preset = task_capability(ScriptTaskKind(str(kind))).validator_preset
+        except ValueError as exc:
+            raise ValueError(f"Unsupported script-build task kind: {kind!r}") from exc
         preflight_rules = [
             "immutable-reference-present",
             "sha256-digest-shape",

script_build_host/src/script_build_host/api/routes.py

@@ -12,6 +12,7 @@ from uuid import uuid4
 from fastapi import APIRouter, Depends, Header, Query, Request, WebSocket, WebSocketDisconnect
 from fastapi.responses import HTMLResponse
 
+from script_build_host.agents.model_manifest import resolve_request_role_model_manifest
 from script_build_host.agents.prompt_catalog import script_prompt_requests
 from script_build_host.application.mission_service import (
     ScriptMissionService,
@@ -771,9 +772,7 @@ def create_script_build_router(
         await _owned_trace(trace_store, binding.root_trace_id, trace_id)
         messages = await trace_store.get_trace_messages(trace_id)
         return {
-            "messages": [
-                item.to_dict() for item in messages if int(item.sequence) > after_sequence
-            ]
+            "messages": [item.to_dict() for item in messages if int(item.sequence) > after_sequence]
         }
 
     @router.get("/api/pattern/script_builds/{script_build_id}/trace_messages")
@@ -895,66 +894,11 @@ async def _start_command(
         raw_config.get("max_iterations"), bool
     ):
         raise problem(400, "INVALID_AGENT_CONFIG", "model run configuration is invalid")
-    default_presets = (default_model_manifest or {}).get(
-        "presets",
-        default_model_manifest or {},
-    )
-    planner_default = (
-        default_presets.get("script_planner", {}) if isinstance(default_presets, Mapping) else {}
-    )
-    if not isinstance(planner_default, Mapping):
-        planner_default = {}
     try:
-        model = str(raw_config.get("model_name", planner_default.get("model", "gpt-4o")))
-        temperature = float(raw_config.get("temperature", 0.3))
-        max_iterations = int(raw_config.get("max_iterations", 120))
-        presets: dict[str, dict[str, object]] = {}
-        preset_keys = {
-            "script_planner": "model_name",
-            "script_direction_worker": "model_name",
-            "script_pattern_retrieval_worker": "model_name_means_pattern_relation",
-            "script_decode_retrieval_worker": "model_name_means_case",
-            "script_external_retrieval_worker": "model_name_means_data",
-            "script_knowledge_retrieval_worker": "model_name_means_knowledge",
-            "script_retrieval_validator": "model_name_script_evaluator",
-            "script_candidate_validator": "model_name_script_evaluator",
-            "script_structure_worker": "model_name",
-            "script_paragraph_worker": "model_name",
-            "script_element_set_worker": "model_name",
-            "script_candidate_compare_worker": "model_name_script_evaluator",
-            "script_compose_worker": "model_name",
-            "script_candidate_portfolio_worker": "model_name_script_evaluator",
-        }
-        for name, key in preset_keys.items():
-            preset_default = (
-                default_presets.get(name, {}) if isinstance(default_presets, Mapping) else {}
-            )
-            if not isinstance(preset_default, Mapping):
-                preset_default = {}
-            presets[name] = {
-                "model": str(raw_config.get(key, preset_default.get("model", model))),
-                "temperature": float(
-                    cast(
-                        Any,
-                        raw_config.get(
-                            "temperature",
-                            preset_default.get(
-                                "temperature",
-                                0.0 if "validator" in name else temperature,
-                            ),
-                        ),
-                    )
-                ),
-                "max_iterations": int(
-                    cast(
-                        Any,
-                        raw_config.get(
-                            "max_iterations",
-                            preset_default.get("max_iterations", max_iterations),
-                        ),
-                    )
-                ),
-            }
+        resolved_manifest = resolve_request_role_model_manifest(
+            raw_config, default_model_manifest or {}
+        )
+        presets = cast(dict[str, dict[str, object]], resolved_manifest["presets"])
     except (TypeError, ValueError, OverflowError) as exc:
         raise problem(400, "INVALID_AGENT_CONFIG", "model run configuration is invalid") from exc
     model_pattern = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,199}$")
@@ -977,7 +921,7 @@ async def _start_command(
         strategies_on_demand=tuple(body.strategies_on_demand or ()),
         prompt_requests=script_prompt_requests(),
         datasource_manifest=datasource_manifest,
-        model_manifest={"presets": presets},
+        model_manifest=resolved_manifest,
     )
 
 

script_build_host/src/script_build_host/application/input_snapshot_service.py

@@ -6,6 +6,7 @@ from typing import Any
 from script_build_host.domain.canonical_json import canonical_sha256
 from script_build_host.domain.errors import InputRelationMismatch
 from script_build_host.domain.input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
+from script_build_host.domain.model_contracts import validate_model_prompt_closure
 from script_build_host.domain.ports import (
     InputSnapshotRepository,
     LegacyInputReader,
@@ -90,6 +91,13 @@ class ScriptInputSnapshotService:
         )
         sanitized_strategies = _redacted_rows(strategies, "strategies")
         sanitized_prompts = _redacted_rows(prompts, "prompt manifest")
+        sanitized_models = redact(request.model_manifest)
+        if not isinstance(sanitized_models, dict):
+            raise InputRelationMismatch()
+        try:
+            validate_model_prompt_closure(sanitized_prompts, sanitized_models)
+        except ValueError as exc:
+            raise InputRelationMismatch() from exc
         persona_digest = canonical_sha256(
             {
                 "persona_points": sanitized_persona_points,
@@ -115,7 +123,7 @@ class ScriptInputSnapshotService:
             strategies=sanitized_strategies,
             prompt_manifest=sanitized_prompts,
             datasource_manifest=redact(request.datasource_manifest),
-            model_manifest=redact(request.model_manifest),
+            model_manifest=sanitized_models,
         )
 
     async def freeze(self, assembled: ScriptBuildInput) -> ScriptBuildInputSnapshotV1:
@@ -162,11 +170,17 @@ class ScriptInputSnapshotService:
             strategies=snapshot.strategies,
             prompt_manifest=tuple(merged[key] for key in sorted(merged)),
             datasource_manifest=snapshot.datasource_manifest,
-            model_manifest=redact(model_manifest or snapshot.model_manifest),
+            model_manifest=redact(
+                _merge_model_manifests(snapshot.model_manifest, model_manifest or {})
+            ),
             parent_snapshot_id=snapshot.snapshot_id,
             parent_snapshot_sha256=snapshot.canonical_sha256,
             business_input_sha256=business_digest,
         )
+        try:
+            validate_model_prompt_closure(extended.prompt_manifest, extended.model_manifest)
+        except ValueError as exc:
+            raise InputRelationMismatch() from exc
         return await self._snapshots.freeze(extended, version=version)
 
 
@@ -178,3 +192,13 @@ def _redacted_rows(
     if not isinstance(sanitized, list) or any(not isinstance(item, dict) for item in sanitized):
         raise InputRelationMismatch()
     return tuple(sanitized)
+
+
+def _merge_model_manifests(
+    current: dict[str, Any], extension: dict[str, Any]
+) -> dict[str, Any]:
+    current_presets = current.get("presets", current)
+    extension_presets = extension.get("presets", extension)
+    if not isinstance(current_presets, dict) or not isinstance(extension_presets, dict):
+        raise InputRelationMismatch()
+    return {"presets": {**current_presets, **extension_presets}}

script_build_host/src/script_build_host/application/mission_factory.py

@@ -13,6 +13,8 @@ from agent.tools.builtin.knowledge import KnowledgeConfig
 from script_build_host.domain.input_snapshot import ScriptBuildInputSnapshotV1
 from script_build_host.domain.records import MissionBinding
 
+_PLANNER_STALL_LIMIT = 3
+
 
 def normalize_topic_summary(value: object, *, max_length: int = 160) -> str:
     """Extract a bounded topic summary without inventing a title field."""
@@ -93,10 +95,8 @@ class ScriptMissionFactory:
         return json.dumps(
             {
                 "mission": "plan_toward_root",
-                "script_build_id": binding.script_build_id,
-                "input_snapshot_ref": f"script-build://inputs/{snapshot.snapshot_id}",
-                "input_sha256": snapshot.canonical_sha256,
                 "phase": 1,
+                "context_entrypoint": "get_current_context",
                 "phase_boundary": "PHASE_ONE_CAPABILITY_BOUNDARY",
                 "instruction": (
                     "Create Direction under Root without dispatching it; create Retrieval "
@@ -122,7 +122,10 @@ class ScriptMissionFactory:
             max_iterations=planner_config["max_iterations"],
             system_prompt=_frozen_prompt(snapshot, "script_planner"),
             completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
-            no_progress=NoProgressPolicy(progress_epoch="script-build-phase-one"),
+            no_progress=NoProgressPolicy(
+                planner_stall_limit=_PLANNER_STALL_LIMIT,
+                progress_epoch="script-build-phase-one",
+            ),
             new_trace_id=binding.root_trace_id,
             root_task_spec=self.build_root_task_spec(snapshot),
             tool_groups=None,
@@ -148,8 +151,7 @@ class ScriptMissionFactory:
             {
                 "policy": "script-build-phase-two/v1",
                 "phase": 2,
-                "input_snapshot_ref": f"script-build://inputs/{snapshot.snapshot_id}",
-                "input_sha256": snapshot.canonical_sha256,
+                "context_entrypoint": "get_current_context",
                 "instruction": (
                     "Create exactly one CandidatePortfolio as Root's Phase2 child. Plan any "
                     "number and order of Structure, Paragraph, ElementSet, Retrieval and Compare "
@@ -193,8 +195,8 @@ class ScriptMissionFactory:
         return json.dumps(
             {
                 "mission": "recover_planner_from_durable_ledger",
-                "script_build_id": script_build_id,
                 "phase": phase,
+                "context_entrypoint": "get_current_context",
                 "instruction": (
                     "Inspect the durable ledger, preserve completed Attempts and ACCEPT "
                     "decisions, and create new work only for needs_replan Tasks. Never replay "
@@ -216,13 +218,8 @@ class ScriptMissionFactory:
         return json.dumps(
             {
                 "mission": "continue_toward_root",
-                "script_build_id": binding.script_build_id,
                 "phase": 2,
-                "input_snapshot_ref": f"script-build://inputs/{snapshot.snapshot_id}",
-                "input_sha256": snapshot.canonical_sha256,
-                "accepted_direction_ref": (
-                    f"script-build://artifact-versions/{direction_artifact_version_id}"
-                ),
+                "context_entrypoint": "get_current_context",
                 "active_direction_goal_ids": list(direction_goal_ids),
                 "phase_boundary": "PHASE_TWO_CANDIDATE_PORTFOLIO_READY",
             },
@@ -242,7 +239,10 @@ class ScriptMissionFactory:
             temperature=planner_config["temperature"],
             max_iterations=planner_config["max_iterations"],
             completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
-            no_progress=NoProgressPolicy(progress_epoch="script-build-phase-two"),
+            no_progress=NoProgressPolicy(
+                planner_stall_limit=_PLANNER_STALL_LIMIT,
+                progress_epoch="script-build-phase-two",
+            ),
             trace_id=binding.root_trace_id,
             root_task_spec=None,
             tool_groups=None,
@@ -268,8 +268,7 @@ class ScriptMissionFactory:
             {
                 "policy": "script-build-phase-three/v1",
                 "phase": 3,
-                "input_snapshot_ref": f"script-build://inputs/{snapshot.snapshot_id}",
-                "input_sha256": snapshot.canonical_sha256,
+                "context_entrypoint": "get_current_context",
                 "instruction": (
                     "Deliver only the accepted Direction and CandidatePortfolio closure. "
                     "Dispatch the Host-frozen root-delivery contract, require independent "
@@ -293,13 +292,8 @@ class ScriptMissionFactory:
         return json.dumps(
             {
                 "mission": "close_root_delivery",
-                "script_build_id": binding.script_build_id,
                 "phase": 3,
-                "input_snapshot_ref": f"script-build://inputs/{snapshot.snapshot_id}",
-                "input_sha256": snapshot.canonical_sha256,
-                "accepted_direction_ref": direction_ref,
-                "accepted_candidate_portfolio_ref": portfolio_ref,
-                "adopted_structured_script_ref": structured_script_ref,
+                "context_entrypoint": "get_current_context",
             },
             ensure_ascii=False,
             sort_keys=True,
@@ -317,7 +311,10 @@ class ScriptMissionFactory:
             temperature=planner_config["temperature"],
             max_iterations=planner_config["max_iterations"],
             completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
-            no_progress=NoProgressPolicy(progress_epoch="script-build-phase-three"),
+            no_progress=NoProgressPolicy(
+                planner_stall_limit=_PLANNER_STALL_LIMIT,
+                progress_epoch="script-build-phase-three",
+            ),
             trace_id=binding.root_trace_id,
             root_task_spec=None,
             tool_groups=None,

script_build_host/src/script_build_host/application/mission_workbench.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import json
+import re
 import secrets
 from collections import Counter, defaultdict
 from collections.abc import Mapping, Sequence
@@ -29,6 +30,13 @@ from script_build_host.domain.context_broker import (
     paginate_context_cards,
     rank_context_cards,
 )
+from script_build_host.domain.errors import ScriptBuildError
+from script_build_host.domain.phase_protocol import (
+    PhaseBoundaryProtocol,
+    PhaseProtocolSnapshot,
+    PhaseTaskProtocol,
+    allowed_child_kinds,
+)
 from script_build_host.domain.phase_two_ports import ScriptTaskContractStore
 from script_build_host.domain.ports import (
     InputSnapshotRepository,
@@ -62,12 +70,12 @@ _ACTIVE = {
     TaskStatus.WAITING_CHILDREN,
     TaskStatus.BLOCKED,
 }
+_SEMANTIC_HANDLE = re.compile(r"^[a-z][a-z0-9_]*_[0-9a-f]{20}$")
 
 
-class WorkbenchError(RuntimeError):
+class WorkbenchError(ScriptBuildError):
     def __init__(self, code: str, message: str) -> None:
-        super().__init__(message)
-        self.code = code
+        super().__init__(code, message)
         self.message = message
 
 
@@ -145,6 +153,9 @@ class ContextBroker:
         if known_revision == revision:
             return NotModified(revision)
         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
+        )
         counts = Counter(task.status.value for task in ledger.tasks.values())
         included = self._active_subgraph_ids(ledger)
         active_tasks: list[dict[str, Any]] = []
@@ -205,20 +216,42 @@ class ContextBroker:
             accepted_cards,
             " ".join((root.current_spec.objective, *(item for item in all_goals))),
         )
-        selected_accepted, used_tokens, omitted = fit_context_cards(ranked_accepted, 3_000)
+        selected_accepted, _accepted_tokens, omitted = fit_context_cards(ranked_accepted, 3_000)
+        input_cards = self._input_cards(snapshot)
+        topic_card = next(
+            (card for card in input_cards if card.source_type == "topic"),
+            input_cards[0] if input_cards else None,
+        )
+        selected_inputs = (
+            ()
+            if topic_card is None
+            else (
+                replace(
+                    topic_card,
+                    summary=topic_card.summary[:160],
+                    excerpt="",
+                    metadata={**dict(topic_card.metadata), "detail_required": True},
+                ),
+            )
+        )
+        input_tokens = sum(estimate_context_tokens(card.to_payload()) for card in selected_inputs)
+        input_omitted = len(input_cards) - len(selected_inputs)
+        input_source_types = tuple(dict.fromkeys(card.source_type for card in input_cards))
         bundle = ContextBundle(
             revision=revision,
-            cards=selected_accepted,
+            cards=selected_inputs,
             required_handles=(),
             sufficiency=assess_context_sufficiency(
-                required_source_types=(), cards=selected_accepted, has_more=bool(omitted)
+                required_source_types=input_source_types,
+                cards=selected_inputs,
+                has_more=bool(input_omitted),
             ),
             receipt=ContextReceipt(
                 revision=revision,
-                candidate_count=len(accepted_cards),
-                selected_count=len(selected_accepted),
-                estimated_tokens=used_tokens,
-                omitted_count=omitted,
+                candidate_count=len(input_cards),
+                selected_count=len(selected_inputs),
+                estimated_tokens=input_tokens,
+                omitted_count=input_omitted,
             ),
         )
         await self._emit_receipt(
@@ -226,12 +259,13 @@ class ContextBroker:
             role="planner",
             task_kind="root",
             receipt=bundle.receipt,
-            has_more=bool(omitted),
+            has_more=bool(input_omitted),
         )
         selected_active = tuple(active_tasks[:12])
+        phase = _phase(ledger, root, binding.active_direction_artifact_version_id)
         state = PlannerState(
             state_revision=revision,
-            phase=_phase(ledger, root, binding.active_direction_artifact_version_id),
+            phase=phase,
             root={
                 "task_id": root.task_id,
                 "status": root.status.value,
@@ -257,13 +291,93 @@ class ContextBroker:
             active_has_more=len(active_tasks) > len(selected_active),
             accepted_total=len(accepted_cards),
             accepted_has_more=bool(omitted),
-            context_bundle=_bundle_metadata(bundle),
+            context_bundle={
+                **_bundle_metadata(bundle),
+                "cards": [_compact_card(card) for card in bundle.cards],
+            },
+            phase_protocol=(await self._phase_protocol(ledger, phase)).to_payload(),
         )
         _bounded_context(
             state.to_payload(), char_limit=24_000, token_limit=6_000, label="PlannerState"
         )
         return state
 
+    async def _phase_protocol(self, ledger: Any, phase: int) -> PhaseProtocolSnapshot:
+        root = ledger.tasks[ledger.root_task_id]
+        current = ledger.tasks.get(ledger.focused_task_id or "") or root
+        current_is_root = current.task_id == ledger.root_task_id
+        current_kind_value = (
+            None
+            if current_is_root
+            else task_kind_from_context_refs(current.current_spec.context_refs)
+        )
+        current_kind = ScriptTaskKind(current_kind_value) if current_kind_value else None
+        selectable: dict[str, list[str]] = defaultdict(list)
+        for child_id in current.child_task_ids:
+            child = ledger.tasks.get(child_id)
+            if child is None:
+                continue
+            decision = self._accepted_decision(ledger, child)
+            child_kind = task_kind_from_context_refs(child.current_spec.context_refs)
+            if decision is not None and child_kind:
+                selectable[child_kind].append(decision.decision_id)
+        actions: tuple[str, ...]
+        if current_is_root:
+            actions = ("plan", "submit_boundary")
+        elif current.status is TaskStatus.AWAITING_DECISION:
+            actions = ("validate", "accept", "repair", "retry", "revise", "split", "block")
+        elif current.status in {TaskStatus.NEEDS_REPLAN, TaskStatus.BLOCKED}:
+            actions = ("retry", "revise", "split", "block", "cancel")
+        else:
+            actions = ("plan", "dispatch", "inspect")
+        if phase == 1:
+            boundary_reason = "PHASE_ONE_CAPABILITY_BOUNDARY"
+            ready = any(
+                self._accepted_decision(ledger, item) is not None
+                and task_kind_from_context_refs(item.current_spec.context_refs)
+                == ScriptTaskKind.DIRECTION.value
+                for item in ledger.tasks.values()
+            )
+            missing = () if ready else ("accepted_direction",)
+        elif phase == 2:
+            boundary_reason = "PHASE_TWO_CANDIDATE_PORTFOLIO_READY"
+            ready = any(
+                self._accepted_decision(ledger, item) is not None
+                and task_kind_from_context_refs(item.current_spec.context_refs)
+                == ScriptTaskKind.CANDIDATE_PORTFOLIO.value
+                for item in ledger.tasks.values()
+            )
+            missing = () if ready else ("accepted_candidate_portfolio",)
+        else:
+            boundary_reason = "PHASE_THREE_ROOT_DELIVERY_READY"
+            ready = any(
+                self._accepted_decision(ledger, item) is not None
+                and task_kind_from_context_refs(item.current_spec.context_refs)
+                == ScriptTaskKind.ROOT_DELIVERY.value
+                for item in ledger.tasks.values()
+            )
+            missing = () if ready else ("accepted_root_delivery",)
+        return PhaseProtocolSnapshot(
+            phase=phase,
+            phase_epoch=f"script-build-phase-{phase}",
+            topology_version="v1",
+            current_task=PhaseTaskProtocol(
+                task_id=current.task_id,
+                task_kind=current_kind.value if current_kind is not None else "root",
+                allowed_child_kinds=tuple(
+                    sorted(
+                        item.value
+                        for item in allowed_child_kinds(phase=phase, parent_kind=current_kind)
+                    )
+                ),
+                valid_actions=actions,
+                selectable_decisions={
+                    kind: tuple(sorted(values)) for kind, values in sorted(selectable.items())
+                },
+            ),
+            boundary=PhaseBoundaryProtocol(boundary_reason, ready, missing),
+        )
+
     async def worker_state(self, request: RoleContextRequest) -> WorkerState:
         ledger = await self._task_store.load(request.root_trace_id)
         task = ledger.tasks[request.task_id]
@@ -857,6 +971,14 @@ class ContextBroker:
                 omitted_count=max(0, len(filtered) - page.offset - len(page.items)),
             ),
             has_more=page.has_more,
+            progress_key=_digest(
+                {
+                    "operation": "search",
+                    "filter": request.filter_digest(),
+                    "cursor": cursor or "start",
+                    "revision": revision,
+                }
+            ),
         )
         _bounded_json(payload, 8_000, "Context search page")
         _bounded_tokens(payload, request.token_budget, "Context search page")
@@ -875,6 +997,12 @@ class ContextBroker:
     ) -> Mapping[str, Any]:
         """Read one authorized immutable source through a bounded page."""
 
+        if not _SEMANTIC_HANDLE.fullmatch(handle):
+            raise WorkbenchError(
+                "CONTEXT_HANDLE_FORMAT_INVALID",
+                "use a semantic handle returned by get_current_context or search_mission_context",
+            )
+
         ledger = await self._task_store.load(root_trace_id)
         binding = await self._bindings.get_by_root(root_trace_id)
         snapshot = await self._snapshots.get(
@@ -894,7 +1022,7 @@ class ContextBroker:
             )
             if handle not in authorized_handles:
                 raise WorkbenchError(
-                    "CONTEXT_HANDLE_INVALID", "context handle is stale or unauthorized"
+                    "CONTEXT_HANDLE_UNAUTHORIZED", "context handle is outside this role and Task"
                 )
             authorized_ref = self._authorized_artifact_refs.get(
                 (root_trace_id, task_id, attempt_id or ""), {}
@@ -1018,6 +1146,9 @@ class ContextBroker:
             detail_page_key=_digest(
                 {"handle": handle, "section": section, "start": start, "revision": revision}
             ),
+            progress_key=_digest(
+                {"operation": "read", "handle": handle, "section": section, "start": start}
+            ),
         )
         _bounded_json(payload, 8_000, "Context detail page")
         return payload
@@ -1316,7 +1447,7 @@ class ContextBroker:
         ):
             if card.handle == handle:
                 return value, revision
-        raise WorkbenchError("CONTEXT_HANDLE_INVALID", "context handle is stale or unauthorized")
+        raise WorkbenchError("CONTEXT_HANDLE_UNKNOWN", "context handle is stale or unknown")
 
     async def _authorize_cards(
         self,
@@ -1507,6 +1638,7 @@ class ContextBroker:
         receipt: ContextReceipt,
         has_more: bool,
         detail_page_key: str | None = None,
+        progress_key: str | None = None,
     ) -> None:
         append = getattr(self._receipt_store, "append_event", None)
         if not callable(append):
@@ -1528,9 +1660,39 @@ class ContextBroker:
                 "omitted_count": receipt.omitted_count,
                 "detail_reads": receipt.detail_reads,
                 "detail_page_key": detail_page_key,
+                "progress_key": progress_key or detail_page_key,
             },
         )
 
+    async def progress_snapshot(self, root_trace_id: str) -> Mapping[str, Any]:
+        """Return durable, duplicate-insensitive Context Broker progress."""
+
+        get_events = getattr(self._receipt_store, "get_events", None)
+        if not callable(get_events):
+            return {}
+        events = await get_events(root_trace_id)
+        progress_keys = {
+            str(item.get("progress_key"))
+            for item in events
+            if item.get("event") == "context_broker_receipt" and item.get("progress_key")
+        }
+        required = {
+            str(key)
+            for item in events
+            if item.get("event") == "context_broker_required_context"
+            for key in item.get("required_source_keys") or ()
+        }
+        completed = {
+            str(item.get("source_key"))
+            for item in events
+            if item.get("event") == "context_broker_required_source_read"
+            and item.get("source_key")
+        }
+        return {
+            "unique_observations": sorted(progress_keys),
+            "required_remaining": sorted(required - completed),
+        }
+
     async def _record_required_context(
         self,
         root_trace_id: str,
@@ -1707,6 +1869,9 @@ class MissionWorkbench:
     async def planner_state(self, root_trace_id: str, **kwargs: Any) -> PlannerState | NotModified:
         return await self.broker.planner_state(root_trace_id, **kwargs)
 
+    async def progress_snapshot(self, root_trace_id: str) -> Mapping[str, Any]:
+        return await self.broker.progress_snapshot(root_trace_id)
+
     async def worker_state(self, request: RoleContextRequest) -> WorkerState:
         return await self.broker.worker_state(request)
 
@@ -2055,7 +2220,15 @@ def _phase(ledger: Any, root: Any, active_direction_id: int | None) -> int:
         for item in ledger.tasks.values()
     ):
         return 3
-    return 2 if active_direction_id is not None else 1
+    if any(
+        task_kind_from_context_refs(item.current_spec.context_refs)
+        == ScriptTaskKind.CANDIDATE_PORTFOLIO.value
+        for item in ledger.tasks.values()
+    ):
+        return 2
+    if active_direction_id is not None and root.status is not TaskStatus.WAITING_CHILDREN:
+        return 2
+    return 1
 
 
 def _scope_selector(scope_ref: str) -> dict[str, Any]:

script_build_host/src/script_build_host/application/phase_two_candidates.py

@@ -118,6 +118,12 @@ class CandidateWorkspaceOperations(Protocol):
 
     async def snapshot(self, context: CandidateWriteContext) -> CandidateWorkspaceSnapshot: ...
 
+    async def seed_from_accepted_artifacts(
+        self,
+        context: CandidateWriteContext,
+        sources: Sequence[ArtifactVersion],
+    ) -> CandidateWorkspaceSnapshot: ...
+
     async def create_paragraph(self, context: CandidateWriteContext, **values: Any) -> int: ...
 
     async def create_paragraphs(
@@ -1304,12 +1310,9 @@ class PhaseTwoCandidateService:
                     ]
                 )
                 if paragraph_versions:
-                    await self._materialize(
-                        scope,
+                    await self._workspace_repo().seed_from_accepted_artifacts(
+                        scope.write_context,
                         paragraph_versions,
-                        (),
-                        require_structure_coverage=False,
-                        require_complete=False,
                     )
         return scope
 
@@ -1501,13 +1504,36 @@ class PhaseTwoCandidateService:
         while pending:
             version = pending.pop()
             lineage = getattr(version.artifact, "lineage", None)
+            source_refs = {
+                str(item.get("source_artifact_ref"))
+                for item in getattr(lineage, "source_lineage", ())
+                if item.get("source_artifact_ref")
+            }
             base_ref = getattr(lineage, "base_artifact_ref", None)
-            if not base_ref or base_ref in version_by_ref:
-                continue
-            base_id = int(str(base_ref).rsplit("/", 1)[-1])
-            base = await self._artifacts.get_by_id(base_id, script_build_id=scope.script_build_id)
-            version_by_ref[base_ref] = base
-            pending.append(base)
+            if base_ref:
+                source_refs.add(str(base_ref))
+            for source_ref in source_refs:
+                if source_ref in version_by_ref:
+                    continue
+                source_id = int(source_ref.rsplit("/", 1)[-1])
+                source = await self._artifacts.get_by_id(
+                    source_id, script_build_id=scope.script_build_id
+                )
+                expected_digest = next(
+                    (
+                        str(item.get("source_artifact_digest"))
+                        for item in getattr(lineage, "source_lineage", ())
+                        if item.get("source_artifact_ref") == source_ref
+                        and item.get("source_artifact_digest")
+                    ),
+                    None,
+                )
+                if expected_digest and source.canonical_sha256 != expected_digest:
+                    raise PhaseTwoCandidateError(
+                        "STALE_BASE_REVISION", "candidate source identity digest changed"
+                    )
+                version_by_ref[source_ref] = source
+                pending.append(source)
 
         entity_key_cache: dict[tuple[str, str, int], tuple[str, str, int]] = {}
 
@@ -1537,7 +1563,16 @@ class PhaseTwoCandidateService:
                     "LEGACY_REFERENCE_INVALID", "candidate source identity is ambiguous"
                 )
             base_ref = getattr(lineage, "base_artifact_ref", None)
-            if matches and base_ref:
+            qualified_ref = str(matches[0].get("source_artifact_ref") or "") if matches else ""
+            if matches and qualified_ref:
+                source_id = int(matches[0]["source_local_id"])
+                source = version_by_ref.get(qualified_ref)
+                resolved = (
+                    entity_key(source, entity, source_id, trail | {cache_key})
+                    if source is not None
+                    else (entity, qualified_ref, source_id)
+                )
+            elif matches and base_ref:
                 source_id = int(matches[0]["source_local_id"])
                 base = version_by_ref.get(base_ref)
                 resolved = (

script_build_host/src/script_build_host/application/phase_two_planning.py

@@ -31,11 +31,13 @@ from script_build_host.domain.input_compatibility import (
     AcceptedInputSource,
     validate_accepted_input,
 )
+from script_build_host.domain.phase_protocol import allowed_child_kinds
 from script_build_host.domain.phase_two_ports import ScriptTaskContractStore
 from script_build_host.domain.ports import (
     MissionBindingRepository,
     ScriptBusinessArtifactRepository,
 )
+from script_build_host.domain.task_capabilities import task_capability
 from script_build_host.domain.task_contracts import (
     AcceptedDecisionRef,
     FrozenTaskContract,
@@ -57,18 +59,7 @@ PHASE_TWO_CANDIDATE_PORTFOLIO_READY = "PHASE_TWO_CANDIDATE_PORTFOLIO_READY"
 PHASE_ONE_CAPABILITY_BOUNDARY = "PHASE_ONE_CAPABILITY_BOUNDARY"
 
 TASK_PRESET_BY_KIND: dict[ScriptTaskKind, str] = {
-    ScriptTaskKind.DIRECTION: "script_direction_worker",
-    ScriptTaskKind.PATTERN_RETRIEVAL: "script_pattern_retrieval_worker",
-    ScriptTaskKind.DECODE_RETRIEVAL: "script_decode_retrieval_worker",
-    ScriptTaskKind.EXTERNAL_RETRIEVAL: "script_external_retrieval_worker",
-    ScriptTaskKind.KNOWLEDGE_RETRIEVAL: "script_knowledge_retrieval_worker",
-    ScriptTaskKind.STRUCTURE: "script_structure_worker",
-    ScriptTaskKind.PARAGRAPH: "script_paragraph_worker",
-    ScriptTaskKind.ELEMENT_SET: "script_element_set_worker",
-    ScriptTaskKind.COMPARE: "script_candidate_compare_worker",
-    ScriptTaskKind.COMPOSE: "script_compose_worker",
-    ScriptTaskKind.CANDIDATE_PORTFOLIO: "script_candidate_portfolio_worker",
-    ScriptTaskKind.ROOT_DELIVERY: "script_root_worker",
+    kind: task_capability(kind).worker_preset for kind in ScriptTaskKind
 }
 
 _PHASE_TWO_KINDS = frozenset(
@@ -181,29 +172,17 @@ class PhasePolicyGuard:
         if phase is None:
             return
         parent_is_root = parent.task_id == ledger.root_task_id
-        parent_kind = task_kind_from_context_refs(parent.current_spec.context_refs)
+        parent_kind_value = task_kind_from_context_refs(parent.current_spec.context_refs)
+        if not parent_is_root and parent_kind_value is None:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "parent Task kind is missing")
+        parent_kind = None if parent_is_root else ScriptTaskKind(cast(str, parent_kind_value))
         kinds = {item.task_kind for item in contracts}
-        if phase == 1:
-            allowed = (
-                {ScriptTaskKind.DIRECTION}
-                if parent_is_root
-                else (
-                    set(ScriptTaskKind(item) for item in RETRIEVAL_KINDS)
-                    if parent_kind == ScriptTaskKind.DIRECTION.value
-                    else set()
-                )
-            )
-        elif phase == 2:
-            allowed = (
-                {ScriptTaskKind.CANDIDATE_PORTFOLIO} if parent_is_root else set(_PHASE_TWO_KINDS)
-            )
-        else:
-            allowed = set()
+        allowed = set(allowed_child_kinds(phase=phase, parent_kind=parent_kind))
         if not kinds <= allowed:
             self._raise_planning_violation(
                 phase=phase,
                 parent_is_root=parent_is_root,
-                parent_kind=parent_kind,
+                parent_kind=parent_kind_value,
                 attempted=kinds,
                 allowed=allowed,
             )
@@ -228,29 +207,17 @@ class PhasePolicyGuard:
         if phase is None:
             return
         parent_is_root = parent.task_id == ledger.root_task_id
-        parent_kind = task_kind_from_context_refs(parent.current_spec.context_refs)
+        parent_kind_value = task_kind_from_context_refs(parent.current_spec.context_refs)
+        if not parent_is_root and parent_kind_value is None:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "parent Task kind is missing")
+        parent_kind = None if parent_is_root else ScriptTaskKind(cast(str, parent_kind_value))
         kinds = {item.task_kind for item in inputs}
-        if phase == 1:
-            allowed = (
-                {ScriptTaskKind.DIRECTION}
-                if parent_is_root
-                else (
-                    set(ScriptTaskKind(item) for item in RETRIEVAL_KINDS)
-                    if parent_kind == ScriptTaskKind.DIRECTION.value
-                    else set()
-                )
-            )
-        elif phase == 2:
-            allowed = (
-                {ScriptTaskKind.CANDIDATE_PORTFOLIO} if parent_is_root else set(_PHASE_TWO_KINDS)
-            )
-        else:
-            allowed = set()
+        allowed = set(allowed_child_kinds(phase=phase, parent_kind=parent_kind))
         if not kinds <= allowed:
             self._raise_planning_violation(
                 phase=phase,
                 parent_is_root=parent_is_root,
-                parent_kind=parent_kind,
+                parent_kind=parent_kind_value,
                 attempted=kinds,
                 allowed=allowed,
             )
@@ -1946,7 +1913,11 @@ class PhaseTwoPlanningService:
         )
         child_kinds = {item.task_kind.value for item in contracts}
         if parent_kind is None:
-            allowed = {ScriptTaskKind.DIRECTION.value, ScriptTaskKind.CANDIDATE_PORTFOLIO.value}
+            allowed = {
+                item.value
+                for phase in (1, 2)
+                for item in allowed_child_kinds(phase=phase, parent_kind=None)
+            }
             if not child_kinds <= allowed:
                 raise TaskContractError(
                     "TASK_CONTRACT_INVALID", "Root accepts only Direction or CandidatePortfolio"
@@ -2000,53 +1971,13 @@ class PhaseTwoPlanningService:
                 "WRITE_SCOPE_VIOLATION",
                 "child Task write scope must stay within its parent write scope",
             )
-        if parent_kind == ScriptTaskKind.DIRECTION.value:
-            if not child_kinds <= RETRIEVAL_KINDS:
-                raise TaskContractError(
-                    "TASK_CONTRACT_INVALID", "Direction children must be Retrieval Tasks"
-                )
-            return
-        if parent_kind == ScriptTaskKind.CANDIDATE_PORTFOLIO.value:
-            if not child_kinds <= {ScriptTaskKind.COMPOSE.value}:
-                raise TaskContractError(
-                    "TASK_CONTRACT_INVALID", "Portfolio children must be Compose candidates"
-                )
-            return
-        if parent_kind == ScriptTaskKind.COMPOSE.value:
-            allowed = {
-                ScriptTaskKind.STRUCTURE.value,
-                ScriptTaskKind.PARAGRAPH.value,
-                ScriptTaskKind.ELEMENT_SET.value,
-                ScriptTaskKind.COMPARE.value,
-                *RETRIEVAL_KINDS,
-            }
-            if not child_kinds <= allowed:
-                raise TaskContractError(
-                    "TASK_CONTRACT_INVALID", "Compose contains only local creative increments"
-                )
-            return
-        if parent_kind in {
-            ScriptTaskKind.STRUCTURE.value,
-            ScriptTaskKind.PARAGRAPH.value,
-            ScriptTaskKind.ELEMENT_SET.value,
-            ScriptTaskKind.COMPARE.value,
-        }:
-            allowed = {
-                ScriptTaskKind.STRUCTURE.value,
-                ScriptTaskKind.PARAGRAPH.value,
-                ScriptTaskKind.ELEMENT_SET.value,
-                ScriptTaskKind.COMPARE.value,
-                *RETRIEVAL_KINDS,
-            }
-            if not child_kinds <= allowed:
-                raise TaskContractError(
-                    "TASK_CONTRACT_INVALID",
-                    "a local creative Task may split only bounded local or Retrieval increments",
-                )
-            return
-        if child_kinds:
+        allowed = {
+            item.value for item in task_capability(ScriptTaskKind(parent_kind)).child_kinds
+        }
+        if not child_kinds <= allowed:
             raise TaskContractError(
-                "TASK_CONTRACT_INVALID", "this Task kind cannot own additional children"
+                "TASK_CONTRACT_INVALID",
+                f"{parent_kind} cannot own child kinds {sorted(child_kinds - allowed)}",
             )
 
     @staticmethod
@@ -2175,15 +2106,7 @@ def _scope_contains(parent: str, child: str) -> bool:
 
 
 def _write_scope_for(kind: ScriptTaskKind) -> tuple[str, ...]:
-    if kind in {ScriptTaskKind.STRUCTURE, ScriptTaskKind.PARAGRAPH}:
-        return ("script-build://writes/paragraphs",)
-    if kind is ScriptTaskKind.ELEMENT_SET:
-        # ElementSet owns Element rows and their placement links. It may read or
-        # materialize Paragraph endpoints, but it does not own Paragraph prose.
-        return ("script-build://writes/elements",)
-    if kind in {ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO}:
-        return ("script-build://writes",)
-    return ()
+    return task_capability(kind).write_scope
 
 
 def _contract_signature(contract: ScriptTaskContractV1) -> str:

script_build_host/src/script_build_host/domain/errors.py

@@ -119,6 +119,30 @@ class ProtocolViolation(ScriptBuildError):
         super().__init__("PROTOCOL_VIOLATION", summary)
 
 
+class UpstreamFailure(ScriptBuildError):
+    """Typed external dependency failure safe for retry and operator policy."""
+
+    def __init__(
+        self,
+        code: str,
+        summary: str,
+        *,
+        provider: str,
+        operation: str,
+        retryable: bool,
+        retry_after_seconds: float | None = None,
+        succeeded: int = 0,
+        failed: int = 0,
+    ) -> None:
+        super().__init__(code, summary)
+        self.provider = provider
+        self.operation = operation
+        self.retryable = retryable
+        self.retry_after_seconds = retry_after_seconds
+        self.succeeded = succeeded
+        self.failed = failed
+
+
 class LegacyCanonicalIdentityConflict(ScriptBuildError):
     def __init__(self, summary: str = "legacy projection contains conflicting logical identities"):
         super().__init__("LEGACY_CANONICAL_IDENTITY_CONFLICT", summary)

script_build_host/src/script_build_host/domain/model_contracts.py

@@ -0,0 +1,24 @@
+"""Domain checks that bind frozen prompts to executable model presets."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+
+
+def validate_model_prompt_closure(
+    prompt_manifest: tuple[Mapping[str, object], ...],
+    model_manifest: Mapping[str, object],
+) -> None:
+    """Fail before freeze when a frozen Prompt has no model configuration."""
+
+    presets = model_manifest.get("presets")
+    if not isinstance(presets, Mapping):
+        return
+    required = {str(item.get("preset") or "") for item in prompt_manifest}
+    required.discard("")
+    missing = sorted(required - set(map(str, presets)))
+    if missing:
+        raise ValueError(f"model manifest is missing frozen prompt presets: {missing}")
+
+
+__all__ = ["validate_model_prompt_closure"]

script_build_host/src/script_build_host/domain/phase_protocol.py

@@ -0,0 +1,66 @@
+"""Machine-readable phase protocol shared by guards and Planner projections."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import asdict, dataclass, field
+from typing import Any
+
+from .task_capabilities import task_capability
+from .task_contracts import ScriptTaskKind
+
+
+@dataclass(frozen=True, slots=True)
+class PhaseTaskProtocol:
+    task_id: str
+    task_kind: str
+    allowed_child_kinds: tuple[str, ...]
+    valid_actions: tuple[str, ...]
+    selectable_decisions: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class PhaseBoundaryProtocol:
+    reason: str
+    ready: bool
+    missing: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True, slots=True)
+class PhaseProtocolSnapshot:
+    phase: int
+    phase_epoch: str
+    topology_version: str
+    current_task: PhaseTaskProtocol
+    boundary: PhaseBoundaryProtocol
+
+    def to_payload(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+def allowed_child_kinds(
+    *, phase: int, parent_kind: ScriptTaskKind | None
+) -> frozenset[ScriptTaskKind]:
+    if phase == 1:
+        if parent_kind is None:
+            return frozenset({ScriptTaskKind.DIRECTION})
+        if parent_kind is ScriptTaskKind.DIRECTION:
+            return task_capability(parent_kind).child_kinds
+        return frozenset()
+    if phase == 2:
+        return (
+            frozenset({ScriptTaskKind.CANDIDATE_PORTFOLIO})
+            if parent_kind is None
+            else task_capability(parent_kind).child_kinds
+        )
+    if phase == 3:
+        return frozenset({ScriptTaskKind.ROOT_DELIVERY}) if parent_kind is None else frozenset()
+    return frozenset()
+
+
+__all__ = [
+    "PhaseBoundaryProtocol",
+    "PhaseProtocolSnapshot",
+    "PhaseTaskProtocol",
+    "allowed_child_kinds",
+]

script_build_host/src/script_build_host/domain/phase_two_ports.py

@@ -68,6 +68,12 @@ class CandidateWorkspaceRepository(Protocol):
         links: Sequence[Mapping[str, Any]],
     ) -> tuple[Mapping[str, int], int]: ...
 
+    async def seed_from_accepted_artifacts(
+        self,
+        context: CandidateWriteContext,
+        sources: Sequence[ArtifactVersion],
+    ) -> CandidateWorkspaceSnapshot: ...
+
     async def freeze(
         self,
         context: CandidateWriteContext,

script_build_host/src/script_build_host/domain/task_capabilities.py

@@ -0,0 +1,333 @@
+"""Single source of truth for Script Task execution capabilities.
+
+The Planner owns business intent.  The Host owns the executable envelope:
+worker, output schema, tools, criterion vocabulary, write ownership and child
+topology.  Keeping those facts together makes impossible contracts fail before
+an Agent or an external dependency is started.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from enum import StrEnum
+from types import MappingProxyType
+from typing import Any
+
+from .task_contracts import PlannerCriterionInput, ScriptTaskKind, TaskContractError
+
+
+class CriterionType(StrEnum):
+    CUSTOM = "custom"
+    EVIDENCE_NONEMPTY = "evidence_nonempty"
+    SOURCE_ATTRIBUTABLE = "source_attributable"
+    SEMANTIC_RELEVANCE = "semantic_relevance"
+    CONCEPT_COVERAGE = "concept_coverage"
+    FRESHNESS = "freshness"
+
+
+class CompletionMode(StrEnum):
+    WORKER_SUBMITS = "worker_submits"
+    RETRIEVAL_AUTO_SUBMIT = "retrieval_auto_submit"
+    HOST_ASSEMBLES = "host_assembles"
+
+
+@dataclass(frozen=True, slots=True)
+class TaskCapability:
+    task_kind: ScriptTaskKind
+    output_schemas: frozenset[str]
+    worker_preset: str
+    validator_preset: str
+    allowed_tools: frozenset[str]
+    criterion_types: frozenset[CriterionType]
+    write_scope: tuple[str, ...]
+    child_kinds: frozenset[ScriptTaskKind]
+    completion_mode: CompletionMode = CompletionMode.WORKER_SUBMITS
+
+
+_RETRIEVAL_CRITERIA = frozenset(
+    {
+        CriterionType.EVIDENCE_NONEMPTY,
+        CriterionType.SOURCE_ATTRIBUTABLE,
+        CriterionType.SEMANTIC_RELEVANCE,
+        CriterionType.CONCEPT_COVERAGE,
+        CriterionType.FRESHNESS,
+    }
+)
+_CUSTOM = frozenset({CriterionType.CUSTOM})
+_RETRIEVAL_KINDS = frozenset(
+    {
+        ScriptTaskKind.PATTERN_RETRIEVAL,
+        ScriptTaskKind.DECODE_RETRIEVAL,
+        ScriptTaskKind.EXTERNAL_RETRIEVAL,
+        ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
+    }
+)
+_LOCAL_CHILDREN = frozenset(
+    {
+        ScriptTaskKind.STRUCTURE,
+        ScriptTaskKind.PARAGRAPH,
+        ScriptTaskKind.ELEMENT_SET,
+        ScriptTaskKind.COMPARE,
+        *_RETRIEVAL_KINDS,
+    }
+)
+_CONTEXT_READS = frozenset({"search_mission_context", "read_mission_context"})
+
+
+def _retrieval(
+    kind: ScriptTaskKind,
+    preset: str,
+    tool: str,
+) -> TaskCapability:
+    return TaskCapability(
+        task_kind=kind,
+        output_schemas=frozenset({"evidence-record/v1"}),
+        worker_preset=preset,
+        validator_preset="script_retrieval_validator",
+        allowed_tools=frozenset({*_CONTEXT_READS, tool, "submit_attempt"}),
+        criterion_types=_RETRIEVAL_CRITERIA,
+        write_scope=(),
+        child_kinds=frozenset(),
+        completion_mode=CompletionMode.RETRIEVAL_AUTO_SUBMIT,
+    )
+
+
+_CAPABILITIES: Mapping[ScriptTaskKind, TaskCapability] = MappingProxyType(
+    {
+        ScriptTaskKind.DIRECTION: TaskCapability(
+            ScriptTaskKind.DIRECTION,
+            frozenset({"script-direction/v1"}),
+            "script_direction_worker",
+            "script_candidate_validator",
+            frozenset({*_CONTEXT_READS, "save_direction_candidate", "submit_attempt"}),
+            _CUSTOM,
+            (),
+            _RETRIEVAL_KINDS,
+        ),
+        ScriptTaskKind.PATTERN_RETRIEVAL: _retrieval(
+            ScriptTaskKind.PATTERN_RETRIEVAL,
+            "script_pattern_retrieval_worker",
+            "query_pattern_qa",
+        ),
+        ScriptTaskKind.DECODE_RETRIEVAL: _retrieval(
+            ScriptTaskKind.DECODE_RETRIEVAL,
+            "script_decode_retrieval_worker",
+            "search_script_decode_case",
+        ),
+        ScriptTaskKind.EXTERNAL_RETRIEVAL: _retrieval(
+            ScriptTaskKind.EXTERNAL_RETRIEVAL,
+            "script_external_retrieval_worker",
+            "external_search_case",
+        ),
+        ScriptTaskKind.KNOWLEDGE_RETRIEVAL: _retrieval(
+            ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
+            "script_knowledge_retrieval_worker",
+            "search_knowledge",
+        ),
+        ScriptTaskKind.STRUCTURE: TaskCapability(
+            ScriptTaskKind.STRUCTURE,
+            frozenset({"structure-artifact/v1"}),
+            "script_structure_worker",
+            "script_candidate_validator",
+            frozenset(
+                {*_CONTEXT_READS, "view_frozen_images", "save_script_paragraphs", "submit_attempt"}
+            ),
+            _CUSTOM,
+            ("script-build://writes/paragraphs",),
+            _LOCAL_CHILDREN,
+        ),
+        ScriptTaskKind.PARAGRAPH: TaskCapability(
+            ScriptTaskKind.PARAGRAPH,
+            frozenset({"paragraph-artifact/v1", "paragraph-patch/v1"}),
+            "script_paragraph_worker",
+            "script_candidate_validator",
+            frozenset(
+                {*_CONTEXT_READS, "view_frozen_images", "save_script_paragraphs", "submit_attempt"}
+            ),
+            _CUSTOM,
+            ("script-build://writes/paragraphs",),
+            _LOCAL_CHILDREN,
+        ),
+        ScriptTaskKind.ELEMENT_SET: TaskCapability(
+            ScriptTaskKind.ELEMENT_SET,
+            frozenset({"element-set-artifact/v1", "element-set-patch/v1"}),
+            "script_element_set_worker",
+            "script_candidate_validator",
+            frozenset(
+                {*_CONTEXT_READS, "view_frozen_images", "save_script_elements", "submit_attempt"}
+            ),
+            _CUSTOM,
+            ("script-build://writes/elements",),
+            _LOCAL_CHILDREN,
+        ),
+        ScriptTaskKind.COMPARE: TaskCapability(
+            ScriptTaskKind.COMPARE,
+            frozenset({"comparison-artifact/v1"}),
+            "script_candidate_compare_worker",
+            "script_candidate_validator",
+            frozenset({*_CONTEXT_READS, "save_comparison_candidate", "submit_attempt"}),
+            _CUSTOM,
+            (),
+            _LOCAL_CHILDREN,
+        ),
+        ScriptTaskKind.COMPOSE: TaskCapability(
+            ScriptTaskKind.COMPOSE,
+            frozenset({"structured-script/v1"}),
+            "script_compose_worker",
+            "script_candidate_validator",
+            frozenset({"submit_attempt"}),
+            _CUSTOM,
+            ("script-build://writes",),
+            _LOCAL_CHILDREN,
+            CompletionMode.HOST_ASSEMBLES,
+        ),
+        ScriptTaskKind.CANDIDATE_PORTFOLIO: TaskCapability(
+            ScriptTaskKind.CANDIDATE_PORTFOLIO,
+            frozenset({"candidate-portfolio/v1"}),
+            "script_candidate_portfolio_worker",
+            "script_candidate_validator",
+            frozenset({"submit_attempt"}),
+            _CUSTOM,
+            ("script-build://writes",),
+            frozenset({ScriptTaskKind.COMPOSE}),
+            CompletionMode.HOST_ASSEMBLES,
+        ),
+        ScriptTaskKind.ROOT_DELIVERY: TaskCapability(
+            ScriptTaskKind.ROOT_DELIVERY,
+            frozenset({"root-delivery-manifest/v1"}),
+            "script_root_worker",
+            "script_root_validator",
+            frozenset(
+                {
+                    *_CONTEXT_READS,
+                    "legacy_projection_dry_run",
+                    "save_root_delivery_manifest",
+                    "submit_attempt",
+                }
+            ),
+            frozenset(),
+            (),
+            frozenset(),
+        ),
+    }
+)
+
+
+def task_capability(kind: ScriptTaskKind) -> TaskCapability:
+    return _CAPABILITIES[kind]
+
+
+def task_capabilities() -> tuple[TaskCapability, ...]:
+    return tuple(_CAPABILITIES[kind] for kind in ScriptTaskKind)
+
+
+def validate_business_criteria(
+    kind: ScriptTaskKind,
+    criteria: tuple[PlannerCriterionInput, ...],
+) -> None:
+    capability = task_capability(kind)
+    invalid: list[str] = []
+    for item in criteria:
+        try:
+            criterion_type = CriterionType(item.criterion_type)
+        except ValueError:
+            invalid.append(str(item.criterion_type))
+            continue
+        if criterion_type not in capability.criterion_types:
+            invalid.append(criterion_type.value)
+    invalid = sorted(set(invalid))
+    if invalid:
+        raise TaskContractError(
+            "TASK_CAPABILITY_MISMATCH",
+            f"{kind.value} cannot produce or validate criteria types {invalid}",
+            details={
+                "task_kind": kind.value,
+                "output_schemas": sorted(capability.output_schemas),
+                "completion_mode": capability.completion_mode.value,
+                "allowed_criterion_types": sorted(
+                    item.value for item in capability.criterion_types
+                ),
+                "invalid_criterion_types": invalid,
+            },
+        )
+
+
+def render_criterion(value: PlannerCriterionInput) -> str:
+    kind = CriterionType(value.criterion_type)
+    params = dict(value.params)
+    if kind is CriterionType.CUSTOM:
+        return value.description
+    if value.description:
+        raise TaskContractError(
+            "TASK_CAPABILITY_MISMATCH",
+            "typed criteria are rendered by the Host and cannot carry free-text descriptions",
+        )
+    if kind is CriterionType.EVIDENCE_NONEMPTY:
+        minimum = _bounded_int(params, "minimum_count", default=1, minimum=1, maximum=20)
+        _only(params, {"minimum_count"})
+        return f"Evidence contains at least {minimum} usable item(s)"
+    if kind is CriterionType.SOURCE_ATTRIBUTABLE:
+        _only(params, set())
+        return "Every accepted evidence item retains an attributable source"
+    if kind in {CriterionType.SEMANTIC_RELEVANCE, CriterionType.CONCEPT_COVERAGE}:
+        concepts = params.get("concepts")
+        if (
+            not isinstance(concepts, list)
+            or not 1 <= len(concepts) <= 12
+            or any(not isinstance(item, str) or not item.strip() for item in concepts)
+        ):
+            raise TaskContractError(
+                "TASK_CAPABILITY_MISMATCH", "typed evidence criterion requires 1-12 concepts"
+            )
+        minimum = _bounded_int(
+            params,
+            "minimum_matches",
+            default=1,
+            minimum=1,
+            maximum=len(concepts),
+        )
+        _only(params, {"concepts", "minimum_matches"})
+        label = "semantically relates to" if kind is CriterionType.SEMANTIC_RELEVANCE else "covers"
+        concept_list = ", ".join(item.strip() for item in concepts)
+        return f"Evidence {label} at least {minimum} of: {concept_list}"
+    if kind is CriterionType.FRESHNESS:
+        days = _bounded_int(params, "max_age_days", default=365, minimum=1, maximum=3650)
+        _only(params, {"max_age_days"})
+        return f"Evidence is no older than {days} day(s) when publication time is available"
+    raise AssertionError(f"unhandled criterion type: {kind}")
+
+
+def _bounded_int(
+    values: Mapping[str, Any],
+    key: str,
+    *,
+    default: int,
+    minimum: int,
+    maximum: int,
+) -> int:
+    raw = values.get(key, default)
+    if isinstance(raw, bool) or not isinstance(raw, int) or not minimum <= raw <= maximum:
+        raise TaskContractError(
+            "TASK_CAPABILITY_MISMATCH", f"{key} must be between {minimum} and {maximum}"
+        )
+    return raw
+
+
+def _only(values: Mapping[str, Any], allowed: set[str]) -> None:
+    unknown = sorted(set(values) - allowed)
+    if unknown:
+        raise TaskContractError(
+            "TASK_CAPABILITY_MISMATCH", f"criterion params contain unknown fields: {unknown}"
+        )
+
+
+__all__ = [
+    "CompletionMode",
+    "CriterionType",
+    "TaskCapability",
+    "render_criterion",
+    "task_capabilities",
+    "task_capability",
+    "validate_business_criteria",
+]

script_build_host/src/script_build_host/domain/task_contracts.py

@@ -116,12 +116,20 @@ class ScopeSelector:
 @dataclass(frozen=True, slots=True)
 class PlannerCriterionInput:
     client_key: str
-    description: str
+    description: str = ""
     hard: bool = True
+    criterion_type: str = "custom"
+    params: Mapping[str, Any] = field(default_factory=dict)
 
     def __post_init__(self) -> None:
         _require_identifier(self.client_key, "criterion client_key")
-        _require_bounded_text(self.description, "criterion description", 500)
+        if self.criterion_type == "custom":
+            _require_bounded_text(self.description, "criterion description", 500)
+        elif self.description:
+            _require_bounded_text(self.description, "criterion description", 500)
+        _require_identifier(self.criterion_type, "criterion_type")
+        if not isinstance(self.params, Mapping):
+            raise TaskContractError("TASK_CONTRACT_INVALID", "criterion params must be an object")
         if not isinstance(self.hard, bool):
             raise TaskContractError("TASK_CONTRACT_INVALID", "criterion hard must be boolean")
 
@@ -129,7 +137,7 @@ class PlannerCriterionInput:
     def from_payload(cls, value: object) -> PlannerCriterionInput:
         if not isinstance(value, Mapping):
             raise TaskContractError("TASK_CONTRACT_INVALID", "criteria must contain objects")
-        if set(value) - {"client_key", "description", "hard"}:
+        if set(value) - {"client_key", "description", "hard", "criterion_type", "params"}:
             raise TaskContractError(
                 "TASK_CONTRACT_INVALID", "Planner criteria cannot provide criterion_id"
             )
@@ -137,6 +145,8 @@ class PlannerCriterionInput:
             client_key=str(value.get("client_key", "")),
             description=str(value.get("description", "")),
             hard=value.get("hard", True),
+            criterion_type=str(value.get("criterion_type", "custom")),
+            params=dict(value.get("params") or {}),
         )
 
 

script_build_host/src/script_build_host/domain/task_policy.py

@@ -6,6 +6,7 @@ from collections.abc import Sequence
 from hashlib import sha256
 from typing import Any
 
+from .task_capabilities import render_criterion, validate_business_criteria
 from .task_contracts import (
     PlannerCriterionInput,
     ScriptCriterion,
@@ -88,6 +89,7 @@ class TaskPolicyCatalog:
         business: tuple[PlannerCriterionInput, ...],
         direction_constraints: Sequence[Any] = (),
     ) -> tuple[ScriptCriterion, ...]:
+        validate_business_criteria(task_kind, business)
         fixed = _FIXED[task_kind]
         fixed_ids = {item.criterion_id for item in fixed}
         constraints = tuple(
@@ -101,7 +103,7 @@ class TaskPolicyCatalog:
         extra = tuple(
             ScriptCriterion(
                 stable_semantic_id(task_id, "criterion", item.client_key),
-                item.description,
+                render_criterion(item),
                 item.hard,
             )
             for item in business

script_build_host/src/script_build_host/domain/workbench.py

@@ -36,6 +36,7 @@ class PlannerState:
     accepted_total: int = 0
     accepted_has_more: bool = False
     context_bundle: Mapping[str, Any] = field(default_factory=dict)
+    phase_protocol: Mapping[str, Any] = field(default_factory=dict)
 
     def to_payload(self) -> dict[str, Any]:
         return asdict(self)

script_build_host/src/script_build_host/infrastructure/tables.py

@@ -100,6 +100,35 @@ artifact_version_table = Table(
     Index("ix_artifact_task", "script_build_id", "task_id"),
 )
 
+candidate_source_identity_table = Table(
+    "script_build_candidate_source_identity",
+    metadata,
+    Column("id", _primary_key_type, primary_key=True, autoincrement=True),
+    Column("workspace_artifact_version_id", BigInteger, nullable=False),
+    Column("entity_kind", String(32), nullable=False),
+    Column("target_local_id", BigInteger, nullable=False),
+    Column("source_artifact_version_id", BigInteger, nullable=False),
+    Column("source_artifact_digest", String(64), nullable=False),
+    Column("source_local_id", BigInteger, nullable=False),
+    Column("created_at", _timestamp_type, nullable=False),
+    UniqueConstraint(
+        "workspace_artifact_version_id",
+        "entity_kind",
+        "target_local_id",
+        name="uq_candidate_source_target",
+    ),
+    UniqueConstraint(
+        "workspace_artifact_version_id",
+        "source_artifact_version_id",
+        "entity_kind",
+        "source_local_id",
+        name="uq_candidate_source_origin",
+    ),
+    CheckConstraint("entity_kind IN ('paragraph', 'element')", name="ck_candidate_source_kind"),
+    CheckConstraint("length(source_artifact_digest) = 64", name="ck_candidate_source_digest"),
+    Index("ix_candidate_source_workspace", "workspace_artifact_version_id"),
+)
+
 publication_table = Table(
     "script_build_publication",
     metadata,

script_build_host/src/script_build_host/repositories/workspace.py

@@ -45,7 +45,10 @@ from script_build_host.infrastructure.legacy_tables import (
     script_build_paragraph_element,
     script_build_task_plan_step,
 )
-from script_build_host.infrastructure.tables import artifact_version_table
+from script_build_host.infrastructure.tables import (
+    artifact_version_table,
+    candidate_source_identity_table,
+)
 from script_build_host.repositories.sqlalchemy import (
     SqlAlchemyScriptBusinessArtifactRepository,
 )
@@ -180,6 +183,123 @@ class SqlAlchemyCandidateWorkspaceRepository:
             workspace = await self._locked_workspace(session, context, require_draft=False)
             return await self._snapshot_in_session(session, workspace)
 
+    async def seed_from_accepted_artifacts(
+        self,
+        context: CandidateWriteContext,
+        sources: Sequence[ArtifactVersion],
+    ) -> CandidateWorkspaceSnapshot:
+        """Atomically seed read-only Paragraph endpoints into an ElementSet draft."""
+
+        if not sources:
+            raise WorkspaceError("LEGACY_REFERENCE_INVALID", "workspace seed requires a source")
+        expected: set[tuple[int, int]] = set()
+        validated_sources: list[tuple[ArtifactVersion, ParagraphArtifactV1]] = []
+        for source in sources:
+            paragraph_artifact = source.artifact
+            if not isinstance(paragraph_artifact, ParagraphArtifactV1) or source.state not in {
+                ArtifactState.FROZEN,
+                ArtifactState.PUBLISHED,
+            }:
+                raise WorkspaceError(
+                    "LEGACY_REFERENCE_INVALID", "workspace seed sources must be frozen Paragraphs"
+                )
+            source_ref = f"script-build://artifact-versions/{source.artifact_version_id}"
+            if source_ref not in set(context.input_refs):
+                raise WorkspaceError(
+                    "LEGACY_REFERENCE_INVALID", "workspace seed source is outside accepted inputs"
+                )
+            expected.update(
+                (source.artifact_version_id, item.paragraph_id)
+                for item in paragraph_artifact.paragraphs
+            )
+            validated_sources.append((source, paragraph_artifact))
+        async with self._sessions() as session, session.begin():
+            workspace = await self._locked_workspace(session, context)
+            if workspace.artifact_kind is not ArtifactKind.ELEMENT_SET:
+                raise WorkspaceError(
+                    "LEGACY_WRITE_INVALID", "only ElementSet workspaces accept Paragraph seeds"
+                )
+            existing = tuple(
+                (
+                    await session.execute(
+                        select(candidate_source_identity_table).where(
+                            candidate_source_identity_table.c.workspace_artifact_version_id
+                            == workspace.artifact_version_id,
+                            candidate_source_identity_table.c.entity_kind == "paragraph",
+                        )
+                    )
+                )
+                .mappings()
+                .all()
+            )
+            if existing:
+                actual = {
+                    (int(item["source_artifact_version_id"]), int(item["source_local_id"]))
+                    for item in existing
+                }
+                if actual != expected:
+                    raise WorkspaceError(
+                        "STALE_BASE_REVISION", "workspace was seeded from a different input closure"
+                    )
+                return await self._snapshot_in_session(session, workspace)
+            current = await self._snapshot_in_session(session, workspace)
+            if current.paragraphs or current.elements or current.links:
+                raise WorkspaceError(
+                    "HOST_CONTRACT_INVARIANT_BROKEN",
+                    "unseeded ElementSet workspace is not empty",
+                )
+            now = _now()
+            for source, paragraph_artifact in sorted(
+                validated_sources, key=lambda item: item[0].artifact_version_id
+            ):
+                paragraph_map: dict[int, int] = {}
+                for item in sorted(
+                    paragraph_artifact.paragraphs,
+                    key=lambda value: (value.level, value.paragraph_index, value.paragraph_id),
+                ):
+                    parent_id = (
+                        paragraph_map.get(item.parent_id) if item.parent_id is not None else None
+                    )
+                    result = await session.execute(
+                        insert(self._paragraphs).values(
+                            script_build_id=workspace.script_build_id,
+                            branch_id=workspace.branch_id,
+                            base_ref_id=item.paragraph_id,
+                            paragraph_index=item.paragraph_index,
+                            level=item.level,
+                            parent_id=parent_id,
+                            name=item.name,
+                            content_range=item.content_range,
+                            theme=item.theme,
+                            form=item.form,
+                            function=item.function,
+                            feeling=item.feeling,
+                            theme_elements=list(item.theme_elements),
+                            form_elements=list(item.form_elements),
+                            function_elements=list(item.function_elements),
+                            feeling_elements=list(item.feeling_elements),
+                            description=item.description,
+                            full_description=item.full_description,
+                            is_active=item.is_active,
+                            created_at=now,
+                            updated_at=now,
+                        )
+                    )
+                    target_id = int(cast(Any, result).inserted_primary_key[0])
+                    paragraph_map[item.paragraph_id] = target_id
+                    await session.execute(
+                        insert(candidate_source_identity_table).values(
+                            workspace_artifact_version_id=workspace.artifact_version_id,
+                            entity_kind="paragraph",
+                            target_local_id=target_id,
+                            source_artifact_version_id=source.artifact_version_id,
+                            source_artifact_digest=source.canonical_sha256.removeprefix("sha256:"),
+                            source_local_id=item.paragraph_id,
+                            created_at=now,
+                        )
+                    )
+            return await self._snapshot_in_session(session, workspace)
+
     async def create_paragraph(
         self,
         context: CandidateWriteContext,
@@ -1485,7 +1605,36 @@ class SqlAlchemyCandidateWorkspaceRepository:
             .mappings()
             .all()
         )
+        qualified_source_rows = tuple(
+            (
+                await session.execute(
+                    select(candidate_source_identity_table).where(
+                        candidate_source_identity_table.c.workspace_artifact_version_id
+                        == workspace.artifact_version_id
+                    )
+                )
+            )
+            .mappings()
+            .all()
+        )
+        qualified_targets = {
+            (str(row["entity_kind"]), int(row["target_local_id"]))
+            for row in qualified_source_rows
+        }
         source_identity_map: list[dict[str, int | str]] = [
+            {
+                "entity": str(row["entity_kind"]),
+                "source_artifact_ref": (
+                    "script-build://artifact-versions/"
+                    f"{int(row['source_artifact_version_id'])}"
+                ),
+                "source_artifact_digest": "sha256:" + str(row["source_artifact_digest"]),
+                "source_local_id": int(row["source_local_id"]),
+                "local_id": int(row["target_local_id"]),
+            }
+            for row in qualified_source_rows
+        ]
+        source_identity_map.extend(
             {
                 "entity": "paragraph",
                 "source_local_id": int(row["base_ref_id"]),
@@ -1493,7 +1642,8 @@ class SqlAlchemyCandidateWorkspaceRepository:
             }
             for row in paragraph_rows
             if row["base_ref_id"] is not None
-        ]
+            and ("paragraph", int(row["id"])) not in qualified_targets
+        )
         source_identity_map.extend(
             {
                 "entity": "element",
@@ -1502,6 +1652,7 @@ class SqlAlchemyCandidateWorkspaceRepository:
             }
             for row in element_rows
             if row["base_ref_id"] is not None
+            and ("element", int(row["id"])) not in qualified_targets
         )
         return CandidateWorkspaceSnapshot(
             workspace=workspace,

script_build_host/src/script_build_host/tools/failures.py

@@ -22,6 +22,7 @@ _REPLAN_CODES = frozenset(
         "TASK_KIND_PRESET_MISMATCH",
         "TASK_STATE_CONFLICT",
         "ATTEMPT_WORKSPACE_FROZEN",
+        "CONTEXT_HANDLE_UNKNOWN",
         "WRITE_SCOPE_VIOLATION",
         "SUPERSESSION_CYCLE",
     }
@@ -41,6 +42,12 @@ _RETRY_CODES = frozenset(
         "STALE_WORKBENCH_STATE",
         "VALIDATION_EVIDENCE_INVALID",
         "CONTEXT_NOT_EXHAUSTED",
+        "CONTEXT_HANDLE_INVALID",
+        "CONTEXT_HANDLE_FORMAT_INVALID",
+        "TRANSIENT_TIMEOUT",
+        "TRANSIENT_CONNECTION_LOST",
+        "RATE_LIMITED",
+        "UPSTREAM_UNAVAILABLE",
     }
 )
 _ABORT_MARKERS = (

script_build_host/src/script_build_host/tools/gateway.py

@@ -7,6 +7,7 @@ framework's protected context and then checked against durable bindings.
 
 from __future__ import annotations
 
+import asyncio
 import re
 from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import asdict, dataclass
@@ -126,7 +127,7 @@ class LegacyScriptToolGateway:
         self.budget_guard = budget_guard
         self.root_tools = root_tools
         self.workbench = workbench
-        self._active_retrieval_attempts: set[tuple[str, str, str]] = set()
+        self._retrieval_locks: dict[tuple[str, str, str], asyncio.Lock] = {}
 
     async def ensure_dispatch_allowed(self, context: Mapping[str, Any]) -> None:
         binding = await self.bindings.get_by_root(_required(context, "root_trace_id"))
@@ -369,24 +370,44 @@ class LegacyScriptToolGateway:
         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:
+        lock = self._retrieval_locks.setdefault(retrieval_key, asyncio.Lock())
+        async with lock:
+            binding = await self.bindings.get_by_root(root_trace_id)
+            try:
+                existing = 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:
+                artifact = existing.artifact
+                frozen_query = redact(dict(query))
+                if not isinstance(artifact, EvidenceRecordV1) or not isinstance(
+                    frozen_query, dict
+                ):
+                    raise ProtocolViolation(
+                        "Retrieval Attempt already owns a non-Evidence artifact"
+                    )
+                stored_query = dict(artifact.query)
+                stored_query.pop("_retrieval_metadata", None)
+                if (
+                    artifact.source_type != source_type
+                    or artifact.tool_name != tool_name
+                    or stored_query != frozen_query
+                ):
+                    raise ProtocolViolation(
+                        "Retrieval Attempt is already bound to a different external query"
+                    )
+                return _evidence_tool_payload(
+                    existing,
+                    root_trace_id=root_trace_id,
+                    task_id=task_id,
+                    attempt_id=attempt_id,
+                    resumed=True,
+                )
             return await self._retrieve_once(source_type, tool_name, query, context)
-        finally:
-            self._active_retrieval_attempts.discard(retrieval_key)
 
     async def _retrieve_once(
         self,
@@ -431,26 +452,20 @@ class LegacyScriptToolGateway:
             content_sha256="",
             created_at=datetime.now(UTC),
         )
-        version, ref = await self.artifacts.freeze(
+        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 {
-            "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),
-        }
+        return _evidence_tool_payload(
+            version,
+            root_trace_id=_required(context, "root_trace_id"),
+            task_id=_required(context, "task_id"),
+            attempt_id=_required(context, "attempt_id"),
+            resumed=False,
+        )
 
     async def view_frozen_images(
         self, image_handles: Sequence[str], context: Mapping[str, Any]
@@ -933,6 +948,28 @@ class LegacyScriptToolGateway:
         return self.root_tools
 
 
+def _evidence_tool_payload(
+    version: Any,
+    *,
+    root_trace_id: str,
+    task_id: str,
+    attempt_id: str,
+    resumed: bool,
+) -> dict[str, Any]:
+    artifact = version.artifact
+    if not isinstance(artifact, EvidenceRecordV1):
+        raise ProtocolViolation("Retrieval Attempt does not own Evidence")
+    digest = str(version.canonical_sha256 or version.artifact_version_id)
+    return {
+        "evidence_handle": semantic_handle("evidence", root_trace_id, task_id, digest),
+        "evidence": protect_model_value(
+            _json_values(asdict(artifact)), namespace=attempt_id
+        ),
+        "source_count": len(artifact.source_refs),
+        "resumed_from_frozen_evidence": resumed,
+    }
+
+
 def _required(context: Mapping[str, Any], key: str) -> str:
     value = context.get(key)
     if not isinstance(value, str) or not value.strip():

script_build_host/src/script_build_host/tools/registry.py

@@ -12,24 +12,14 @@ 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.task_capabilities import task_capabilities
 from script_build_host.domain.workbench import protect_model_value
 
 from .failures import classify_script_tool_failure
 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",
-    "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",
-    "root-delivery": "script_root_worker",
+    item.task_kind.value: item.worker_preset for item in task_capabilities()
 }
 
 _MAX_STRUCTURED_ARGUMENT_BYTES = 2 * 1024 * 1024
@@ -947,10 +937,22 @@ def _planner_task_input_schema() -> dict[str, Any]:
                     "type": "object",
                     "properties": {
                         "client_key": {"type": "string"},
-                        "description": {"type": "string"},
+                        "criterion_type": {
+                            "type": "string",
+                            "enum": [
+                                "custom",
+                                "evidence_nonempty",
+                                "source_attributable",
+                                "semantic_relevance",
+                                "concept_coverage",
+                                "freshness",
+                            ],
+                        },
+                        "description": {"type": "string", "default": ""},
+                        "params": {"type": "object", "default": {}},
                         "hard": {"type": "boolean"},
                     },
-                    "required": ["client_key", "description", "hard"],
+                    "required": ["client_key", "criterion_type", "hard"],
                     "additionalProperties": False,
                 },
             },

script_build_host/tests/test_context_broker_integration.py

@@ -330,6 +330,16 @@ async def test_planner_hot_context_is_bounded_and_every_item_is_pageable() -> No
     assert "FORGED REASON" not in " ".join(str(item["summary"]) for item in state.accepted_frontier)
 
     assert any("REAL ARTIFACT CONTENT" in str(item["summary"]) for item in state.accepted_frontier)
+    input_cards = state.context_bundle["cards"]
+    assert input_cards
+    assert input_cards[0]["source_type"] == "topic"
+    assert input_cards[0]["handle"].startswith("input_")
+    input_page = await workbench.read_mission_context(
+        root_trace_id="root",
+        role="planner",
+        handle=input_cards[0]["handle"],
+    )
+    assert input_page["content_fragment"]
 
     cursor = None
     found: list[str] = []

script_build_host/tests/test_input_snapshot_service.py

@@ -112,15 +112,16 @@ async def test_extend_prompt_lineage_is_append_only_and_preserves_business_diges
             topic_id=30,
             topic={"topic": {"id": 30}},
             account={"account_name": "acct"},
-            prompt_manifest=(
+                prompt_manifest=(
                 {
                     "preset": "script_planner",
                     "role": "planner",
                     "content": "phase one",
                     "content_sha256": canonical_sha256("phase one").wire,
-                },
-            ),
-        )
+                    },
+                ),
+                model_manifest={"presets": {"script_planner": {"model": "fake"}}},
+            )
     )
 
     class Prompts:

script_build_host/tests/test_internal_runtime.py

@@ -7,6 +7,10 @@ import pytest
 from pydantic import SecretStr
 
 from script_build_host.agents.model_manifest import resolve_script_role_model_manifest
+from script_build_host.agents.prompts import (
+    phase_three_prompt_manifest,
+    script_build_prompt_manifest,
+)
 from script_build_host.domain.records import Principal
 from script_build_host.infrastructure.config import ScriptBuildSettings
 from script_build_host.internal_runtime import (
@@ -34,7 +38,7 @@ def _settings(tmp_path: Path, **overrides: Any) -> ScriptBuildSettings:
     return ScriptBuildSettings(**values)
 
 
-def test_role_model_manifest_covers_all_sixteen_presets_and_keeps_overrides() -> None:
+def test_role_model_manifest_covers_frozen_prompt_catalog_and_keeps_overrides() -> None:
     result = resolve_script_role_model_manifest(
         {
             "embedding_model": "text-embedding-v4",
@@ -45,7 +49,11 @@ def test_role_model_manifest_covers_all_sixteen_presets_and_keeps_overrides() ->
     )
     presets = result["presets"]
     assert isinstance(presets, dict)
-    assert len(presets) == 17
+    required = {
+        str(item["preset"])
+        for item in (*script_build_prompt_manifest(), *phase_three_prompt_manifest())
+    }
+    assert set(presets) == required
     assert presets["script_context_broker"]["temperature"] == 0.0
     assert {value["model"] for value in presets.values()} == {"qwen3.7-max"}
     assert presets["script_root_validator"]["temperature"] == 0.0

script_build_host/tests/test_migrations.py

@@ -59,7 +59,8 @@ def test_mysql_offline_sql_compiles_without_credentials(monkeypatch: object) ->
     config = Config(str(root / "alembic.ini"), output_buffer=output)
     command.upgrade(config, "head", sql=True)
     sql = output.getvalue()
-    assert sql.count("CREATE TABLE script_build_") == 5
+    assert sql.count("CREATE TABLE script_build_") == 6
+    assert "CREATE TABLE script_build_candidate_source_identity" in sql
     assert "CREATE TABLE script_build_paragraph" not in sql
     assert "CREATE TABLE script_build_element" not in sql
     assert "DROP CHECK ck_artifact_phase_one_type" in sql
@@ -144,7 +145,7 @@ async def test_mysql_phase_three_schema_semantics() -> None:
                 "READ-COMMITTED"
             )
             assert await connection.scalar(text("SELECT version_num FROM alembic_version")) == (
-                "0003_phase_three_publication"
+                "0004_candidate_source_identity"
             )
             precision = await connection.scalar(
                 text(

script_build_host/tests/test_mission_factory.py

@@ -64,7 +64,9 @@ def test_factory_reads_nested_real_topic_and_does_not_double_prefix_digest() ->
     assert "title" not in str(spec)
     assert "workflow" not in str(spec).lower()
     message = json.loads(factory.build_planner_message(_binding(), _snapshot()))
-    assert message["input_sha256"] == "sha256:" + "b" * 64
+    assert message["context_entrypoint"] == "get_current_context"
+    assert "input_snapshot_ref" not in message
+    assert "input_sha256" not in message
     assert "Retrieval children under that Direction" in message["instruction"]
     assert "exact reason PHASE_ONE_CAPABILITY_BOUNDARY" in message["instruction"]
     config = factory.build_run_config(_binding(), _snapshot())
@@ -72,6 +74,7 @@ def test_factory_reads_nested_real_topic_and_does_not_double_prefix_digest() ->
     assert config.model == "planner-model"
     assert config.temperature == 0.1
     assert config.max_iterations == 42
+    assert config.no_progress.planner_stall_limit == 3
     assert config.new_trace_id == "root"
     assert config.context["phase"] == 1
     assert config.enable_memory is False

script_build_host/tests/test_mission_workbench.py

@@ -8,7 +8,7 @@ from typing import Any
 import pytest
 from agent.orchestration import AgentRole, ArtifactRef, RoleContextRequest, TaskStatus
 
-from script_build_host.application.mission_workbench import MissionWorkbench
+from script_build_host.application.mission_workbench import MissionWorkbench, WorkbenchError
 from script_build_host.domain.task_contracts import (
     ScriptCriterion,
     ScriptIntentClass,
@@ -16,7 +16,7 @@ from script_build_host.domain.task_contracts import (
     ScriptTaskContractV1,
     ScriptTaskKind,
 )
-from script_build_host.domain.workbench import NotModified
+from script_build_host.domain.workbench import NotModified, semantic_handle
 
 
 class _TaskStore:
@@ -28,6 +28,17 @@ class _TaskStore:
         return self.ledger
 
 
+class _Receipts:
+    def __init__(self) -> None:
+        self.events: list[dict[str, Any]] = []
+
+    async def append_event(self, _trace_id: str, event: str, payload: dict[str, Any]) -> None:
+        self.events.append({"event": event, "event_id": len(self.events) + 1, **payload})
+
+    async def get_events(self, _trace_id: str) -> list[dict[str, Any]]:
+        return list(self.events)
+
+
 class _Bindings:
     async def get_by_root(self, root_trace_id: str) -> Any:
         assert root_trace_id == "root"
@@ -192,18 +203,22 @@ async def test_planner_projection_is_bounded_incremental_and_does_not_copy_retir
         decisions={},
         operations={},
     )
+    receipts = _Receipts()
     workbench = MissionWorkbench(
         task_store=_TaskStore(ledger),
         bindings=_Bindings(),
         snapshots=_Snapshots(),
         artifacts=_Artifacts(),
         contracts=_Contracts(),
+        receipt_store=receipts,
     )
 
     state = await workbench.planner_state("root")
     assert not isinstance(state, NotModified)
     assert state.retired_task_count == 120
     assert len(state.active_subgraph) == 1
+    assert state.phase_protocol["phase_epoch"] == "script-build-phase-1"
+    assert state.phase_protocol["current_task"]["allowed_child_kinds"] == ("direction",)
     encoded = await workbench.render("root", ledger)
     assert len(encoded) < 24_000
     assert "retired detail" not in encoded
@@ -227,6 +242,23 @@ async def test_planner_projection_is_bounded_incremental_and_does_not_copy_retir
     assert first_page["page"]["has_more"] is True
     assert first_page["page"]["next_cursor"]
     assert len(json.dumps(first_page, ensure_ascii=False, separators=(",", ":"))) < 8_000
+    await workbench.search_mission_context(
+        root_trace_id="root",
+        role="planner",
+        collection="tasks",
+        page_size=20,
+    )
+    progress = await workbench.progress_snapshot("root")
+    assert len(progress["unique_observations"]) == 1
+    await workbench.search_mission_context(
+        root_trace_id="root",
+        role="planner",
+        collection="tasks",
+        query="different semantic scope",
+        page_size=20,
+    )
+    progress = await workbench.progress_snapshot("root")
+    assert len(progress["unique_observations"]) == 2
 
 
 @pytest.mark.asyncio
@@ -292,3 +324,30 @@ async def test_role_bootstrap_uses_semantic_targets_and_protects_artifact_identi
     assert "paragraph_id" not in encoded_validator
     assert "script-build://artifact-versions/77" not in encoded_validator
     assert "canonical_sha256" not in encoded_validator
+
+    with pytest.raises(WorkbenchError) as physical:
+        await workbench.read_mission_context(
+            root_trace_id="root",
+            role="planner",
+            handle="script-build://artifact-versions/77",
+        )
+    assert physical.value.code == "CONTEXT_HANDLE_FORMAT_INVALID"
+
+    unknown_handle = semantic_handle("unknown", "root", "not-present")
+    with pytest.raises(WorkbenchError) as unknown:
+        await workbench.read_mission_context(
+            root_trace_id="root",
+            role="planner",
+            handle=unknown_handle,
+        )
+    assert unknown.value.code == "CONTEXT_HANDLE_UNKNOWN"
+
+    with pytest.raises(WorkbenchError) as unauthorized:
+        await workbench.read_mission_context(
+            root_trace_id="root",
+            role="worker",
+            task_id="worker",
+            attempt_id="attempt",
+            handle=unknown_handle,
+        )
+    assert unauthorized.value.code == "CONTEXT_HANDLE_UNAUTHORIZED"

script_build_host/tests/test_phase_one_e2e.py

@@ -77,6 +77,21 @@ def _contract(
     *,
     scope: str,
 ) -> dict[str, object]:
+    criterion = (
+        {
+            "client_key": criterion_id,
+            "criterion_type": "evidence_nonempty",
+            "params": {"minimum_count": 1},
+            "hard": True,
+        }
+        if kind.endswith("-retrieval")
+        else {
+            "client_key": criterion_id,
+            "criterion_type": "custom",
+            "description": description,
+            "hard": True,
+        }
+    )
     return {
         "task_kind": kind,
         "scope_selector": {
@@ -88,13 +103,7 @@ def _contract(
         "input_decision_ids": [],
         "base_decision_id": None,
         "gap_key": None,
-        "criteria": [
-            {
-                "client_key": criterion_id,
-                "description": description,
-                "hard": True,
-            }
-        ],
+        "criteria": [criterion],
         "goal_ids": [],
         "supersedes_decision_ids": [],
         "comparison_decision_ids": [],

script_build_host/tests/test_phase_two_agent_tools.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import base64
 import hashlib
 from collections.abc import Mapping
@@ -121,6 +122,7 @@ class _BusinessArtifacts:
 class _RetrievalArtifacts:
     def __init__(self) -> None:
         self.version: Any | None = None
+        self.ref: ArtifactRef | None = None
 
     async def get_by_attempt(self, **_owners: Any) -> Any:
         if self.version is None:
@@ -137,10 +139,17 @@ class _RetrievalArtifacts:
         self.version = SimpleNamespace(
             artifact_version_id=8,
             state=ArtifactState.FROZEN,
+            artifact_type=SimpleNamespace(value="evidence"),
+            canonical_sha256=ref.digest,
             artifact=artifact,
         )
+        self.ref = ref
         return self.version, ref
 
+    async def read_by_ref(self, ref: ArtifactRef, **_owners: Any) -> Any:
+        assert self.ref == ref
+        return self.version
+
 
 class _CountingRetrievalAdapter:
     def __init__(self) -> None:
@@ -148,6 +157,7 @@ class _CountingRetrievalAdapter:
 
     async def retrieve(self, **_kwargs: Any) -> RetrievalResult:
         self.calls += 1
+        await asyncio.sleep(0)
         return RetrievalResult(
             source_refs=("decode://case/1", "decode://case/1"),
             summary="one immutable result",
@@ -305,7 +315,7 @@ async def test_retrieval_budget_is_checked_before_adapter_or_snapshot_access() -
 
 
 @pytest.mark.asyncio
-async def test_one_retrieval_attempt_never_calls_adapter_twice() -> None:
+async def test_retrieval_replay_reuses_frozen_evidence_and_can_submit() -> None:
     artifacts = _RetrievalArtifacts()
     adapter = _CountingRetrievalAdapter()
 
@@ -327,10 +337,31 @@ async def test_one_retrieval_attempt_never_calls_adapter_twice() -> None:
         "attempt_id": "attempt-a",
         "spec_version": 1,
     }
-    await gateway.retrieve("decode", "search_script_decode_case", {"query": "opening"}, context)
+    first, resumed = await asyncio.gather(
+        gateway.retrieve(
+            "decode", "search_script_decode_case", {"query": "opening"}, context
+        ),
+        gateway.retrieve(
+            "decode", "search_script_decode_case", {"query": "opening"}, context
+        ),
+    )
     assert artifacts.version.artifact.source_refs == ("decode://case/1",)
     assert artifacts.version.artifact.supports == ("opening",)
-    with pytest.raises(ProtocolViolation, match="only one Evidence"):
+    assert {first["resumed_from_frozen_evidence"], resumed["resumed_from_frozen_evidence"]} == {
+        False,
+        True,
+    }
+    assert adapter.calls == 1
+
+    assert artifacts.ref is not None
+    coordinator = _Coordinator(artifacts.ref)
+    gateway.coordinator = coordinator
+    gateway.candidate_tools = _CandidateTools(artifacts.ref)
+    await gateway.submit_current_attempt(context)
+    assert coordinator.attempt_submission is not None
+    assert coordinator.attempt_submission[1].evidence_refs == [artifacts.ref]
+
+    with pytest.raises(ProtocolViolation, match="different external query"):
         await gateway.retrieve(
             "decode", "search_script_decode_case", {"query": "different"}, context
         )

script_build_host/tests/test_phase_two_candidates.py

@@ -10,6 +10,7 @@ import pytest
 from agent.orchestration import ArtifactRef
 
 from script_build_host.application.phase_two_candidates import (
+    CandidateScope,
     PhaseTwoCandidateError,
     PhaseTwoCandidateService,
     _candidate_state_revision,
@@ -66,7 +67,7 @@ WRITE = "script-build://writes/paragraphs/opening"
 
 
 def _complete_paragraph() -> ScriptParagraphV1:
-    atom = ({"原子点": "内容", "维度": "开场"},)
+    atom = ({"原子点": "内容", "维度": "开场", "维度类型": "主维度"},)
     return ScriptParagraphV1(
         1,
         1,
@@ -644,10 +645,7 @@ async def test_element_set_materializes_all_accepted_paragraphs_before_linking(
         objective="attach concrete elements to every accepted paragraph",
         input_decision_refs=(accepted_ref,),
         base_artifact_ref=None,
-        write_scope=(
-            "script-build://writes/paragraphs",
-            "script-build://writes/elements",
-        ),
+        write_scope=("script-build://writes/elements",),
         gap_ref=None,
         output_schema="element-set-artifact/v1",
         criteria=(ScriptCriterion("closed", "all elements are linked"),),
@@ -762,6 +760,175 @@ async def test_element_set_materializes_all_accepted_paragraphs_before_linking(
     assert manifest.artifact_ref.kind == ArtifactKind.ELEMENT_SET.value
 
 
+@pytest.mark.asyncio
+async def test_element_seed_lineage_survives_full_compose_projection(database: Any) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    paragraph = _complete_paragraph()
+
+    structure_version, structure_ref = await artifacts.freeze(
+        script_build_id=7,
+        task_id="structure-source",
+        attempt_id="structure-attempt",
+        spec_version=1,
+        artifact=StructureArtifactV1(_lineage(), (paragraph,)),
+    )
+    paragraph_lineage = replace(
+        _lineage(),
+        base_artifact_ref=structure_ref.uri,
+        base_artifact_digest=structure_ref.digest,
+        base_revision=structure_version.artifact_version_id,
+        source_lineage=(
+            {"entity": "paragraph", "local_id": 1, "source_local_id": 1},
+        ),
+    )
+    paragraph_version, paragraph_ref = await artifacts.freeze(
+        script_build_id=7,
+        task_id="paragraph-source",
+        attempt_id="paragraph-attempt",
+        spec_version=1,
+        artifact=ParagraphArtifactV1(
+            paragraph_lineage,
+            (paragraph,),
+            (),
+            (),
+            {"created": {"paragraph_ids": [1]}},
+        ),
+    )
+
+    element_context = CandidateWriteContext(
+        script_build_id=7,
+        task_id="element-task",
+        attempt_id="element-attempt",
+        spec_version=1,
+        objective="link one grounded element",
+        input_refs=(paragraph_ref.uri,),
+        write_scope=("script-build://writes/elements",),
+    )
+    await workspaces.get_or_create(element_context, artifact_kind=ArtifactKind.ELEMENT_SET)
+    seeded = await workspaces.seed_from_accepted_artifacts(
+        element_context, (paragraph_version,)
+    )
+    seeded_id = seeded.paragraphs[0].paragraph_id
+    await workspaces.create_elements(
+        element_context,
+        (
+            {
+                "client_key": "element",
+                "name": "grounded element",
+                "dimension_primary": "实质",
+                "dimension_secondary": "观察",
+            },
+        ),
+        ({"paragraph_id": seeded_id, "element_client_keys": ["element"]},),
+    )
+    element_version, element_ref = await workspaces.freeze(
+        element_context,
+        lineage=replace(
+            _lineage(),
+            write_scope=("script-build://writes/elements",),
+        ),
+    )
+    assert element_version.artifact.lineage.source_lineage[0]["source_artifact_ref"] == (
+        paragraph_ref.uri
+    )
+
+    compose_contract = replace(
+        _contract(ScriptTaskKind.COMPOSE),
+        write_scope=("script-build://writes",),
+    )
+    compose_context = CandidateWriteContext(
+        script_build_id=7,
+        task_id="compose-task",
+        attempt_id="compose-attempt",
+        spec_version=1,
+        objective=compose_contract.objective,
+        input_refs=(structure_ref.uri, paragraph_ref.uri, element_ref.uri),
+        write_scope=compose_contract.write_scope,
+    )
+    await workspaces.get_or_create(
+        compose_context, artifact_kind=ArtifactKind.STRUCTURED_SCRIPT
+    )
+    scope = CandidateScope(
+        root_trace_id="root",
+        script_build_id=7,
+        input_snapshot_id="11",
+        task=SimpleNamespace(task_id="compose-task"),
+        attempt=SimpleNamespace(attempt_id="compose-attempt", spec_version=1),
+        contract=compose_contract,
+        write_context=compose_context,
+    )
+    service = PhaseTwoCandidateService(
+        bindings=cast(Any, _Bindings()),
+        task_store=cast(Any, _TaskStore(SimpleNamespace())),
+        framework_artifact_store=_FrameworkArtifacts(),
+        artifacts=artifacts,
+        accepted_inputs=cast(Any, None),
+        active_frontier=ActiveFrontierResolver(),
+        workspaces=workspaces,
+    )
+    await service._materialize(
+        scope,
+        (structure_version, paragraph_version, element_version),
+        (),
+    )
+    composed = await workspaces.snapshot(compose_context)
+    assert len(composed.paragraphs) == 1
+    assert len(composed.elements) == 1
+    assert len(composed.links) == 1
+
+
+@pytest.mark.asyncio
+async def test_element_seed_disambiguates_same_local_id_and_replays_idempotently(
+    database: Any,
+) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    versions: list[ArtifactVersion] = []
+    refs: list[ArtifactRef] = []
+    for index in (1, 2):
+        paragraph = replace(
+            _complete_paragraph(),
+            name=f"paragraph-{index}",
+            full_description=f"complete source paragraph {index}",
+        )
+        version, ref = await artifacts.freeze(
+            script_build_id=7,
+            task_id=f"paragraph-{index}",
+            attempt_id=f"paragraph-attempt-{index}",
+            spec_version=1,
+            artifact=ParagraphArtifactV1(
+                _lineage(),
+                (paragraph,),
+                (),
+                (),
+                {"created": {"paragraph_ids": [1]}},
+            ),
+        )
+        versions.append(version)
+        refs.append(ref)
+    context = CandidateWriteContext(
+        script_build_id=7,
+        task_id="element-multi",
+        attempt_id="element-multi-attempt",
+        spec_version=1,
+        objective="seed two accepted paragraphs",
+        input_refs=tuple(item.uri for item in refs),
+        write_scope=("script-build://writes/elements",),
+    )
+    await workspaces.get_or_create(context, artifact_kind=ArtifactKind.ELEMENT_SET)
+    first = await workspaces.seed_from_accepted_artifacts(context, tuple(versions))
+    replay = await workspaces.seed_from_accepted_artifacts(context, tuple(reversed(versions)))
+
+    assert len(first.paragraphs) == len(replay.paragraphs) == 2
+    assert {item["source_local_id"] for item in first.source_identity_map} == {1}
+    assert {item["source_artifact_ref"] for item in first.source_identity_map} == {
+        item.uri for item in refs
+    }
+
+
 @pytest.mark.asyncio
 async def test_candidate_service_discards_existing_draft_workspace_on_abandoned_attempt(
     database: Any,

script_build_host/tests/test_phase_two_contracts.py

@@ -30,6 +30,7 @@ from script_build_host.application.phase_two_planning import (
     PhasePolicyGuard,
     PhaseTwoPlanningService,
     _execution_seconds,
+    _write_scope_for,
 )
 from script_build_host.domain.artifacts import DirectionArtifact, DirectionGoal
 from script_build_host.domain.errors import ProtocolViolation
@@ -37,11 +38,16 @@ from script_build_host.domain.records import BuildStatus, MissionBinding, Princi
 from script_build_host.domain.task_contracts import (
     PhaseTwoLimits,
     ScriptTaskContractV1,
+    ScriptTaskKind,
     TaskContractError,
 )
 from script_build_host.infrastructure.task_contract_store import FileScriptTaskContractStore
 
 
+def test_element_set_owns_only_elements_even_when_host_materializes_paragraph_endpoints() -> None:
+    assert _write_scope_for(ScriptTaskKind.ELEMENT_SET) == ("script-build://writes/elements",)
+
+
 def test_execution_budget_counts_stage_intervals_not_planner_idle_time() -> None:
     now = datetime.now(UTC)
     stages = (
@@ -145,14 +151,26 @@ def _planner_input(payload: dict[str, object]) -> dict[str, object]:
         "intent_class": payload["intent_class"],
         "objective": payload["objective"],
         "goal_ids": payload["goal_ids"],
-        "criteria": [
-            {
-                "client_key": item["criterion_id"],
-                "description": item["description"],
-                "hard": item.get("hard", True),
-            }
-            for item in cast(list[dict[str, Any]], payload["criteria"])
-        ],
+        "criteria": (
+            [
+                {
+                    "client_key": "evidence-present",
+                    "criterion_type": "evidence_nonempty",
+                    "params": {"minimum_count": 1},
+                    "hard": True,
+                }
+            ]
+            if str(payload["task_kind"]).endswith("retrieval")
+            else [
+                {
+                    "client_key": item["criterion_id"],
+                    "criterion_type": "custom",
+                    "description": item["description"],
+                    "hard": item.get("hard", True),
+                }
+                for item in cast(list[dict[str, Any]], payload["criteria"])
+            ]
+        ),
         "input_decision_ids": [item["decision_id"] for item in input_refs],
         "base_decision_id": next(
             (
@@ -189,10 +207,8 @@ def test_phase_policy_rejects_workspace_contract_without_capability_write_scope(
     guard = PhasePolicyGuard()
     ledger = SimpleNamespace(root_task_id="root")
     parent = SimpleNamespace(
-        task_id="portfolio",
-        current_spec=SimpleNamespace(
-            context_refs=("script-build://task-kinds/candidate-portfolio",)
-        ),
+        task_id="compose",
+        current_spec=SimpleNamespace(context_refs=("script-build://task-kinds/compose",)),
     )
     invalid = ScriptTaskContractV1.from_payload(_contract("structure"))
     with pytest.raises(TaskContractError, match="WRITE_SCOPE_VIOLATION"):

script_build_host/tests/test_phase_two_real_runner_e2e.py

@@ -161,6 +161,21 @@ def _contract(
             None,
         )
         assert base_decision_id is not None
+    criterion = (
+        {
+            "client_key": f"{task_kind}-closed",
+            "criterion_type": "evidence_nonempty",
+            "params": {"minimum_count": 1},
+            "hard": True,
+        }
+        if task_kind.endswith("-retrieval")
+        else {
+            "client_key": f"{task_kind}-closed",
+            "criterion_type": "custom",
+            "description": "the immutable output is concrete and causally closed",
+            "hard": True,
+        }
+    )
     return {
         "task_kind": task_kind,
         "scope_selector": {
@@ -178,13 +193,7 @@ def _contract(
         "input_decision_ids": input_decision_ids,
         "base_decision_id": base_decision_id,
         "gap_key": None,
-        "criteria": [
-            {
-                "client_key": f"{task_kind}-closed",
-                "description": "the immutable output is concrete and causally closed",
-                "hard": True,
-            }
-        ],
+        "criteria": [criterion],
         "goal_ids": list(goal_ids or ()),
         "supersedes_decision_ids": list(supersedes_decision_ids),
         "comparison_decision_ids": [str(item["decision_id"]) for item in comparison_decision_refs],

script_build_host/tests/test_phase_two_sql_flow.py

@@ -93,6 +93,21 @@ def _contract(
         intent = "replace"
     else:
         intent = "explore"
+    criterion = (
+        {
+            "client_key": "closed",
+            "criterion_type": "evidence_nonempty",
+            "params": {"minimum_count": 1},
+            "hard": True,
+        }
+        if kind is ScriptTaskKind.DECODE_RETRIEVAL
+        else {
+            "client_key": "closed",
+            "criterion_type": "custom",
+            "description": "the frozen increment is concrete and independently verifiable",
+            "hard": True,
+        }
+    )
     payload: dict[str, Any] = {
         "task_kind": kind.value,
         "scope_selector": {
@@ -112,13 +127,7 @@ def _contract(
             None,
         ),
         "gap_key": None,
-        "criteria": [
-            {
-                "client_key": "closed",
-                "description": "the frozen increment is concrete and independently verifiable",
-                "hard": True,
-            }
-        ],
+        "criteria": [criterion],
         "goal_ids": (
             []
             if kind in {ScriptTaskKind.DIRECTION, ScriptTaskKind.DECODE_RETRIEVAL}

script_build_host/tests/test_retrieval_adapters.py

@@ -7,6 +7,7 @@ from types import SimpleNamespace
 import httpx
 import pytest
 
+from script_build_host.adapters.legacy_retrieval import LegacyExternalRetrievalAdapter
 from script_build_host.adapters.retrieval import (
     ExternalRetrievalAdapter,
     FileDecodeRetrievalAdapter,
@@ -17,7 +18,7 @@ from script_build_host.adapters.retrieval import (
     SafeImageAdapter,
 )
 from script_build_host.adapters.uploaded_topic import SqlUploadedTopicGateway
-from script_build_host.domain.errors import ProtocolViolation, UnsafeOutboundTarget
+from script_build_host.domain.errors import ProtocolViolation, UnsafeOutboundTarget, UpstreamFailure
 from script_build_host.infrastructure.outbound import OutboundPolicy
 from script_build_host.repositories.legacy_input import LegacySqlAlchemyInputReader
 
@@ -88,6 +89,107 @@ async def test_http_retrieval_revalidates_redirect_and_freezes_response_metadata
             await safe.request("GET", "https://api.example.com/large")
 
 
+@pytest.mark.asyncio
+async def test_safe_http_retries_transport_and_classifies_auth_without_retry() -> None:
+    calls = 0
+
+    def flaky(_request: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise httpx.RemoteProtocolError("disconnected")
+        return httpx.Response(200, json={"ok": True})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(flaky)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+            retry_backoff_seconds=0,
+        )
+        assert await safe.json(
+            "POST",
+            "https://api.example.com/search",
+            retryable=True,
+            provider="xhs",
+            operation="search",
+        ) == {"ok": True}
+    assert calls == 2
+
+    calls = 0
+
+    def unauthorized(_request: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        return httpx.Response(401)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(unauthorized)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+            retry_backoff_seconds=0,
+        )
+        with pytest.raises(UpstreamFailure) as raised:
+            await safe.json(
+                "POST",
+                "https://api.example.com/search",
+                retryable=True,
+                provider="xhs",
+                operation="search",
+            )
+    assert raised.value.code == "UPSTREAM_AUTH_REJECTED"
+    assert calls == 1
+
+
+@pytest.mark.asyncio
+async def test_xhs_detail_failure_keeps_partial_search_cards() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/search":
+            return httpx.Response(
+                200,
+                json={
+                    "code": 0,
+                    "data": {
+                        "data": [
+                            {"id": f"note-{index}", "title": f"card-{index}"}
+                            for index in range(5)
+                        ],
+                        "has_more": False,
+                    },
+                },
+            )
+        identifier = request.read().decode("utf-8")
+        if "note-2" in identifier:
+            raise httpx.RemoteProtocolError("detail disconnected")
+        return httpx.Response(
+            200,
+            json={
+                "code": 0,
+                "data": {"data": {"id": identifier, "title": "complete detail"}},
+            },
+        )
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
+        safe = SafeHttpClient(
+            client,
+            OutboundPolicy(frozenset({"api.example.com"}), resolver=_public_resolver),
+            retry_backoff_seconds=0,
+        )
+        result = await LegacyExternalRetrievalAdapter(
+            xhs_search_endpoint="https://api.example.com/search",
+            xhs_detail_endpoint="https://api.example.com/detail",
+            zhihu_search_endpoint="https://api.example.com/zhihu",
+            http=safe,
+        ).retrieve(
+            query={"keyword": "case", "platform_channel": "xhs", "max_count": 5},
+            snapshot=SimpleNamespace(),
+        )
+
+    assert len(json.loads(result.summary)) == 5
+    assert result.metadata["partial_result"] is True
+    assert result.metadata["detail_failures"] == 1
+    assert any("search cards retained" in item for item in result.limitations)
+
+
 @pytest.mark.asyncio
 async def test_external_response_secrets_are_redacted_before_evidence_summary() -> None:
     def handler(_request: httpx.Request) -> httpx.Response:

script_build_host/tests/test_task_capabilities.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from agent.core.presets import AGENT_PRESETS
+
+from script_build_host.agents.presets import register_script_presets
+from script_build_host.domain.phase_protocol import allowed_child_kinds
+from script_build_host.domain.task_capabilities import task_capabilities, task_capability
+from script_build_host.domain.task_contracts import ScriptTaskKind, output_schema_for
+
+
+def test_capability_catalog_closes_worker_tool_schema_and_topology_contracts() -> None:
+    register_script_presets()
+    capabilities = task_capabilities()
+    assert {item.task_kind for item in capabilities} == set(ScriptTaskKind)
+
+    for capability in capabilities:
+        preset = AGENT_PRESETS[capability.worker_preset]
+        assert set(preset.allowed_tools) == set(capability.allowed_tools)
+        assert output_schema_for(capability.task_kind) in capability.output_schemas
+        assert allowed_child_kinds(
+            phase=2,
+            parent_kind=capability.task_kind,
+        ) == capability.child_kinds
+
+
+def test_phase_root_topology_is_machine_readable_and_fail_closed() -> None:
+    assert allowed_child_kinds(phase=1, parent_kind=None) == frozenset(
+        {ScriptTaskKind.DIRECTION}
+    )
+    assert allowed_child_kinds(phase=2, parent_kind=None) == frozenset(
+        {ScriptTaskKind.CANDIDATE_PORTFOLIO}
+    )
+    assert allowed_child_kinds(phase=3, parent_kind=None) == frozenset(
+        {ScriptTaskKind.ROOT_DELIVERY}
+    )
+    assert not allowed_child_kinds(phase=99, parent_kind=None)
+
+
+def test_catalog_keeps_element_write_ownership_narrow() -> None:
+    capability = task_capability(ScriptTaskKind.ELEMENT_SET)
+    assert capability.write_scope == ("script-build://writes/elements",)
+    assert "save_script_paragraphs" not in capability.allowed_tools

script_build_host/tests/test_task_policy.py

@@ -1,7 +1,13 @@
 from __future__ import annotations
 
+import pytest
+
 from script_build_host.domain.artifacts import DirectionConstraint
-from script_build_host.domain.task_contracts import PlannerCriterionInput, ScriptTaskKind
+from script_build_host.domain.task_contracts import (
+    PlannerCriterionInput,
+    ScriptTaskKind,
+    TaskContractError,
+)
 from script_build_host.domain.task_policy import TaskPolicyCatalog
 
 
@@ -32,3 +38,33 @@ def test_task_policy_injects_protected_kind_and_direction_constraint_criteria()
     assert criteria[1].hard is True
     assert criteria[2].description == "the opening is memorable"
     assert criteria[2].hard is False
+
+
+def test_retrieval_criteria_are_typed_and_host_rendered() -> None:
+    criteria = TaskPolicyCatalog().criteria(
+        task_id="external",
+        task_kind=ScriptTaskKind.EXTERNAL_RETRIEVAL,
+        business=(
+            PlannerCriterionInput(
+                client_key="coverage",
+                criterion_type="concept_coverage",
+                params={"concepts": ["占器械", "玩手机", "双重标准"], "minimum_matches": 2},
+            ),
+        ),
+    )
+
+    assert criteria[1].description == "Evidence covers at least 2 of: 占器械, 玩手机, 双重标准"
+
+
+def test_retrieval_rejects_free_text_synthesis_contract_before_dispatch() -> None:
+    with pytest.raises(TaskContractError, match="TASK_CAPABILITY_MISMATCH"):
+        TaskPolicyCatalog().criteria(
+            task_id="external",
+            task_kind=ScriptTaskKind.EXTERNAL_RETRIEVAL,
+            business=(
+                PlannerCriterionInput(
+                    client_key="persona-analysis",
+                    description="write a persona analysis and element mapping document",
+                ),
+            ),
+        )