统一阶段二工作区与资源编排

script_build_host/src/script_build_host/agents/validation.py

@@ -30,38 +30,29 @@ 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
+from script_build_host.domain.task_capabilities import (
+    TaskCapability,
+    TaskCapabilityCatalog,
+)
+from script_build_host.domain.task_contracts import (
+    ScriptTaskContractV2,
+    ScriptTaskKind,
+    TaskContractError,
+)
 
 TASK_KIND_PREFIX = "script-build://task-kinds/"
+_CATALOG = TaskCapabilityCatalog()
 RETRIEVAL_KINDS = frozenset(
-    {
-        "pattern-retrieval",
-        "decode-retrieval",
-        "external-retrieval",
-        "knowledge-retrieval",
-    }
+    item.task_kind.value for item in _CATALOG.all() if item.evidence_retrieval
 )
 PHASE_TWO_KINDS = frozenset(
-    {
-        "structure",
-        "paragraph",
-        "element-set",
-        "compare",
-        "compose",
-        "candidate-portfolio",
-    }
+    item.value for item in _CATALOG.kinds_for_phase(2)
 )
-ROOT_DELIVERY_KIND = "root-delivery"
+ROOT_DELIVERY_KIND = ScriptTaskKind.ROOT_DELIVERY.value
 _ARTIFACT_KIND_BY_TASK_KIND = {
-    "direction": ArtifactKind.DIRECTION,
-    "structure": ArtifactKind.STRUCTURE,
-    "paragraph": ArtifactKind.PARAGRAPH,
-    "element-set": ArtifactKind.ELEMENT_SET,
-    "compare": ArtifactKind.COMPARISON,
-    "compose": ArtifactKind.STRUCTURED_SCRIPT,
-    "candidate-portfolio": ArtifactKind.CANDIDATE_PORTFOLIO,
-    ROOT_DELIVERY_KIND: ArtifactKind.ROOT_DELIVERY_MANIFEST,
+    item.task_kind.value: ArtifactKind(item.artifact_kind)
+    for item in _CATALOG.all()
+    if item.artifact_kind
 }
 _DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
 
@@ -82,45 +73,26 @@ class ScriptBuildValidationPolicy:
 
     def plan(self, context: ValidationContext) -> ValidationPlan:
         kind = task_kind(context.task.current_spec.context_refs)
-        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
+        capability = _capability(kind)
+        preset = _frozen_validator_preset(context)
+        if preset is None:
+            raise ValueError(
+                "V2 validation requires one atomically frozen executable contract"
+            )
         preflight_rules = [
             "immutable-reference-present",
             "sha256-digest-shape",
             (
                 "root-delivery-artifact-shape"
-                if kind == ROOT_DELIVERY_KIND
+                if capability.validation_layer == "root-delivery"
                 else (
                     "phase-two-artifact-shape"
-                    if kind in PHASE_TWO_KINDS
+                    if capability.phase == 2
                     else "phase-one-artifact-shape"
                 )
             ),
         ]
-        if kind in PHASE_TWO_KINDS:
-            preflight_rules.extend(["business-artifact-owner", "realized-content"])
-            if kind in {"structure", "paragraph", "element-set", "compare"}:
-                preflight_rules.append("candidate-lineage")
-            if kind == "compare":
-                preflight_rules.append("comparison-fairness")
-            if kind == "compose":
-                preflight_rules.append("goal-coverage")
-            if kind == "candidate-portfolio":
-                preflight_rules.extend(["input-closure-digest", "goal-coverage"])
-        elif kind == ROOT_DELIVERY_KIND:
-            preflight_rules.extend(
-                [
-                    "root-manifest-shape",
-                    "root-blocking-defects-empty",
-                    "root-input-closure",
-                    "root-script-content",
-                    "root-goal-coverage",
-                    "root-direction-contract",
-                    "root-legacy-projection",
-                ]
-            )
+        preflight_rules.extend(capability.preflight_rules)
         return ValidationPlan(
             mode=ValidationMode.AGENT,
             validator_preset=preset,
@@ -131,6 +103,25 @@ class ScriptBuildValidationPolicy:
         )
 
 
+def _frozen_validator_preset(context: ValidationContext) -> str | None:
+    refs = [
+        item
+        for item in context.task.current_spec.context_refs
+        if item.startswith("script-build://task-contracts/sha256/")
+        and item in context.context_documents
+    ]
+    if not refs:
+        return None
+    if len(refs) != 1:
+        raise ValueError("TaskSpec contains multiple frozen V2 contracts")
+    document = context.context_documents[refs[0]]
+    payload = getattr(document, "payload", None)
+    if not isinstance(payload, Mapping):
+        raise ValueError("frozen V2 contract payload is invalid")
+    contract = ScriptTaskContractV2.from_payload(payload)
+    return contract.execution_profile.validator_preset
+
+
 def precheck_snapshot(context: ValidationContext) -> list[DeterministicRuleResult]:
     """Validate immutable reference shape without interpreting creative quality."""
 
@@ -158,15 +149,16 @@ def precheck_snapshot(context: ValidationContext) -> list[DeterministicRuleResul
         )
     )
     kind = task_kind(context.task.current_spec.context_refs)
-    if kind in RETRIEVAL_KINDS:
+    capability = _capability(kind)
+    if capability.evidence_retrieval:
         shape_ok = (
             not context.snapshot.artifact_refs
             and len(context.snapshot.evidence_refs) == 1
             and context.snapshot.evidence_refs[0].kind == ArtifactKind.EVIDENCE.value
         )
         expected = "one evidence reference"
-    elif kind in _ARTIFACT_KIND_BY_TASK_KIND:
-        expected_kind = _ARTIFACT_KIND_BY_TASK_KIND[kind]
+    elif capability.artifact_kind:
+        expected_kind = ArtifactKind(capability.artifact_kind)
         shape_ok = (
             len(context.snapshot.artifact_refs) == 1
             and not context.snapshot.evidence_refs
@@ -180,10 +172,10 @@ def precheck_snapshot(context: ValidationContext) -> list[DeterministicRuleResul
         DeterministicRuleResult(
             rule_id=(
                 "root-delivery-artifact-shape"
-                if kind == ROOT_DELIVERY_KIND
+                if capability.validation_layer == "root-delivery"
                 else (
                     "phase-two-artifact-shape"
-                    if kind in PHASE_TWO_KINDS
+                    if capability.phase == 2
                     else "phase-one-artifact-shape"
                 )
             ),
@@ -198,20 +190,16 @@ def precheck_snapshot(context: ValidationContext) -> list[DeterministicRuleResul
 
 def validation_layer(kind: str) -> str:
     """Return the semantic layer without granting the Validator decision rights."""
+    return _capability(kind).validation_layer
+
 
-    if kind == "compare":
-        return "compare"
-    if kind == "compose":
-        return "global"
-    if kind == "candidate-portfolio":
-        return "governance"
-    if kind == ROOT_DELIVERY_KIND:
-        return "root-delivery"
-    if kind in PHASE_TWO_KINDS:
-        return "local"
-    if kind in RETRIEVAL_KINDS or kind == "direction":
-        return "phase-one"
-    raise ValueError(f"unsupported script-build task kind: {kind!r}")
+def _capability(kind: str | None) -> TaskCapability:
+    if kind is None:
+        raise ValueError("script-build task kind is missing")
+    try:
+        return _CATALOG.get(ScriptTaskKind(kind))
+    except (TypeError, ValueError, TaskContractError) as exc:
+        raise ValueError(f"unsupported script-build task kind: {kind!r}") from exc
 
 
 def precheck_business_artifact(artifact: BusinessArtifact) -> tuple[str, ...]:
@@ -270,14 +258,15 @@ class ScriptBuildDeterministicValidator:
             "snapshot_id": context.snapshot.snapshot_id,
             "spec_version": context.task.current_spec_version,
         }
-        if kind in PHASE_TWO_KINDS and self._candidates is not None:
+        layer = validation_layer(kind or "")
+        if layer in {"local", "compare", "global", "governance"} and self._candidates is not None:
             values.extend(
                 _rule_result(item)
                 for item in await self._candidates.deterministic_precheck(
                     context=protected_context
                 )
             )
-        elif kind == ROOT_DELIVERY_KIND and self._root_delivery is not None:
+        elif layer == "root-delivery" and self._root_delivery is not None:
             values.extend(
                 _rule_result(item)
                 for item in await self._root_delivery.deterministic_precheck(

script_build_host/src/script_build_host/application/mission_workbench.py

@@ -13,7 +13,16 @@ from hashlib import sha256
 from typing import Any, cast
 
 from agent.failures import ToolExecutionError
-from agent.orchestration import AgentRole, DecisionAction, RoleContextRequest, TaskStatus
+from agent.orchestration import (
+    AgentRole,
+    ArtifactRef,
+    DecisionAction,
+    RoleContextRequest,
+    TaskStatus,
+    ValidationContextBootstrap,
+    ValidationEvidenceGrant,
+)
+from agent.orchestration.validation_evidence import evidence_handle
 
 from script_build_host.application.context_semantics import AdaptiveSemanticSelector
 from script_build_host.application.failure_policy import classify_script_failure
@@ -44,8 +53,10 @@ from script_build_host.domain.ports import (
     MissionBindingRepository,
     ScriptBusinessArtifactRepository,
 )
+from script_build_host.domain.task_capabilities import TaskCapabilityCatalog
 from script_build_host.domain.task_contracts import (
     ScriptTaskContractV1,
+    ScriptTaskContractV2,
     ScriptTaskKind,
     task_kind_from_context_refs,
 )
@@ -110,8 +121,6 @@ class ContextBroker:
         self._receipt_store = receipt_store
         self._receipt_lock = asyncio.Lock()
         self._cursor_codec = ContextCursorCodec(cursor_secret or secrets.token_bytes(32))
-        self._required_reads: dict[tuple[str, str, str], set[str]] = {}
-        self._required_read_offsets: dict[tuple[str, str, str, str], int] = {}
         self._authorized_artifact_refs: dict[
             tuple[str, str, str], dict[str, Mapping[str, Any]]
         ] = {}
@@ -140,6 +149,18 @@ class ContextBroker:
                 )
             ) from exc
 
+    async def build_validation_context(
+        self, request: RoleContextRequest
+    ) -> ValidationContextBootstrap:
+        """Project evidence once; Coordinator becomes the durable authority."""
+
+        state, grants = await self._validator_projection(request)
+        return ValidationContextBootstrap(
+            role_context=_json_safe(state.to_payload()),
+            evidence_grants=grants,
+            required_handles=tuple(state.required_handles),
+        )
+
     async def planner_state(
         self,
         root_trace_id: str,
@@ -296,7 +317,7 @@ class ContextBroker:
             },
             phase_protocol=(await self._phase_protocol(ledger, phase)).to_payload(),
         )
-        return _fit_planner_state(state)
+        return state
 
     async def _phase_protocol(self, ledger: Any, phase: int) -> PhaseProtocolSnapshot:
         root = ledger.tasks[ledger.root_task_id]
@@ -584,10 +605,7 @@ class ContextBroker:
             8_000
             if contract.task_kind
             in {
-                ScriptTaskKind.PATTERN_RETRIEVAL,
-                ScriptTaskKind.DECODE_RETRIEVAL,
-                ScriptTaskKind.EXTERNAL_RETRIEVAL,
-                ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
+                ScriptTaskKind.EVIDENCE_RETRIEVAL,
                 ScriptTaskKind.DIRECTION,
             }
             else 12_000
@@ -666,12 +684,15 @@ class ContextBroker:
                 else None
             ),
         )
-        _bounded_context(
-            state.to_payload(), char_limit=32_000, token_limit=token_budget, label="WorkerState"
-        )
         return state
 
     async def validator_state(self, request: RoleContextRequest) -> ValidatorState:
+        state, _ = await self._validator_projection(request)
+        return state
+
+    async def _validator_projection(
+        self, request: RoleContextRequest
+    ) -> tuple[ValidatorState, tuple[ValidationEvidenceGrant, ...]]:
         ledger = await self._task_store.load(request.root_trace_id)
         task = ledger.tasks[request.task_id]
         contract = await self._contract(request.root_trace_id, task)
@@ -698,69 +719,63 @@ class ContextBroker:
         if resolver is not None and hasattr(resolver, "validation_evidence_refs"):
             for item in await resolver.validation_evidence_refs(context=context):
                 refs.append(asdict(item))
-        handles: list[dict[str, Any]] = []
-        artifacts: list[dict[str, Any]] = []
-        seen_uris: set[str] = set()
+        grants_by_uri: dict[str, ValidationEvidenceGrant] = {}
+        raw_by_uri: dict[str, Mapping[str, Any]] = {}
         for raw in refs:
             uri = str(raw.get("uri", ""))
-            if not uri or uri in seen_uris:
+            if not uri:
                 continue
-            seen_uris.add(uri)
-            handle = _source_handle(uri, str(raw.get("digest") or raw.get("version") or ""))
-            handles.append(
-                {"evidence_handle": handle, "kind": raw.get("kind"), "summary": raw.get("summary")}
-            )
-            ref = _artifact_ref(raw)
-            version = await self._artifacts.read_by_ref(
-                ref, script_build_id=binding.script_build_id
+            grant = ValidationEvidenceGrant(
+                artifact_ref=_artifact_ref(raw),
+                handle=evidence_handle(_artifact_ref(raw)),
+                source="host_validation_closure",
             )
-            artifacts.append(
-                {
-                    "source_handle": handle,
-                    "content": _protected_artifact_payload(version.artifact, handle),
-                }
-            )
-        self._authorized_artifact_refs[
-            (request.root_trace_id, request.task_id, request.attempt_id)
-        ] = {
-            _source_handle(
-                str(raw.get("uri", "")),
-                str(raw.get("digest") or raw.get("version") or ""),
-            ): raw
-            for raw in refs
-            if raw.get("uri")
-        }
+            prior = grants_by_uri.get(uri)
+            if prior is not None and prior.artifact_ref.digest != grant.artifact_ref.digest:
+                raise WorkbenchError(
+                    "VALIDATION_EVIDENCE_CONFLICT",
+                    "one evidence URI has multiple immutable digests",
+                )
+            grants_by_uri[uri] = grant
+            raw_by_uri[uri] = raw
+        grants = tuple(grants_by_uri.values())
+        loaded = await self._load_unique_versions(
+            tuple(raw_by_uri.values()),
+            script_build_id=binding.script_build_id,
+        )
+        handles = tuple(
+            {
+                "evidence_handle": evidence_handle(_artifact_ref(raw)),
+                "kind": raw.get("kind"),
+                "summary": raw.get("summary"),
+            }
+            for raw, _ in loaded
+        )
+        artifacts = tuple(
+            {
+                "source_handle": evidence_handle(_artifact_ref(raw)),
+                "content": _protected_artifact_payload(
+                    version.artifact,
+                    evidence_handle(_artifact_ref(raw)),
+                ),
+            }
+            for raw, version in loaded
+        )
         cards = tuple(
             self._artifact_card(
                 version.artifact,
-                handle=_source_handle(
-                    str(raw.get("uri", "")),
-                    str(raw.get("digest") or raw.get("version") or ""),
-                ),
+                handle=evidence_handle(_artifact_ref(raw)),
                 source_type=str(raw.get("kind") or "artifact"),
             )
-            for raw, version in await self._load_unique_versions(
-                refs, script_build_id=binding.script_build_id
-            )
+            for raw, version in loaded
+        )
+        selected_cards = cards
+        used_tokens = estimate_context_tokens(
+            {"cards": [item.to_payload() for item in cards]}
         )
-        selected_cards, used_tokens, omitted = fit_context_cards(cards, 32_000)
         full_artifact_payload = {"items": artifacts}
         full_artifact_tokens = estimate_context_tokens(full_artifact_payload)
-        full_content_in_bootstrap = not omitted and full_artifact_tokens + used_tokens <= 32_000
-        required_handles = () if full_content_in_bootstrap else tuple(card.handle for card in cards)
-        read_key = (request.root_trace_id, request.task_id, request.attempt_id)
-        if required_handles:
-            self._required_reads[read_key] = set(required_handles)
-            for handle in required_handles:
-                self._required_read_offsets[(*read_key, handle)] = 0
-        else:
-            self._required_reads.pop(read_key, None)
-        await self._record_required_context(
-            request.root_trace_id,
-            task_id=request.task_id,
-            attempt_id=request.attempt_id,
-            required_handles=required_handles,
-        )
+        required_handles: tuple[str, ...] = ()
         bundle = ContextBundle(
             revision=_digest({"ledger": ledger.revision, "refs": refs}),
             cards=selected_cards,
@@ -778,10 +793,8 @@ class ContextBroker:
                 revision=_digest({"ledger": ledger.revision, "refs": refs}),
                 candidate_count=len(cards),
                 selected_count=len(selected_cards),
-                estimated_tokens=(
-                    used_tokens + full_artifact_tokens if full_content_in_bootstrap else used_tokens
-                ),
-                omitted_count=(0 if full_content_in_bootstrap else len(cards)),
+                estimated_tokens=used_tokens + full_artifact_tokens,
+                omitted_count=0,
             ),
         )
         await self._emit_receipt(
@@ -789,12 +802,11 @@ class ContextBroker:
             role="validator",
             task_kind=contract.task_kind.value,
             receipt=bundle.receipt,
-            has_more=not full_content_in_bootstrap,
+            has_more=False,
         )
         artifact_payload = {
-            "items": artifacts if full_content_in_bootstrap else [],
-            "cards": [card.to_payload() for card in selected_cards],
-            "full_content_in_bootstrap": full_content_in_bootstrap,
+            "items": artifacts,
+            "full_content_in_bootstrap": True,
         }
         state = ValidatorState(
             state_revision=_digest(
@@ -806,17 +818,14 @@ class ContextBroker:
             ),
             task=_semantic_contract(contract),
             artifact=artifact_payload,
-            evidence_handles=tuple(handles),
+            evidence_handles=handles,
             accepted_closure=tuple(
                 self._accepted_input_handle(item) for item in _context_decision_refs(contract)
             ),
             required_handles=required_handles,
             context_bundle=bundle.to_payload(),
         )
-        metadata = state.to_payload() | {"artifact": {"item_count": len(artifacts)}}
-        _bounded_json(metadata, 16_000, "ValidatorState metadata")
-        _bounded_tokens(state.to_payload(), 32_000, "ValidatorState")
-        return state
+        return state, grants
 
     async def search_mission_context(
         self,
@@ -1074,18 +1083,6 @@ class ContextBroker:
                 raise WorkbenchError(code, str(exc)) from exc
         fragment = _context_fragment(encoded, start=start, token_budget=1_500)
         next_offset = start + len(fragment)
-        read_offset_key = (
-            root_trace_id,
-            task_id or "",
-            attempt_id or "",
-            handle,
-        )
-        expected_offset = self._required_read_offsets.get(read_offset_key)
-        if section == "content" and expected_offset is not None and start != expected_offset:
-            raise WorkbenchError(
-                "CONTEXT_CURSOR_OUT_OF_SEQUENCE",
-                "required context pages must be read once in order",
-            )
         next_cursor = (
             self._cursor_codec.encode(
                 root_key=root_trace_id,
@@ -1105,28 +1102,6 @@ class ContextBroker:
             "next_cursor": next_cursor,
             "exhausted": next_offset >= len(encoded),
         }
-        if (
-            payload["exhausted"]
-            and section == "content"
-            and task_id is not None
-            and attempt_id is not None
-        ):
-            key = (root_trace_id, task_id, attempt_id)
-            pending = self._required_reads.get(key)
-            if pending is not None:
-                self._required_read_offsets[read_offset_key] = next_offset
-                pending.discard(handle)
-                if not pending:
-                    self._required_reads.pop(key, None)
-                self._required_read_offsets.pop(read_offset_key, None)
-            await self._record_required_source_read(
-                root_trace_id,
-                task_id=task_id,
-                attempt_id=attempt_id,
-                handle=handle,
-            )
-        elif section == "content" and expected_offset is not None:
-            self._required_read_offsets[read_offset_key] = next_offset
         await self._emit_receipt(
             root_trace_id,
             role=role,
@@ -1149,61 +1124,78 @@ class ContextBroker:
         _bounded_json(payload, 8_000, "Context detail page")
         return payload
 
-    def assert_context_exhausted(
-        self, *, root_trace_id: str, task_id: str, attempt_id: str
-    ) -> None:
-        pending = self._required_reads.get((root_trace_id, task_id, attempt_id), set())
-        if pending:
-            raise WorkbenchError(
-                "CONTEXT_NOT_EXHAUSTED",
-                f"Validator must finish {len(pending)} required context source(s)",
-            )
-
-    async def verify_context_exhausted(
-        self, *, root_trace_id: str, task_id: str, attempt_id: str
-    ) -> None:
-        """Verify required reads from durable Trace receipts when available."""
+    async def read_validation_evidence(
+        self,
+        *,
+        root_trace_id: str,
+        validation_id: str,
+        artifact_ref: ArtifactRef,
+        handle: str,
+        section: str = "content",
+        cursor: str | None = None,
+    ) -> Mapping[str, Any]:
+        """Read a Coordinator-authorized immutable ref without local ACL state."""
 
-        get_events = getattr(self._receipt_store, "get_events", None)
-        if not callable(get_events):
-            self.assert_context_exhausted(
-                root_trace_id=root_trace_id, task_id=task_id, attempt_id=attempt_id
-            )
-            return
-        events = await get_events(root_trace_id)
-        required_event = next(
-            (
-                item
-                for item in reversed(events)
-                if item.get("event") == "context_broker_required_context"
-                and item.get("task_id") == task_id
-                and item.get("attempt_id") == attempt_id
-            ),
-            None,
+        if not _SEMANTIC_HANDLE.fullmatch(handle):
+            raise WorkbenchError("CONTEXT_HANDLE_FORMAT_INVALID", "invalid evidence handle")
+        if section not in {"content", "outline"}:
+            raise WorkbenchError("INVALID_TOOL_ARGUMENT", "unsupported context section")
+        binding = await self._bindings.get_by_root(root_trace_id)
+        version = await self._artifacts.read_by_ref(
+            artifact_ref, script_build_id=binding.script_build_id
+        )
+        content = protect_model_value(
+            _artifact_payload(version.artifact), namespace=handle
         )
-        if required_event is None:
-            self.assert_context_exhausted(
-                root_trace_id=root_trace_id, task_id=task_id, attempt_id=attempt_id
+        if section == "outline":
+            content = _outline(content)
+        encoded = _json(content)
+        revision = str(artifact_ref.digest or artifact_ref.version or handle)
+        filter_digest = _digest(
+            {
+                "handle": handle,
+                "section": section,
+                "validation_id": validation_id,
+            }
+        )
+        start = 0
+        if cursor is not None:
+            try:
+                decoded = self._cursor_codec.decode(
+                    cursor,
+                    expected_root_key=root_trace_id,
+                    expected_revision=revision,
+                    expected_view=f"validation:{validation_id}:{handle}:{section}",
+                    expected_filter_digest=filter_digest,
+                )
+                start = int(decoded.last_sort_key[0])
+            except (ContextCursorError, ValueError) as exc:
+                raise WorkbenchError(
+                    getattr(exc, "code", "CONTEXT_CURSOR_INVALID"), str(exc)
+                ) from exc
+        fragment = _context_fragment(encoded, start=start, token_budget=1_500)
+        next_offset = start + len(fragment)
+        next_cursor = (
+            self._cursor_codec.encode(
+                root_key=root_trace_id,
+                revision=revision,
+                view=f"validation:{validation_id}:{handle}:{section}",
+                filter_digest=filter_digest,
+                last_sort_key=(str(next_offset),),
             )
-            return
-        required = set(required_event.get("required_source_keys") or ())
-        completed = {
-            str(item.get("source_key"))
-            for item in events
-            if int(item.get("event_id") or 0) > int(required_event.get("event_id") or 0)
-            and item.get("event") == "context_broker_required_source_read"
-            and item.get("task_id") == task_id
-            and item.get("attempt_id") == attempt_id
+            if next_offset < len(encoded)
+            else None
+        )
+        payload = {
+            "state_revision": revision,
+            "source_handle": handle,
+            "section": section,
+            "content_fragment": fragment,
+            "next_cursor": next_cursor,
+            "exhausted": next_cursor is None,
         }
-        pending = required - completed
-        if pending:
-            raise WorkbenchError(
-                "CONTEXT_NOT_EXHAUSTED",
-                f"Validator must finish {len(pending)} required context source(s)",
-            )
-
-    def pending_required_context(self, *, root_trace_id: str, task_id: str, attempt_id: str) -> int:
-        return len(self._required_reads.get((root_trace_id, task_id, attempt_id), ()))
+        _bounded_json(payload, 8_000, "Validation evidence page")
+        return payload
 
     async def _accepted_decision_card(
         self,
@@ -1672,64 +1664,7 @@ class ContextBroker:
             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,
-        *,
-        task_id: str,
-        attempt_id: str,
-        required_handles: Sequence[str],
-    ) -> None:
-        append = getattr(self._receipt_store, "append_event", None)
-        if callable(append):
-            await self._append_trace_event(
-                root_trace_id,
-                "context_broker_required_context",
-                {
-                    "task_id": task_id,
-                    "attempt_id": attempt_id,
-                    "required_source_keys": [
-                        _digest({"source_handle": item}) for item in required_handles
-                    ],
-                },
-            )
-
-    async def _record_required_source_read(
-        self,
-        root_trace_id: str,
-        *,
-        task_id: str,
-        attempt_id: str,
-        handle: str,
-    ) -> None:
-        append = getattr(self._receipt_store, "append_event", None)
-        if callable(append):
-            await self._append_trace_event(
-                root_trace_id,
-                "context_broker_required_source_read",
-                {
-                    "task_id": task_id,
-                    "attempt_id": attempt_id,
-                    "source_key": _digest({"source_handle": handle}),
-                },
-            )
+        return {"unique_observations": sorted(progress_keys)}
 
     async def _append_trace_event(
         self, root_trace_id: str, event: str, payload: dict[str, Any]
@@ -1862,6 +1797,11 @@ class MissionWorkbench:
     async def build(self, request: RoleContextRequest) -> Mapping[str, Any]:
         return await self.broker.build(request)
 
+    async def build_validation_context(
+        self, request: RoleContextRequest
+    ) -> ValidationContextBootstrap:
+        return await self.broker.build_validation_context(request)
+
     async def planner_state(self, root_trace_id: str, **kwargs: Any) -> PlannerState | NotModified:
         return await self.broker.planner_state(root_trace_id, **kwargs)
 
@@ -1880,36 +1820,21 @@ class MissionWorkbench:
     async def read_mission_context(self, **kwargs: Any) -> Mapping[str, Any]:
         return await self.broker.read_mission_context(**kwargs)
 
-    def assert_context_exhausted(self, **kwargs: Any) -> None:
-        self.broker.assert_context_exhausted(**kwargs)
-
-    async def verify_context_exhausted(self, **kwargs: Any) -> None:
-        await self.broker.verify_context_exhausted(**kwargs)
-
-    def pending_required_context(self, **kwargs: Any) -> int:
-        return self.broker.pending_required_context(**kwargs)
-
+    async def read_validation_evidence(self, **kwargs: Any) -> Mapping[str, Any]:
+        return await self.broker.read_validation_evidence(**kwargs)
 
 def _semantic_contract(contract: ScriptTaskContractV1) -> dict[str, Any]:
-    return {
+    value = {
         "task_kind": contract.task_kind.value,
-        "scope_selector": _scope_selector(contract.scope_ref),
-        "intent_class": contract.intent_class.value,
         "objective": contract.objective,
         "goal_ids": list(contract.goal_ids),
-        "criteria": [item.to_payload() for item in contract.criteria],
-        "input_decision_ids": [item.decision_id for item in contract.input_decision_refs],
-        "comparison_decision_ids": [item.decision_id for item in contract.comparison_decision_refs],
-        "base_decision_id": next(
-            (
-                item.decision_id
-                for item in contract.input_decision_refs
-                if contract.base_artifact_ref is not None
-                and item.artifact_ref.uri == contract.base_artifact_ref.uri
-            ),
-            None,
-        ),
+        "decision_ids": [
+            item.decision_id for item in _context_decision_refs(contract)
+        ],
     }
+    if isinstance(contract, ScriptTaskContractV2):
+        value["branch_ref"] = contract.branch_ref
+    return value
 
 
 def _context_decision_refs(contract: ScriptTaskContractV1) -> tuple[Any, ...]:
@@ -1942,25 +1867,17 @@ def _input_summary(snapshot: Any) -> dict[str, Any]:
 
 
 def required_context_source_types(task_kind: ScriptTaskKind) -> tuple[str, ...]:
-    return {
-        ScriptTaskKind.DIRECTION: ("topic", "persona", "section_pattern", "strategy"),
-        ScriptTaskKind.STRUCTURE: ("direction", "section_pattern"),
-        ScriptTaskKind.PARAGRAPH: ("direction", "structure", "persona"),
-        ScriptTaskKind.ELEMENT_SET: ("paragraph",),
-        ScriptTaskKind.COMPARE: (),
-        ScriptTaskKind.ROOT_DELIVERY: ("direction", "candidate-portfolio", "structured-script"),
-    }.get(task_kind, ("topic",))
+    try:
+        return TaskCapabilityCatalog().get(task_kind).context_source_types
+    except Exception:
+        return ("topic",)
 
 
 def _persona_limit(task_kind: ScriptTaskKind) -> int:
-    return {
-        ScriptTaskKind.DIRECTION: 32,
-        ScriptTaskKind.PATTERN_RETRIEVAL: 16,
-        ScriptTaskKind.STRUCTURE: 24,
-        ScriptTaskKind.PARAGRAPH: 12,
-        ScriptTaskKind.ELEMENT_SET: 12,
-        ScriptTaskKind.ROOT_DELIVERY: 16,
-    }.get(task_kind, 16)
+    try:
+        return TaskCapabilityCatalog().get(task_kind).persona_limit
+    except Exception:
+        return 16
 
 
 def _select_persona_entries(
@@ -2286,47 +2203,6 @@ def _bounded_tokens(value: Any, limit: int, label: str) -> None:
         raise WorkbenchError("CONTEXT_BUDGET_EXCEEDED", f"{label} exceeds {limit} tokens")
 
 
-def _bounded_context(value: Any, *, char_limit: int, token_limit: int, label: str) -> None:
-    _bounded_json(value, char_limit, label)
-    _bounded_tokens(value, token_limit, label)
-
-
-def _fit_planner_state(state: PlannerState) -> PlannerState:
-    candidates = (
-        state,
-        replace(
-            state,
-            active_subgraph=state.active_subgraph[:8],
-            accepted_frontier=state.accepted_frontier[:4],
-            active_operations=state.active_operations[:4],
-            latest_failures=state.latest_failures[:4],
-            changes=state.changes[:4],
-            active_has_more=state.active_has_more or len(state.active_subgraph) > 8,
-            accepted_has_more=state.accepted_has_more or len(state.accepted_frontier) > 4,
-        ),
-        replace(
-            state,
-            active_subgraph=state.active_subgraph[:3],
-            accepted_frontier=state.accepted_frontier[:2],
-            active_operations=state.active_operations[:2],
-            latest_failures=state.latest_failures[:2],
-            changes=(),
-            active_has_more=state.active_has_more or len(state.active_subgraph) > 3,
-            accepted_has_more=state.accepted_has_more or len(state.accepted_frontier) > 2,
-            context_bundle={
-                **dict(state.context_bundle),
-                "cards": [],
-                "compacted": True,
-            },
-        ),
-    )
-    for candidate in candidates:
-        payload = candidate.to_payload()
-        if len(_json(payload)) <= 24_000 and estimate_context_tokens(payload) <= 6_000:
-            return candidate
-    raise WorkbenchError("CONTEXT_BUDGET_EXCEEDED", "minimal PlannerState exceeds budget")
-
-
 def _context_fragment(value: str, *, start: int, token_budget: int) -> str:
     """Return the largest bounded fragment without splitting context by byte size."""
 

script_build_host/src/script_build_host/application/phase_two_candidates.py

@@ -21,6 +21,7 @@ from script_build_host.application.phase_two_inputs import (
     AcceptedInputResolver,
     ActiveFrontierResolver,
 )
+from script_build_host.application.workspace_bootstrapper import WorkspaceBootstrapper
 from script_build_host.domain.artifacts import (
     ArtifactKind,
     ArtifactState,
@@ -63,6 +64,7 @@ from script_build_host.domain.task_contracts import (
     AcceptedInput,
     AcceptedInputBundleV1,
     ScriptTaskContractV1,
+    ScriptTaskContractV2,
     ScriptTaskKind,
 )
 from script_build_host.domain.workbench import (
@@ -1057,10 +1059,7 @@ class PhaseTwoCandidateService:
             for item in bundle.inputs
             if item.task_kind
             in {
-                ScriptTaskKind.PATTERN_RETRIEVAL,
-                ScriptTaskKind.DECODE_RETRIEVAL,
-                ScriptTaskKind.EXTERNAL_RETRIEVAL,
-                ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
+                ScriptTaskKind.EVIDENCE_RETRIEVAL,
             }
         }
         for version in candidate_versions:
@@ -1302,13 +1301,20 @@ class PhaseTwoCandidateService:
             snapshot = await self._workspace_repo().snapshot(scope.write_context)
             if not snapshot.paragraphs and not snapshot.elements and not snapshot.links:
                 _, bundle = await self._resolve_bundle(context)
-                paragraph_versions = tuple(
-                    [
-                        await self._read_accepted_version(scope, item)
-                        for item in bundle.inputs
-                        if item.task_kind is ScriptTaskKind.PARAGRAPH
-                    ]
-                )
+                if not isinstance(scope.contract, ScriptTaskContractV2):
+                    paragraph_versions = tuple(
+                        [
+                            await self._read_accepted_version(scope, item)
+                            for item in bundle.inputs
+                            if item.task_kind is ScriptTaskKind.PARAGRAPH
+                        ]
+                    )
+                else:
+                    paragraph_versions = await WorkspaceBootstrapper().resolve_sources(
+                        contract=scope.contract,
+                        accepted_inputs=bundle.inputs,
+                        read_version=lambda item: self._read_accepted_version(scope, item),
+                    )
                 if paragraph_versions:
                     await self._workspace_repo().seed_from_accepted_artifacts(
                         scope.write_context,
@@ -1801,10 +1807,7 @@ class PhaseTwoCandidateService:
 
 def _artifact_kind(kind: ScriptTaskKind) -> ArtifactKind:
     mapping = {
-        ScriptTaskKind.PATTERN_RETRIEVAL: ArtifactKind.EVIDENCE,
-        ScriptTaskKind.DECODE_RETRIEVAL: ArtifactKind.EVIDENCE,
-        ScriptTaskKind.EXTERNAL_RETRIEVAL: ArtifactKind.EVIDENCE,
-        ScriptTaskKind.KNOWLEDGE_RETRIEVAL: ArtifactKind.EVIDENCE,
+        ScriptTaskKind.EVIDENCE_RETRIEVAL: ArtifactKind.EVIDENCE,
         ScriptTaskKind.DIRECTION: ArtifactKind.DIRECTION,
         ScriptTaskKind.STRUCTURE: ArtifactKind.STRUCTURE,
         ScriptTaskKind.PARAGRAPH: ArtifactKind.PARAGRAPH,

script_build_host/src/script_build_host/application/phase_two_inputs.py

@@ -20,7 +20,7 @@ from agent.orchestration import (
 from script_build_host.domain.artifacts import ArtifactState, ArtifactVersion
 from script_build_host.domain.canonical_json import canonical_sha256
 from script_build_host.domain.errors import ScriptBuildError
-from script_build_host.domain.input_compatibility import (
+from script_build_host.domain.legacy_input_compatibility import (
     AcceptedInputCompatibilityError,
     AcceptedInputSource,
     scopes_compatible,
@@ -29,11 +29,13 @@ from script_build_host.domain.input_compatibility import (
 from script_build_host.domain.phase_two_artifacts import CandidateLineageV1
 from script_build_host.domain.phase_two_ports import ScriptTaskContractStore
 from script_build_host.domain.ports import ScriptBusinessArtifactRepository
+from script_build_host.domain.task_capabilities import TaskCapabilityCatalog
 from script_build_host.domain.task_contracts import (
     AcceptedDecisionRef,
     AcceptedInput,
     AcceptedInputBundleV1,
     ScriptTaskContractV1,
+    ScriptTaskContractV2,
     ScriptTaskKind,
     task_kind_from_context_refs,
 )
@@ -45,6 +47,7 @@ _TERMINAL = {
     TaskStatus.BLOCKED,
     TaskStatus.CANCELLED,
 }
+_V2_CAPABILITIES = TaskCapabilityCatalog()
 
 
 class PhaseTwoInputError(ScriptBuildError):
@@ -174,19 +177,12 @@ class AcceptedInputResolver:
                     contract,
                     item.contract,
                     direct_child=True,
-                    same_parent=False,
                 )
             else:
-                consumer_task = ledger.tasks[task_id]
-                producer_task = ledger.tasks[item.artifact_version.task_id]
                 _validate_consumer_input(
                     contract,
                     item.contract,
                     direct_child=False,
-                    same_parent=(
-                        consumer_task.parent_task_id is not None
-                        and consumer_task.parent_task_id == producer_task.parent_task_id
-                    ),
                 )
             resolved.append(item)
 
@@ -567,8 +563,20 @@ def _validate_consumer_input(
     produced: ScriptTaskContractV1,
     *,
     direct_child: bool,
-    same_parent: bool,
 ) -> None:
+    if isinstance(consumer, ScriptTaskContractV2) and isinstance(
+        produced, ScriptTaskContractV2
+    ):
+        if not _V2_CAPABILITIES.allows_input(
+            consumer.task_kind,
+            produced.task_kind,
+            same_branch=consumer.branch_ref == produced.branch_ref,
+        ):
+            raise PhaseTwoInputError(
+                "INPUT_SCOPE_MISMATCH",
+                "accepted input violates the frozen branch relationship",
+            )
+        return
     try:
         validate_accepted_input(
             consumer_kind=consumer.task_kind,
@@ -578,7 +586,6 @@ def _validate_consumer_input(
             source=(
                 AcceptedInputSource.DIRECT_CHILD if direct_child else AcceptedInputSource.EXPLICIT
             ),
-            same_parent=same_parent,
         )
     except AcceptedInputCompatibilityError as exc:
         raise PhaseTwoInputError(

script_build_host/src/script_build_host/application/workbench_workers.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, cast
 
 from agent.failures import ToolExecutionError
 from agent.orchestration import (
@@ -12,20 +12,33 @@ from agent.orchestration import (
 )
 
 from script_build_host.domain.errors import ScriptBuildError
-from script_build_host.domain.task_contracts import ScriptTaskKind, task_kind_from_context_refs
+from script_build_host.domain.task_contracts import (
+    ScriptTaskContractV2,
+    ScriptTaskKind,
+    task_kind_from_context_refs,
+)
 from script_build_host.tools.failures import classify_script_tool_failure
 
 
 class ScriptBuildDeterministicWorker:
-    def __init__(self, candidates: Any, *, bindings: Any | None = None) -> None:
+    def __init__(
+        self,
+        candidates: Any,
+        *,
+        bindings: Any | None = None,
+        task_store: Any | None = None,
+    ) -> None:
         self._candidates = candidates
         self._bindings = bindings
+        self._task_store = task_store
+        self.evidence_orchestrator: Any | None = None
         self.dispatch_guard: Any | None = None
 
     async def supports(self, context: DeterministicWorkerContext) -> bool:
         return task_kind_from_context_refs(context.task.current_spec.context_refs) in {
             ScriptTaskKind.COMPOSE.value,
             ScriptTaskKind.CANDIDATE_PORTFOLIO.value,
+            ScriptTaskKind.EVIDENCE_RETRIEVAL.value,
         }
 
     async def execute(self, context: DeterministicWorkerContext) -> DeterministicWorkerResult:
@@ -80,6 +93,31 @@ class ScriptBuildDeterministicWorker:
                     payload={"unresolved_defects": [], "superseded_decision_ids": []},
                     context=context,
                 )
+            if kind == ScriptTaskKind.EVIDENCE_RETRIEVAL.value:
+                if self.evidence_orchestrator is None or self._task_store is None:
+                    raise RuntimeError("logical evidence orchestrator is not configured")
+                ledger = await self._task_store.load(str(context["root_trace_id"]))
+                task = ledger.tasks[str(context["task_id"])]
+                refs = [
+                    item
+                    for item in task.current_spec.context_refs
+                    if item.startswith("script-build://task-contracts/sha256/")
+                ]
+                if len(refs) != 1 or refs[0] not in ledger.context_documents:
+                    raise RuntimeError("logical evidence Task has no frozen V2 contract")
+                contract = ScriptTaskContractV2.from_payload(
+                    ledger.context_documents[refs[0]].payload
+                )
+                spec_version = context.get("spec_version")
+                if isinstance(spec_version, bool) or not isinstance(spec_version, int):
+                    raise RuntimeError("logical evidence Task has no spec version")
+                return await self.evidence_orchestrator.execute(
+                    root_trace_id=str(context["root_trace_id"]),
+                    task_id=str(context["task_id"]),
+                    attempt_id=str(context["attempt_id"]),
+                    spec_version=spec_version,
+                    contract=contract,
+                )
             raise RuntimeError("unsupported deterministic script worker")
 
         guard = self.dispatch_guard
@@ -100,7 +138,7 @@ class ScriptBuildDeterministicWorker:
             }:
                 return None
             raise
-        return manifest.artifact_ref
+        return cast(ArtifactRef, manifest.artifact_ref)
 
 
 __all__ = ["ScriptBuildDeterministicWorker"]

script_build_host/src/script_build_host/application/workspace_bootstrapper.py

@@ -0,0 +1,53 @@
+"""Deterministic workspace seeding from a frozen V2 contract."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Sequence
+from typing import TypeVar
+
+from script_build_host.domain.task_contracts import (
+    ScriptTaskContractV2,
+    TaskContractError,
+)
+
+_Source = TypeVar("_Source")
+_Version = TypeVar("_Version")
+
+
+class WorkspaceBootstrapper:
+    """Resolve only the semantic source selected by Contract Compiler."""
+
+    async def resolve_sources(
+        self,
+        *,
+        contract: ScriptTaskContractV2,
+        accepted_inputs: Sequence[_Source],
+        read_version: Callable[[_Source], Awaitable[_Version]],
+    ) -> tuple[_Version, ...]:
+        seed = contract.workspace_seed
+        if seed is None:
+            return ()
+        selected = tuple(
+            item
+            for item in accepted_inputs
+            if getattr(item, "decision_id", None) == seed.source_decision_id
+            and getattr(item, "task_kind", None) is seed.source_kind
+        )
+        if len(selected) != 1:
+            raise TaskContractError(
+                "TASK_TARGET_AMBIGUOUS",
+                "frozen workspace seed does not identify one accepted source",
+            )
+        version = await read_version(selected[0])
+        expected_prefix = (
+            f"script-build://targets/{getattr(version, 'artifact_version_id', '')}"
+        )
+        if not seed.semantic_target_ref.startswith(expected_prefix):
+            raise TaskContractError(
+                "STALE_BASE_REVISION",
+                "workspace seed target no longer matches its frozen artifact",
+            )
+        return (version,)
+
+
+__all__ = ["WorkspaceBootstrapper"]

script_build_host/src/script_build_host/composition.py

@@ -26,8 +26,11 @@ from script_build_host.agents.prompt_catalog import (
 )
 from script_build_host.api import ApiSecurity, UploadedTopicGateway, create_app
 from script_build_host.application.context_semantics import AdaptiveSemanticSelector
+from script_build_host.application.direction_evidence import DirectionEvidenceOrchestrator
+from script_build_host.application.direction_evidence_gate import DirectionEvidenceGate
 from script_build_host.application.input_snapshot_service import ScriptInputSnapshotService
 from script_build_host.application.mission_factory import ScriptMissionFactory
+from script_build_host.application.mission_failure_flow import MissionFailureFlow
 from script_build_host.application.mission_service import (
     BuildTransitionGate,
     DirectionReconciler,
@@ -48,7 +51,9 @@ from script_build_host.application.recovery import (
     PhaseThreeFinalizationService,
     PhaseThreeRecoveryService,
 )
+from script_build_host.application.retrieval_ports import RetrievalAdapter
 from script_build_host.application.root_delivery import RootDeliveryService
+from script_build_host.application.task_planning_v2 import TaskPlanningServiceV2
 from script_build_host.application.workbench_workers import ScriptBuildDeterministicWorker
 from script_build_host.domain.ports import (
     BuildAuthorizer,
@@ -64,7 +69,6 @@ from script_build_host.domain.task_contracts import PhaseTwoLimits
 from script_build_host.infrastructure.outbound import OutboundPolicy
 from script_build_host.infrastructure.task_contract_store import FileScriptTaskContractStore
 from script_build_host.tools import LegacyScriptToolGateway, register_script_tools
-from script_build_host.tools.gateway import RetrievalAdapter
 
 
 @dataclass(frozen=True)
@@ -106,6 +110,7 @@ class HostDependencies:
     legacy_projection: Any = None
     legacy_api: Any = None
     http_command_journal: Any = None
+    mission_failures: Any = None
 
 
 @dataclass(frozen=True)
@@ -210,7 +215,9 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         receipt_store=getattr(dependencies.runner, "trace_store", None),
     )
     deterministic_worker = ScriptBuildDeterministicWorker(
-        candidate_service, bindings=dependencies.bindings
+        candidate_service,
+        bindings=dependencies.bindings,
+        task_store=dependencies.task_store,
     )
     coordinator = wire_orchestration(
         dependencies.runner,
@@ -230,6 +237,13 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         role_context_provider=workbench,
         deterministic_worker=deterministic_worker,
     )
+    deterministic_worker.evidence_orchestrator = DirectionEvidenceOrchestrator(
+        coordinator=coordinator,
+        bindings=dependencies.bindings,
+        input_snapshots=dependencies.input_snapshots,
+        artifacts=dependencies.business_artifacts,
+        retrieval_adapters=dependencies.retrieval_adapters,
+    )
     planning_service = PhaseTwoPlanningService(
         coordinator=coordinator,
         bindings=dependencies.bindings,
@@ -238,6 +252,15 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         closure_gate=boundary_verifier,
         limits=dependencies.phase_two_limits,
     )
+    planning_service.v2_service = TaskPlanningServiceV2(
+        coordinator=coordinator,
+        bindings=dependencies.bindings,
+        artifacts=dependencies.business_artifacts,
+        input_snapshots=dependencies.input_snapshots,
+    )
+    planning_service.evidence_gate = DirectionEvidenceGate(
+        dependencies.business_artifacts
+    )
     planning_service.workspace_lifecycle = candidate_service
     transition_gate = BuildTransitionGate()
     direction_reconciler = DirectionReconciler(
@@ -268,6 +291,11 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         phase_two_boundary_verifier=boundary_verifier,
         owner_lease_factory=dependencies.owner_lease_factory,
         fenced_command_gate=dependencies.fenced_command_gate,
+        failure_flow=(
+            MissionFailureFlow(dependencies.mission_failures)
+            if dependencies.mission_failures is not None
+            else None
+        ),
     )
     planning_service.dispatch_guard = mission_service
     deterministic_worker.dispatch_guard = mission_service
@@ -276,7 +304,6 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         snapshots=dependencies.input_snapshots,
         artifacts=dependencies.business_artifacts,
         coordinator=coordinator,
-        retrieval_adapters=dependencies.retrieval_adapters,
         dispatch_guard=mission_service,
         planner_tools=planning_service,
         candidate_tools=candidate_service,
@@ -342,6 +369,7 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
             legacy_projection=dependencies.legacy_projection,
             mission_service=mission_service,
             contract_reader=planning_service,
+            failure_repository=dependencies.mission_failures,
         )
     app = create_app(
         mission_service=mission_service,

script_build_host/src/script_build_host/production.py

@@ -55,6 +55,7 @@ from script_build_host.repositories import (
     SqlAlchemyInputSnapshotRepository,
     SqlAlchemyLegacyBuildStateRepository,
     SqlAlchemyMissionBindingRepository,
+    SqlAlchemyMissionFailureRepository,
     SqlAlchemyPublicationRepository,
     SqlAlchemyScriptBusinessArtifactRepository,
 )
@@ -265,6 +266,7 @@ def compose_production_host(
     publications = SqlAlchemyPublicationRepository(database.write)
     final_publications = SqlAlchemyPublicationRepository(database.final)
     legacy_state = SqlAlchemyLegacyBuildStateRepository(database.write)
+    mission_failures = SqlAlchemyMissionFailureRepository(database.write)
     legacy_input = LegacySqlAlchemyInputReader(
         database.read,
         uploaded_sessions=database.write,
@@ -360,6 +362,7 @@ def compose_production_host(
                 include_legacy_logs=not fresh_local_outputs,
             ),
             http_command_journal=HttpCommandJournal(database.write),
+            mission_failures=mission_failures,
         )
     )
     host = ProductionHost(

script_build_host/tests/test_context_broker_integration.py

@@ -14,6 +14,7 @@ from agent.orchestration import (
     RoleContextRequest,
     TaskStatus,
 )
+from agent.orchestration.prompt_budget import RolePromptEnvelopePacker
 
 from script_build_host.application.context_semantics import AdaptiveSemanticSelector
 from script_build_host.application.mission_workbench import (
@@ -762,41 +763,24 @@ async def test_root_validator_gets_full_closure_as_required_pages_when_large() -
             "evidence_refs": [],
         },
     )
-    state = await root_workbench.validator_state(request)
-
-    assert state.artifact["full_content_in_bootstrap"] is False
-    assert len(state.required_handles) == 3
-    assert state.context_bundle["sufficiency"]["status"] == "ambiguous"
-    assert (
-        root_workbench.pending_required_context(
-            root_trace_id="root", task_id="root-task", attempt_id="root-attempt"
-        )
-        == 3
-    )
-    for handle in state.required_handles:
-        cursor = None
-        while True:
-            page = await root_workbench.read_mission_context(
-                root_trace_id="root",
-                role="validator",
-                task_id="root-task",
-                attempt_id="root-attempt",
-                handle=handle,
-                cursor=cursor,
-            )
-            cursor = page["next_cursor"]
-            if cursor is None:
-                break
-    assert (
-        root_workbench.pending_required_context(
-            root_trace_id="root", task_id="root-task", attempt_id="root-attempt"
-        )
-        == 0
+    bootstrap = await root_workbench.build_validation_context(request)
+    packed = RolePromptEnvelopePacker(
+        token_limit=12_000, output_reserve=1_000
+    ).pack(
+        {"role_context": bootstrap.role_context},
+        system_prompt="validator",
+        tool_schemas=(),
     )
 
+    assert bootstrap.role_context["artifact"]["full_content_in_bootstrap"] is True
+    assert bootstrap.required_handles == ()
+    assert len(bootstrap.evidence_grants) == 3
+    assert packed.packing_mode in {"cards", "handles"}
+    assert len(packed.required_handles) == 3
+
 
 @pytest.mark.asyncio
-async def test_root_worker_can_expand_structured_script_and_validator_reads_survive_restart() -> (
+async def test_root_worker_can_expand_structured_script_without_validator_local_state() -> (
     None
 ):
     workbench, ledger, snapshots, _ = _fixture()
@@ -889,38 +873,9 @@ async def test_root_worker_can_expand_structured_script_and_validator_reads_surv
             "evidence_refs": [],
         },
     )
-    validator_state = await root_workbench.validator_state(validator_request)
-    restarted = MissionWorkbench(
-        task_store=workbench.broker._task_store,
-        bindings=_Bindings(),
-        snapshots=snapshots,
-        artifacts=workbench.broker._artifacts,
-        contracts=workbench.broker._contracts,
-        root_tools=RootTools(),
-        receipt_store=receipt_store,
-    )
-    with pytest.raises(WorkbenchError) as incomplete:
-        await restarted.verify_context_exhausted(
-            root_trace_id="root", task_id="root-task", attempt_id="root-attempt"
-        )
-    assert incomplete.value.code == "CONTEXT_NOT_EXHAUSTED"
-    for handle in validator_state.required_handles:
-        cursor = None
-        while True:
-            page = await root_workbench.read_mission_context(
-                root_trace_id="root",
-                role="validator",
-                task_id="root-task",
-                attempt_id="root-attempt",
-                handle=handle,
-                cursor=cursor,
-            )
-            cursor = page["next_cursor"]
-            if cursor is None:
-                break
-    await restarted.verify_context_exhausted(
-        root_trace_id="root", task_id="root-task", attempt_id="root-attempt"
-    )
+    bootstrap = await root_workbench.build_validation_context(validator_request)
+    assert len(bootstrap.evidence_grants) == 3
+    assert bootstrap.required_handles == ()
 
 
 @pytest.mark.asyncio

script_build_host/tests/test_goal_coverage.py

@@ -15,10 +15,6 @@ from script_build_host.application.phase_two_candidates import (
     PhaseTwoCandidateService,
     _require_paragraph_structures,
 )
-from script_build_host.application.phase_two_planning import (
-    PhasePolicyGuard,
-    PhaseTwoPlanningService,
-)
 from script_build_host.application.root_delivery import (
     RootDeliveryService,
     _closure_digest,
@@ -55,8 +51,14 @@ from script_build_host.domain.phase_two_artifacts import (
     StructuredScriptArtifactV1,
     hydrate_phase_two_artifact,
 )
+from script_build_host.domain.task_contract_compiler import (
+    PlanningContainer,
+    PlanningSnapshot,
+    TaskContractCompiler,
+)
 from script_build_host.domain.task_contracts import (
     AcceptedDecisionRef,
+    PlannerTaskIntentV2,
     ScriptCriterion,
     ScriptIntentClass,
     ScriptTaskBudget,
@@ -186,20 +188,6 @@ def test_goal_coverage_rejects_dangling_source_and_old_payloads() -> None:
     assert missing_goals.value.code == "GOAL_SCOPE_INVALID"
 
 
-def test_paragraph_first_is_allowed_but_structure_patch_remains_pinned() -> None:
-    paragraph = _contract(ScriptTaskKind.PARAGRAPH)
-    PhasePolicyGuard._phase_two_contracts((paragraph,))
-
-    patched = replace(
-        paragraph,
-        base_artifact_ref=ArtifactRef(
-            "script-build://artifact-versions/11", "structure", "11", _DIGEST
-        ),
-    )
-    with pytest.raises(TaskContractError, match="Paragraph patch requires"):
-        PhasePolicyGuard._phase_two_contracts((patched,))
-
-
 def test_duplicate_goal_ids_are_rejected_by_the_frozen_contract() -> None:
     with pytest.raises(TaskContractError) as caught:
         replace(_contract(ScriptTaskKind.STRUCTURE), goal_ids=("goal-1", "goal-1"))
@@ -229,47 +217,50 @@ def test_compose_derives_coverage_and_requires_structure_for_adopted_paragraph()
         )
 
 
-@pytest.mark.asyncio
-@pytest.mark.parametrize("goal_ids", [(), ("unknown",)])
-async def test_phase_two_goal_guard_rejects_invalid_scope_before_dispatch(
-    goal_ids: tuple[str, ...],
-) -> None:
-    direction = DirectionArtifact(
-        goals=(
-            DirectionGoal("goal-1", "Complete", "needed", None, ("complete",)),
-            DirectionGoal("goal-2", "Specific", "needed", "goal-1", ("specific",)),
-        ),
-        evidence_refs=("script-build://artifact-versions/90",),
-    )
-    service = PhaseTwoPlanningService(
-        coordinator=cast(Any, SimpleNamespace()),
-        bindings=cast(Any, SimpleNamespace()),
-        contracts=cast(Any, SimpleNamespace()),
-        artifacts=cast(Any, _DirectionArtifacts(direction)),
-    )
+def test_contract_compiler_owns_goal_scope_before_task_creation() -> None:
     direction_ref = AcceptedDecisionRef(
         "direction-accept",
         ArtifactRef("script-build://artifact-versions/10", "direction", "10", _DIGEST),
         "script-build://scopes/full",
         ScriptTaskKind.DIRECTION,
     )
-    contract = replace(
-        _contract(ScriptTaskKind.STRUCTURE),
-        goal_ids=goal_ids,
-        input_decision_refs=(direction_ref,),
+    snapshot = PlanningSnapshot(
+        "root",
+        2,
+        (PlanningContainer("root-task", None, None, None),),
+        (),
+        active_direction_ref=direction_ref,
+        direction_goal_ids=("goal-1", "goal-2"),
+    )
+    compiler = TaskContractCompiler()
+    compiled = compiler.compile_batch(
+        (
+            PlannerTaskIntentV2(
+                ScriptTaskKind.STRUCTURE,
+                "cover the active direction",
+            ),
+        ),
+        snapshot,
+        command_id="valid",
+        budget=ScriptTaskBudget(),
+    )
+    structure = next(
+        item for item in compiled if item.contract.task_kind is ScriptTaskKind.STRUCTURE
     )
+    assert structure.contract.goal_ids == ("goal-1", "goal-2")
 
-    with pytest.raises(GoalPolicyError) as caught:
-        await service._guard_phase_two_goals(
-            context={"phase": 2},
-            binding=SimpleNamespace(
-                active_direction_artifact_version_id=10,
-                script_build_id=7,
+    with pytest.raises(TaskContractError) as caught:
+        compiler.compile_batch(
+            (
+                PlannerTaskIntentV2(
+                    ScriptTaskKind.STRUCTURE,
+                    "invalid goal",
+                    target_goal_ids=("unknown",),
+                ),
             ),
-            root_trace_id="root",
-            ledger=SimpleNamespace(root_task_id="root"),
-            contracts=(contract,),
-            parent=SimpleNamespace(task_id="root"),
+            snapshot,
+            command_id="invalid",
+            budget=ScriptTaskBudget(),
         )
     assert caught.value.code == "GOAL_SCOPE_INVALID"
 
@@ -452,11 +443,3 @@ class _Artifacts:
 
     async def get_by_id(self, identifier: int, **__: Any) -> ArtifactVersion:
         return self.versions[identifier]
-
-
-class _DirectionArtifacts:
-    def __init__(self, direction: DirectionArtifact) -> None:
-        self.direction = direction
-
-    async def get_by_id(self, *_: Any, **__: Any) -> Any:
-        return SimpleNamespace(artifact=self.direction)

script_build_host/tests/test_mission_workbench.py

@@ -11,7 +11,6 @@ from agent.orchestration import AgentRole, ArtifactRef, RoleContextRequest, Task
 from script_build_host.application.mission_workbench import (
     MissionWorkbench,
     WorkbenchError,
-    _fit_planner_state,
 )
 from script_build_host.domain.task_contracts import (
     ScriptCriterion,
@@ -20,7 +19,7 @@ from script_build_host.domain.task_contracts import (
     ScriptTaskContractV1,
     ScriptTaskKind,
 )
-from script_build_host.domain.workbench import NotModified, PlannerState, semantic_handle
+from script_build_host.domain.workbench import NotModified, semantic_handle
 
 
 class _TaskStore:
@@ -125,37 +124,6 @@ class _Candidates:
         }
 
 
-def test_planner_state_compacts_optional_frontiers_instead_of_failing() -> None:
-    state = PlannerState(
-        state_revision="ledger:1",
-        phase=2,
-        root={"task_id": "root", "status": "waiting_children", "objective": "compose"},
-        focused_task_id="root",
-        active_subgraph=tuple(
-            {"task_id": f"task-{index}", "objective": "x" * 500} for index in range(12)
-        ),
-        accepted_frontier=tuple(
-            {"decision_id": f"decision-{index}", "summary": "y" * 1_200}
-            for index in range(10)
-        ),
-        goal_progress={"total": 1, "covered": 0, "uncovered_goal_ids": ["goal"]},
-        counts_by_status={"completed": 10},
-        active_operations=(),
-        latest_failures=(),
-        retired_task_count=20,
-        active_total=12,
-        accepted_total=10,
-        context_bundle={"cards": [{"summary": "z" * 1_000}]},
-        phase_protocol={"phase": 2},
-    )
-
-    compacted = _fit_planner_state(state)
-
-    assert compacted.accepted_has_more is True
-    assert len(compacted.accepted_frontier) < len(state.accepted_frontier)
-    assert len(json.dumps(compacted.to_payload(), ensure_ascii=False)) <= 24_000
-
-
 def _root_task(*, children: list[str] | None = None) -> Any:
     return SimpleNamespace(
         task_id="root-task",

script_build_host/tests/test_phase_two_agent_tools.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import asyncio
 import base64
 import hashlib
 from collections.abc import Mapping
@@ -9,14 +8,25 @@ from typing import Any
 
 import pytest
 from agent import ToolRegistry
-from agent.orchestration import ArtifactRef, ValidationVerdict
+from agent.orchestration import (
+    ArtifactRef,
+    ValidationEvidenceGrant,
+    ValidationVerdict,
+)
 from agent.orchestration.evidence import EvidenceResponse
-
-from script_build_host.domain.artifacts import ArtifactState
-from script_build_host.domain.errors import ArtifactNotFound, ProtocolViolation
-from script_build_host.domain.workbench import semantic_handle
+from agent.orchestration.models import ValidationEvidenceSnapshot
+from agent.orchestration.validation_evidence import (
+    evidence_handle,
+    extend_snapshot,
+    start_read_session,
+)
+
+from script_build_host.domain.errors import (
+    ProtocolViolation,
+    ValidationEvidenceUnauthorized,
+)
 from script_build_host.tools.contracts import AttemptManifest
-from script_build_host.tools.gateway import LegacyScriptToolGateway, RetrievalResult
+from script_build_host.tools.gateway import LegacyScriptToolGateway
 from script_build_host.tools.registry import register_script_tools
 
 
@@ -60,13 +70,23 @@ class _CandidateTools:
 
 
 class _TaskStore:
-    def __init__(self, task: Any) -> None:
+    def __init__(self, task: Any, ref: ArtifactRef) -> None:
         self.task = task
+        snapshot = extend_snapshot(
+            ValidationEvidenceSnapshot(),
+            (ValidationEvidenceGrant(ref, evidence_handle(ref)),),
+        )
+        self.validation = SimpleNamespace(
+            evidence_snapshot=snapshot,
+            read_session=start_read_session(snapshot, ()),
+        )
 
     async def load(self, _root_trace_id: str) -> Any:
         return SimpleNamespace(
             tasks={"task-1": self.task},
             attempts={"attempt-a": SimpleNamespace(attempt_id="attempt-a", task_id="task-1")},
+            validations={"validation-a": self.validation},
+            context_documents={},
         )
 
 
@@ -119,52 +139,6 @@ 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:
-            raise ArtifactNotFound()
-        return self.version
-
-    async def freeze(self, *, artifact: Any, **owners: Any) -> tuple[Any, ArtifactRef]:
-        ref = ArtifactRef(
-            "script-build://artifact-versions/8",
-            "evidence",
-            "8",
-            "sha256:" + "b" * 64,
-        )
-        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:
-        self.calls = 0
-
-    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",
-            supports=("opening", "opening"),
-        )
-
-
 class _Coordinator:
     def __init__(self, ref: ArtifactRef) -> None:
         criterion = SimpleNamespace(criterion_id="criterion-a", hard=True)
@@ -176,7 +150,7 @@ class _Coordinator:
                 context_refs=("script-build://task-kinds/paragraph",),
             ),
         )
-        self.task_store = _TaskStore(task)
+        self.task_store = _TaskStore(task, ref)
         self.artifact_store = _ArtifactStore(ref)
         self.attempt_submission: tuple[dict[str, Any], Any] | None = None
         self.validation_submission: tuple[dict[str, Any], tuple[Any, ...]] | None = None
@@ -289,85 +263,6 @@ async def test_attempt_submission_is_derived_from_protected_attempt_artifact() -
     assert budget.calls == [("root-a", "task-1", "submit")]
 
 
-@pytest.mark.asyncio
-async def test_retrieval_budget_is_checked_before_adapter_or_snapshot_access() -> None:
-    ref = ArtifactRef(
-        uri="script-build://artifact-versions/7",
-        kind="evidence",
-        version="7",
-        digest="sha256:" + "a" * 64,
-    )
-    gateway, _, _ = _gateway(ref)
-    budget = _BudgetGuard(ProtocolViolation("retrieval budget exhausted"))
-    gateway.budget_guard = budget
-    with pytest.raises(ProtocolViolation, match="budget exhausted"):
-        await gateway.retrieve(
-            "decode",
-            "search_script_decode_case",
-            {"query": "opening"},
-            {
-                "root_trace_id": "root-a",
-                "task_id": "task-1",
-                "attempt_id": "attempt-a",
-            },
-        )
-    assert budget.calls == [("root-a", "task-1", "retrieval")]
-
-
-@pytest.mark.asyncio
-async def test_retrieval_replay_reuses_frozen_evidence_and_can_submit() -> None:
-    artifacts = _RetrievalArtifacts()
-    adapter = _CountingRetrievalAdapter()
-
-    class Snapshots:
-        async def get(self, snapshot_id: str, *, script_build_id: int) -> Any:
-            assert (snapshot_id, script_build_id) == ("3", 9)
-            return SimpleNamespace(account={})
-
-    gateway = LegacyScriptToolGateway(
-        bindings=_Bindings(),  # type: ignore[arg-type]
-        snapshots=Snapshots(),  # type: ignore[arg-type]
-        artifacts=artifacts,  # type: ignore[arg-type]
-        coordinator=SimpleNamespace(),
-        retrieval_adapters={"decode": adapter},
-    )
-    context = {
-        "root_trace_id": "root-a",
-        "task_id": "task-1",
-        "attempt_id": "attempt-a",
-        "spec_version": 1,
-    }
-    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",)
-    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
-        )
-    assert adapter.calls == 1
-
-
 @pytest.mark.asyncio
 async def test_candidate_write_receives_identity_only_from_protected_context() -> None:
     ref = ArtifactRef(
@@ -431,7 +326,7 @@ async def test_structured_validation_keeps_excerpt_out_of_framework_summary() ->
                 "criterion_id": "criterion-a",
                 "scope_selector": {"anchor": "mission", "path": ["body", "end"]},
                 "observed_excerpt": excerpt,
-                "evidence_handle_ids": [semantic_handle("src", ref.uri, ref.digest)],
+                "evidence_handle_ids": [evidence_handle(ref)],
                 "severity": "hard",
                 "invalidated_inputs": ["decision-a"],
                 "recommended_action_class": "split",
@@ -488,7 +383,7 @@ async def test_passed_validation_rejects_blocking_defect_and_unfrozen_evidence()
                     "criterion_id": "criterion-a",
                     "verdict": "passed",
                     "reason": "ok",
-                    "evidence_handle_ids": [semantic_handle("src", ref.uri, ref.digest)],
+                    "evidence_handle_ids": [evidence_handle(ref)],
                 }
             ],
             defects=[defect],
@@ -498,7 +393,9 @@ async def test_passed_validation_rejects_blocking_defect_and_unfrozen_evidence()
 
     defect["severity"] = "warning"
     defect["evidence_handle_ids"] = ["src_" + "f" * 20]
-    with pytest.raises(ProtocolViolation, match="unknown evidence handle"):
+    with pytest.raises(
+        ValidationEvidenceUnauthorized, match="unknown evidence handle"
+    ):
         await gateway.submit_structured_validation(
             verdict="passed",
             criterion_results=[
@@ -506,7 +403,7 @@ async def test_passed_validation_rejects_blocking_defect_and_unfrozen_evidence()
                     "criterion_id": "criterion-a",
                     "verdict": "passed",
                     "reason": "ok",
-                    "evidence_handle_ids": [semantic_handle("src", ref.uri, ref.digest)],
+                    "evidence_handle_ids": [evidence_handle(ref)],
                 }
             ],
             defects=[defect],

script_build_host/tests/test_phase_two_candidates.py

@@ -145,6 +145,7 @@ def _contract(
 ) -> ScriptTaskContractV1:
     schemas = {
         ScriptTaskKind.PARAGRAPH: "paragraph-artifact/v1",
+        ScriptTaskKind.EVIDENCE_RETRIEVAL: "evidence-record/v1",
         ScriptTaskKind.DECODE_RETRIEVAL: "evidence-record/v1",
         ScriptTaskKind.COMPARE: "comparison-artifact/v1",
         ScriptTaskKind.COMPOSE: "structured-script/v1",
@@ -349,12 +350,12 @@ def _scope_fixture(
 
 @pytest.mark.asyncio
 async def test_phase_one_manifest_does_not_require_candidate_workspace() -> None:
-    contract = _contract(ScriptTaskKind.DECODE_RETRIEVAL)
+    contract = _contract(ScriptTaskKind.EVIDENCE_RETRIEVAL)
     (ledger, accepted), context = _scope_fixture(contract)
     evidence = EvidenceRecordV1(
         evidence_id="evidence",
-        source_type="decode",
-        tool_name="retrieve_decode",
+        source_type="knowledge",
+        tool_name="retrieval_pipeline",
         query={},
         source_refs=("source",),
         raw_artifact_ref=None,

script_build_host/tests/test_phase_two_inputs_validation.py

@@ -41,7 +41,7 @@ from script_build_host.domain.artifacts import (
     ArtifactVersion,
     EvidenceRecordV1,
 )
-from script_build_host.domain.input_compatibility import (
+from script_build_host.domain.legacy_input_compatibility import (
     AcceptedInputCompatibilityError,
     AcceptedInputSource,
     validate_accepted_input,
@@ -79,15 +79,15 @@ def test_candidate_portfolio_can_pin_the_active_direction_explicitly() -> None:
     )
 
 
-def test_explicit_creative_inputs_can_cross_sibling_scopes_in_one_branch() -> None:
-    validate_accepted_input(
-        consumer_kind=ScriptTaskKind.PARAGRAPH,
-        consumer_scope="script-build://scopes/direction/portfolio/compose/paragraph",
-        producer_kind=ScriptTaskKind.STRUCTURE,
-        producer_scope="script-build://scopes/direction/portfolio/compose/structure",
-        source=AcceptedInputSource.EXPLICIT,
-        same_parent=True,
-    )
+def test_legacy_input_rules_do_not_infer_branch_identity_from_siblings() -> None:
+    with pytest.raises(AcceptedInputCompatibilityError, match="outside the consumer scope"):
+        validate_accepted_input(
+            consumer_kind=ScriptTaskKind.PARAGRAPH,
+            consumer_scope="script-build://scopes/direction/portfolio/compose/paragraph",
+            producer_kind=ScriptTaskKind.STRUCTURE,
+            producer_scope="script-build://scopes/direction/portfolio/compose/structure",
+            source=AcceptedInputSource.EXPLICIT,
+        )
 
 
 def test_explicit_creative_inputs_cannot_cross_candidate_branches() -> None:
@@ -98,7 +98,6 @@ def test_explicit_creative_inputs_cannot_cross_candidate_branches() -> None:
             producer_kind=ScriptTaskKind.STRUCTURE,
             producer_scope="script-build://scopes/direction/portfolio/compose-b/structure",
             source=AcceptedInputSource.EXPLICIT,
-            same_parent=False,
         )
 
 
@@ -741,14 +740,14 @@ def test_four_validation_layers_defects_and_placeholder_preflight() -> None:
     assert validation_layer("compare") == "compare"
     assert validation_layer("compose") == "global"
     assert validation_layer("candidate-portfolio") == "governance"
-    assert (
-        ScriptBuildValidationPolicy()
-        .plan(
-            SimpleNamespace(task=_task("t", parent=None, kind="compose", status=TaskStatus.RUNNING))
+    with pytest.raises(ValueError, match="frozen executable contract"):
+        ScriptBuildValidationPolicy().plan(
+            SimpleNamespace(
+                task=_task(
+                    "t", parent=None, kind="compose", status=TaskStatus.RUNNING
+                )
+            )
         )
-        .validator_preset
-        == "script_candidate_validator"
-    )
 
     payload = {
         "defect_code": "REALIZATION_PLACEHOLDER",

script_build_host/tests/test_production_composition.py

@@ -122,7 +122,10 @@ async def test_production_expands_one_internal_model_to_all_script_roles(tmp_pat
     manifest = host.composition.mission_service.phase_two_model_manifest
     presets = manifest["presets"]
     assert isinstance(presets, dict)
-    assert len(presets) == 17
+    assert len(presets) == 14
+    assert "script_pattern_retrieval_worker" not in presets
+    assert "script_knowledge_retrieval_worker" not in presets
+    assert "script_external_retrieval_worker" not in presets
     assert presets["script_root_worker"]["model"] == "qwen3.7-max"
     assert presets["script_root_validator"]["max_iterations"] == 25
     await host.close()