合同：由语义选择器编译冻结任务合同

Planner 只提交 ScopeSelector、Decision ID、Goal 和业务标准；Host 解析同 Root ACCEPT Decision，生成完整 scope、Artifact、digest、write scope、预算与 closure。合同加入 compiler_manifest_digest、稳定任务标识和重复合同拦截，模型不再拼 URI 或机械闭包字段。

script_build_host/src/script_build_host/application/phase_two_planning.py

@@ -7,15 +7,18 @@ checks script-build invariants before calling Coordinator operations.
 
 from __future__ import annotations
 
+import hashlib
 import json
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, replace
 from datetime import UTC, datetime
 from typing import Any, Protocol, cast
+from uuid import NAMESPACE_URL, uuid4, uuid5
 
 from agent.orchestration import (
     DecisionAction,
     OperationStatus,
+    TaskConflict,
     TaskStatus,
     ValidationVerdict,
 )
@@ -45,6 +48,7 @@ from script_build_host.domain.task_contracts import (
     output_schema_for,
     task_kind_from_context_refs,
 )
+from script_build_host.domain.task_policy import TaskPolicyCatalog
 
 TASK_KIND_REF_PREFIX = "script-build://task-kinds/"
 TASK_CONTRACT_REF_PREFIX = "script-build://task-contracts/sha256/"
@@ -196,9 +200,12 @@ class PhasePolicyGuard:
         else:
             allowed = set()
         if not kinds <= allowed:
-            raise TaskContractError(
-                "PHASE_POLICY_VIOLATION",
-                f"phase {phase} cannot plan {sorted(item.value for item in kinds)} here",
+            self._raise_planning_violation(
+                phase=phase,
+                parent_is_root=parent_is_root,
+                parent_kind=parent_kind,
+                attempted=kinds,
+                allowed=allowed,
             )
         if phase == 1 and any(item.goal_ids for item in contracts):
             raise TaskContractError(
@@ -235,22 +242,56 @@ class PhasePolicyGuard:
             )
         elif phase == 2:
             allowed = (
-                {ScriptTaskKind.CANDIDATE_PORTFOLIO}
-                if parent_is_root
-                else set(_PHASE_TWO_KINDS)
+                {ScriptTaskKind.CANDIDATE_PORTFOLIO} if parent_is_root else set(_PHASE_TWO_KINDS)
             )
         else:
             allowed = set()
         if not kinds <= allowed:
-            raise TaskContractError(
-                "PHASE_POLICY_VIOLATION",
-                f"phase {phase} cannot plan {sorted(item.value for item in kinds)} here",
+            self._raise_planning_violation(
+                phase=phase,
+                parent_is_root=parent_is_root,
+                parent_kind=parent_kind,
+                attempted=kinds,
+                allowed=allowed,
             )
         if phase == 1 and any(item.goal_ids for item in inputs):
             raise TaskContractError(
                 "GOAL_SCOPE_INVALID", "Phase1 contracts must use an empty goal_ids array"
             )
 
+    @staticmethod
+    def _raise_planning_violation(
+        *,
+        phase: int,
+        parent_is_root: bool,
+        parent_kind: str | None,
+        attempted: set[ScriptTaskKind],
+        allowed: set[ScriptTaskKind],
+    ) -> None:
+        attempted_values = sorted(item.value for item in attempted)
+        allowed_values = sorted(item.value for item in allowed)
+        if phase == 2 and parent_is_root:
+            summary = (
+                "phase 2 Root must first plan exactly one candidate-portfolio; "
+                f"received {attempted_values}"
+            )
+        else:
+            summary = (
+                f"phase {phase} cannot plan {attempted_values} here; "
+                f"allowed task kinds are {allowed_values}"
+            )
+        raise TaskContractError(
+            "PHASE_POLICY_VIOLATION",
+            summary,
+            details={
+                "phase": phase,
+                "operation": "plan",
+                "parent_kind": "root" if parent_is_root else parent_kind,
+                "attempted_task_kinds": attempted_values,
+                "allowed_task_kinds": allowed_values,
+            },
+        )
+
     def dispatch(self, *, context: Mapping[str, Any], kind: ScriptTaskKind) -> None:
         phase = self.phase(context)
         if phase is None:
@@ -309,8 +350,7 @@ class PhasePolicyGuard:
                 or contract.compose_order
             )
             if (
-                contract.task_kind
-                in {ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO}
+                contract.task_kind in {ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO}
                 and adoption_started
                 and not contract.execution_ready
             ):
@@ -349,12 +389,8 @@ class PhasePolicyGuard:
                         "Paragraph patch requires the accepted Structure decision in "
                         "input_decision_refs",
                     )
-                if (
-                    contract.base_artifact_ref is not None
-                    and all(
-                        contract.base_artifact_ref != item.artifact_ref
-                        for item in structures
-                    )
+                if contract.base_artifact_ref is not None and all(
+                    contract.base_artifact_ref != item.artifact_ref for item in structures
                 ):
                     raise TaskContractError(
                         "INPUT_SCOPE_MISMATCH",
@@ -370,17 +406,13 @@ class PhasePolicyGuard:
                 if contract.execution_ready and len(directions) != 1:
                     raise TaskContractError(
                         "CHILD_DECISION_INVALID",
-                        "Compose requires exactly one accepted Direction in "
-                        "input_decision_refs",
+                        "Compose requires exactly one accepted Direction in input_decision_refs",
                     )
                 by_id = {
-                    item.decision_id: item
-                    for item in contract.candidate_closure_decision_refs
+                    item.decision_id: item for item in contract.candidate_closure_decision_refs
                 }
                 adopted = tuple(
-                    by_id[item]
-                    for item in contract.adopted_decision_ids
-                    if item in by_id
+                    by_id[item] for item in contract.adopted_decision_ids if item in by_id
                 )
                 if any(
                     item.expected_task_kind is ScriptTaskKind.PARAGRAPH for item in adopted
@@ -428,6 +460,7 @@ class PhaseTwoPlanningService:
         self.workspace_lifecycle = workspace_lifecycle
         self.limits = limits or PhaseTwoLimits()
         self.phase_policy = PhasePolicyGuard()
+        self.task_policy = TaskPolicyCatalog()
 
     async def plan_script_tasks(
         self,
@@ -463,6 +496,10 @@ class PhaseTwoPlanningService:
 
         # Resolve model-owned semantics before writing any content-addressed blob.
         planner_inputs = tuple(PlannerTaskInput.from_payload(value) for value in supplied)
+        planned_task_ids = tuple(
+            _planned_task_id(root_trace_id, _optional(context, "tool_call_id"), index)
+            for index in range(len(planner_inputs))
+        )
         self.phase_policy.planner_inputs(
             context=context,
             ledger=ledger,
@@ -473,11 +510,13 @@ class PhaseTwoPlanningService:
             [
                 await self._resolve_planner_input(
                     value,
+                    task_id=planned_task_ids[index],
+                    parent=parent,
                     binding=binding,
                     ledger=ledger,
                     root_trace_id=root_trace_id,
                 )
-                for value in planner_inputs
+                for index, value in enumerate(planner_inputs)
             ]
         )
         self.phase_policy.contracts(
@@ -507,6 +546,8 @@ class PhaseTwoPlanningService:
         planned = await self._freeze_all(
             root_trace_id, str(binding.input_snapshot_id), parsed_contracts
         )
+        for item, planned_task_id in zip(planned, planned_task_ids, strict=True):
+            item.draft["_task_id"] = planned_task_id
         result = await self.coordinator.create_tasks(
             root_trace_id,
             [item.draft for item in planned],
@@ -521,7 +562,10 @@ class PhaseTwoPlanningService:
             "contracts": [
                 {
                     "task_kind": item.frozen.contract.task_kind.value,
-                    "scope_ref": item.frozen.contract.scope_ref,
+                    "scope_selector": {
+                        "anchor": "mission",
+                        "path": list(_scope_path(item.frozen.contract.scope_ref)),
+                    },
                     "contract_ref": item.frozen.uri,
                     "contract_digest": item.frozen.digest,
                 }
@@ -623,9 +667,7 @@ class PhaseTwoPlanningService:
                         "TASK_CONTRACT_INVALID",
                         "container REVISE requires selected_decision_ids only",
                     )
-                self.phase_policy.revisions(
-                    context=context, contracts=(current.contract,)
-                )
+                self.phase_policy.revisions(context=context, contracts=(current.contract,))
                 contract = await self._adoption_contract(
                     binding=binding,
                     ledger=ledger,
@@ -654,6 +696,8 @@ class PhaseTwoPlanningService:
                 )
                 contract = await self._resolve_planner_input(
                     planner_input,
+                    task_id=task.task_id,
+                    parent=parent,
                     binding=binding,
                     ledger=ledger,
                     root_trace_id=root_trace_id,
@@ -663,9 +707,7 @@ class PhaseTwoPlanningService:
                 raise TaskContractError(
                     "TASK_KIND_PRESET_MISMATCH", "REVISE cannot change the Task kind"
                 )
-            await self._guard_phase_two_direction_scope(
-                context, root_trace_id, ledger, (contract,)
-            )
+            await self._guard_phase_two_direction_scope(context, root_trace_id, ledger, (contract,))
             parent = ledger.tasks.get(task.parent_task_id or "")
             await self._guard_phase_two_goals(
                 context=context,
@@ -689,6 +731,10 @@ class PhaseTwoPlanningService:
             planner_inputs = tuple(
                 PlannerTaskInput.from_payload(value) for value in child_contracts
             )
+            planned_task_ids = tuple(
+                _planned_task_id(root_trace_id, _optional(context, "tool_call_id"), index)
+                for index in range(len(planner_inputs))
+            )
             self.phase_policy.planner_inputs(
                 context=context,
                 ledger=ledger,
@@ -699,19 +745,19 @@ class PhaseTwoPlanningService:
                 [
                     await self._resolve_planner_input(
                         value,
+                        task_id=planned_task_ids[index],
+                        parent=task,
                         binding=binding,
                         ledger=ledger,
                         root_trace_id=root_trace_id,
                     )
-                    for value in planner_inputs
+                    for index, value in enumerate(planner_inputs)
                 ]
             )
             self.phase_policy.contracts(
                 context=context, ledger=ledger, parent=task, contracts=contracts
             )
-            await self._guard_phase_two_direction_scope(
-                context, root_trace_id, ledger, contracts
-            )
+            await self._guard_phase_two_direction_scope(context, root_trace_id, ledger, contracts)
             await self._guard_phase_two_goals(
                 context=context,
                 binding=binding,
@@ -727,12 +773,19 @@ class PhaseTwoPlanningService:
             planned_children = await self._freeze_all(
                 root_trace_id, str(binding.input_snapshot_id), contracts
             )
+            for item, planned_task_id in zip(planned_children, planned_task_ids, strict=True):
+                item.draft["_task_id"] = planned_task_id
             payload["tasks"] = [item.draft for item in planned_children]
         elif replacement_contract is not None or child_contracts or selected_decision_ids:
             raise TaskContractError(
                 "TASK_CONTRACT_INVALID", "this action does not accept Task contracts"
             )
 
+        # The current Validation is durable Task state, not a creative choice.
+        # Bind it here when the Planner omits the mechanical identifier.
+        if validation_id is None and task.status is TaskStatus.AWAITING_DECISION:
+            validation_id = task.validation_ids[-1] if task.validation_ids else None
+
         if decision_action is DecisionAction.BLOCK and is_root:
             self.phase_policy.boundary(context=context, reason=bounded_reason)
             if bounded_reason == PHASE_ONE_CAPABILITY_BOUNDARY:
@@ -756,17 +809,20 @@ class PhaseTwoPlanningService:
                     root_trace_id=root_trace_id,
                     input_snapshot_id=str(binding.input_snapshot_id),
                 )
-        result = cast(
-            dict[str, Any],
-            await self.coordinator.decide_task(
-                root_trace_id,
-                task_id,
-                validation_id,
-                decision_action,
-                payload,
-                _optional(context, "tool_call_id"),
-            ),
-        )
+        try:
+            result = cast(
+                dict[str, Any],
+                await self.coordinator.decide_task(
+                    root_trace_id,
+                    task_id,
+                    validation_id,
+                    decision_action,
+                    payload,
+                    _optional(context, "tool_call_id"),
+                ),
+            )
+        except TaskConflict as exc:
+            raise TaskContractError("TASK_STATE_CONFLICT", str(exc)) from exc
         if not is_root and decision_action in {
             DecisionAction.REPAIR,
             DecisionAction.RETRY,
@@ -832,6 +888,12 @@ class PhaseTwoPlanningService:
             frozen = await self.contract_for_task(root_trace_id, task)
             contract = frozen.contract
             self.phase_policy.dispatch(context=context, kind=contract.task_kind)
+            self._guard_direction_dispatch(
+                ledger,
+                task,
+                contract.task_kind,
+                phase=self.phase_policy.phase(context),
+            )
             parent = ledger.tasks.get(task.parent_task_id or "")
             await self._guard_phase_two_goals(
                 context=context,
@@ -929,23 +991,33 @@ class PhaseTwoPlanningService:
     ) -> dict[str, Any]:
         root_trace_id = _required(context, "root_trace_id")
         ledger = await self.coordinator.task_store.load(root_trace_id)
-        tasks: list[dict[str, Any]] = []
+        active_tasks: list[dict[str, Any]] = []
+        accepted_decisions: list[dict[str, Any]] = []
+        counts_by_status: dict[str, int] = {}
         for task in sorted(ledger.tasks.values(), key=lambda item: item.display_path):
             if task_id is not None and task.task_id != task_id:
                 continue
+            counts_by_status[task.status.value] = counts_by_status.get(task.status.value, 0) + 1
             if task.task_id == ledger.root_task_id:
                 kind_value = "root"
-                contract_ref = None
                 scope_ref = None
             else:
                 frozen = await self.contract_for_task(root_trace_id, task)
                 kind_value = frozen.contract.task_kind.value
-                contract_ref = frozen.uri
                 scope_ref = frozen.contract.scope_ref
-            accepted_decision_ref: dict[str, Any] | None = None
-            if task.task_id != ledger.root_task_id and task.decision_ids:
-                decision_id = task.decision_ids[-1]
-                decision = ledger.decisions.get(decision_id)
+            accepted_decision_id = next(
+                (
+                    decision_id
+                    for decision_id in reversed(task.decision_ids)
+                    if (
+                        ledger.decisions.get(decision_id) is not None
+                        and ledger.decisions[decision_id].action is DecisionAction.ACCEPT
+                    )
+                ),
+                None,
+            )
+            if task.task_id != ledger.root_task_id and accepted_decision_id is not None:
+                decision = ledger.decisions.get(accepted_decision_id)
                 attempt = (
                     ledger.attempts.get(decision.attempt_id)
                     if decision is not None and decision.attempt_id is not None
@@ -956,42 +1028,59 @@ class PhaseTwoPlanningService:
                     if attempt is not None and attempt.submission is not None
                     else []
                 )
-                if (
-                    decision is not None
-                    and decision.action is DecisionAction.ACCEPT
-                    and len(refs) == 1
-                ):
+                if decision is not None and len(refs) == 1:
                     ref = refs[0]
-                    accepted_decision_ref = {
-                        "decision_id": decision_id,
-                        "artifact_ref": {
-                            "uri": ref.uri,
-                            "kind": ref.kind,
-                            "version": ref.version,
-                            "digest": ref.digest,
-                            "summary": ref.summary,
-                            "metadata": ref.metadata,
-                        },
+                    accepted_decisions.append(
+                        {
+                            "decision_id": accepted_decision_id,
+                            "task_id": task.task_id,
+                            "task_kind": kind_value,
+                            "artifact_kind": ref.kind,
+                            "artifact_ref": ref.uri,
+                            "scope_ref": scope_ref,
+                        }
+                    )
+            if task.task_id == ledger.root_task_id or task.status not in {
+                TaskStatus.COMPLETED,
+                TaskStatus.SUPERSEDED,
+                TaskStatus.CANCELLED,
+            }:
+                latest_attempt = (
+                    ledger.attempts.get(task.attempt_ids[-1]) if task.attempt_ids else None
+                )
+                failure = (
+                    latest_attempt.failure.to_dict()
+                    if latest_attempt is not None and latest_attempt.failure is not None
+                    else None
+                )
+                active_tasks.append(
+                    {
+                        "task_id": task.task_id,
+                        "parent_task_id": task.parent_task_id,
+                        "display_path": task.display_path,
+                        "status": task.status.value,
+                        "task_kind": kind_value,
                         "scope_ref": scope_ref,
-                        "expected_task_kind": kind_value,
+                        "objective": task.current_spec.objective,
+                        "attempt_count": len(task.attempt_ids),
+                        "blocked_reason": task.blocked_reason,
+                        "latest_failure": failure,
                     }
-            tasks.append(
-                {
-                    "task_id": task.task_id,
-                    "parent_task_id": task.parent_task_id,
-                    "display_path": task.display_path,
-                    "status": task.status.value,
-                    "task_kind": kind_value,
-                    "scope_ref": scope_ref,
-                    "contract_ref": contract_ref,
-                    "current_spec_version": task.current_spec_version,
-                    "attempt_count": len(task.attempt_ids),
-                    "child_task_ids": list(task.child_task_ids),
-                    "blocked_reason": task.blocked_reason,
-                    "accepted_decision_ref": accepted_decision_ref,
-                }
-            )
-        return {"root_trace_id": root_trace_id, "revision": ledger.revision, "tasks": tasks}
+                )
+        root = ledger.tasks[ledger.root_task_id]
+        return {
+            "root_trace_id": root_trace_id,
+            "revision": ledger.revision,
+            "focused_task_id": ledger.focused_task_id,
+            "root": {
+                "task_id": root.task_id,
+                "status": root.status.value,
+                "blocked_reason": root.blocked_reason,
+            },
+            "active_tasks": active_tasks,
+            "accepted_decisions": accepted_decisions,
+            "counts_by_status": counts_by_status,
+        }
 
     async def contract_for_task(self, root_trace_id: str, task: Any) -> FrozenTaskContract:
         refs = [
@@ -1030,6 +1119,8 @@ class PhaseTwoPlanningService:
         self,
         value: PlannerTaskInput,
         *,
+        task_id: str,
+        parent: Any,
         binding: Any,
         ledger: Any,
         root_trace_id: str,
@@ -1046,6 +1137,7 @@ class PhaseTwoPlanningService:
             )
             for decision_id in decision_ids
         ]
+        direction_constraints: Sequence[Any] = ()
         if value.task_kind in _DIRECTION_INPUT_KINDS:
             direction = await self._active_direction_decision_ref(
                 binding=binding,
@@ -1053,11 +1145,18 @@ class PhaseTwoPlanningService:
                 root_trace_id=root_trace_id,
             )
             refs = [
-                item
-                for item in refs
-                if item.expected_task_kind is not ScriptTaskKind.DIRECTION
+                item for item in refs if item.expected_task_kind is not ScriptTaskKind.DIRECTION
             ]
             refs.insert(0, direction)
+            direction_version = await self.artifacts.get_by_id(
+                binding.active_direction_artifact_version_id,
+                script_build_id=binding.script_build_id,
+            )
+            if not isinstance(direction_version.artifact, DirectionArtifact):
+                raise TaskContractError(
+                    "GOAL_SCOPE_INVALID", "active Direction artifact has the wrong type"
+                )
+            direction_constraints = direction_version.artifact.constraints
 
         comparison_refs = tuple(
             [
@@ -1070,26 +1169,40 @@ class PhaseTwoPlanningService:
             ]
         )
         base_ref = next(
-            (
-                item.artifact_ref
-                for item in refs
-                if item.decision_id == value.base_decision_id
-            ),
+            (item.artifact_ref for item in refs if item.decision_id == value.base_decision_id),
             None,
         )
+        scope_ref = await self._resolve_scope_selector(
+            root_trace_id=root_trace_id,
+            parent=parent,
+            selector=value.scope_selector,
+        )
+        criteria = self.task_policy.criteria(
+            task_id=task_id,
+            task_kind=value.task_kind,
+            business=value.criteria,
+            direction_constraints=direction_constraints,
+        )
+        manifest_digest = _compiler_manifest_digest(
+            policy_digest=self.task_policy.manifest_digest,
+            task_kind=value.task_kind,
+            scope_ref=scope_ref,
+            decision_ids=tuple(item.decision_id for item in refs),
+            comparison_ids=tuple(item.decision_id for item in comparison_refs),
+        )
         return ScriptTaskContractV1(
             task_kind=value.task_kind,
-            scope_ref=value.scope_ref,
+            scope_ref=scope_ref,
             intent_class=value.intent_class,
             objective=value.objective,
             input_decision_refs=tuple(refs),
             base_artifact_ref=base_ref,
             write_scope=_write_scope_for(value.task_kind),
-            gap_ref=value.gap_ref,
+            gap_ref=(f"script-build://gaps/{value.gap_key}" if value.gap_key else None),
             output_schema=output_schema_for(
                 value.task_kind, patched=value.base_decision_id is not None
             ),
-            criteria=value.criteria,
+            criteria=criteria,
             budget=ScriptTaskBudget(
                 max_attempts=self.limits.max_attempts_per_task,
                 max_tokens=self.limits.max_task_tokens,
@@ -1100,8 +1213,26 @@ class PhaseTwoPlanningService:
             goal_ids=value.goal_ids,
             supersedes_decision_ids=value.supersedes_decision_ids,
             comparison_decision_refs=comparison_refs,
+            compiler_manifest_digest=manifest_digest,
         )
 
+    async def _resolve_scope_selector(
+        self, *, root_trace_id: str, parent: Any, selector: Any
+    ) -> str:
+        if selector.anchor.value == "mission":
+            base: tuple[str, ...] = ()
+        elif parent.task_id == (await self.coordinator.task_store.load(root_trace_id)).root_task_id:
+            base = ()
+        else:
+            frozen = await self.contract_for_task(root_trace_id, parent)
+            base = _scope_path(frozen.contract.scope_ref)
+        path = (*base, *selector.path)
+        if not path:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "scope_selector resolves to an empty mission scope"
+            )
+        return "script-build://scopes/" + "/".join(path)
+
     async def _resolve_decision_ref(
         self,
         *,
@@ -1568,9 +1699,7 @@ class PhaseTwoPlanningService:
                         if decision is not None and decision.attempt_id is not None
                         else None
                     )
-                    producer = (
-                        ledger.tasks.get(attempt.task_id) if attempt is not None else None
-                    )
+                    producer = ledger.tasks.get(attempt.task_id) if attempt is not None else None
                     if producer is None:
                         raise TaskContractError(
                             "GOAL_SCOPE_INVALID", "Compare references an unknown candidate Task"
@@ -1644,6 +1773,14 @@ class PhaseTwoPlanningService:
                     raise TaskContractError(
                         "INPUT_SCOPE_MISMATCH",
                         str(exc),
+                        details={
+                            "decision_id": imported_ref.decision_id,
+                            "consumer_kind": consumer.task_kind.value,
+                            "consumer_scope": consumer.scope_ref,
+                            "producer_kind": producer.task_kind.value,
+                            "producer_scope": producer.scope_ref,
+                            "source": AcceptedInputSource.EXPLICIT.value,
+                        },
                     ) from exc
 
     async def _guard_runtime_budget(
@@ -1767,12 +1904,15 @@ class PhaseTwoPlanningService:
         if action is DecisionAction.REVISE and replacement_contract is not None:
             replacement = PlannerTaskInput.from_payload(replacement_contract)
             old_contract = current_contract.contract
+            old_path = tuple(
+                item
+                for item in old_contract.scope_ref.removeprefix("script-build://scopes/").split("/")
+                if item
+            )
             if (
                 replacement.intent_class != old_contract.intent_class
-                or (
-                    _scope_contains(old_contract.scope_ref, replacement.scope_ref)
-                    and replacement.scope_ref != old_contract.scope_ref
-                )
+                or replacement.scope_selector.anchor.value != "mission"
+                or replacement.scope_selector.path != old_path
             ):
                 return
         raise TaskContractError(
@@ -1827,11 +1967,24 @@ class PhaseTwoPlanningService:
                     )
             return
         parent_contract = (await self.contract_for_task(root_trace_id, parent)).contract
-        if any(
-            not _scope_contains(parent_contract.scope_ref, item.scope_ref) for item in contracts
-        ):
+        outside = next(
+            (
+                item
+                for item in contracts
+                if not _scope_contains(parent_contract.scope_ref, item.scope_ref)
+            ),
+            None,
+        )
+        if outside is not None:
             raise TaskContractError(
-                "INPUT_SCOPE_MISMATCH", "child Task scope must stay within its parent scope"
+                "INPUT_SCOPE_MISMATCH",
+                "child Task scope must stay within its parent scope",
+                details={
+                    "parent_kind": parent_contract.task_kind.value,
+                    "parent_scope": parent_contract.scope_ref,
+                    "child_kind": outside.task_kind.value,
+                    "child_scope": outside.scope_ref,
+                },
             )
         if parent_contract.write_scope and any(
             any(
@@ -1914,6 +2067,66 @@ class PhaseTwoPlanningService:
                 "a completed consumer requires a new replacement Task for new Evidence",
             )
 
+    @staticmethod
+    def _guard_direction_dispatch(
+        ledger: Any,
+        task: Any,
+        kind: ScriptTaskKind,
+        *,
+        phase: int | None,
+    ) -> None:
+        if phase != 1 or kind is not ScriptTaskKind.DIRECTION:
+            return
+        for child_id in task.child_task_ids:
+            child = ledger.tasks.get(child_id)
+            if child is None or child.status is not TaskStatus.COMPLETED:
+                continue
+            child_kind = task_kind_from_context_refs(child.current_spec.context_refs)
+            if child_kind not in RETRIEVAL_KINDS or not child.decision_ids:
+                continue
+            decision = ledger.decisions.get(child.decision_ids[-1])
+            if decision is not None and decision.action is DecisionAction.ACCEPT:
+                return
+        raise TaskContractError(
+            "CHILD_DECISION_INVALID",
+            "Phase1 Direction requires at least one direct-child Retrieval ACCEPT before dispatch",
+        )
+
+
+def _planned_task_id(root_trace_id: str, tool_call_id: str | None, index: int) -> str:
+    if tool_call_id:
+        return str(uuid5(NAMESPACE_URL, f"{root_trace_id}:{tool_call_id}:{index}"))
+    return str(uuid4())
+
+
+def _scope_path(scope_ref: str) -> tuple[str, ...]:
+    prefix = "script-build://scopes/"
+    if not scope_ref.startswith(prefix):
+        raise TaskContractError("TASK_CONTRACT_INVALID", "parent scope is not canonical")
+    return tuple(item for item in scope_ref.removeprefix(prefix).split("/") if item)
+
+
+def _compiler_manifest_digest(
+    *,
+    policy_digest: str,
+    task_kind: ScriptTaskKind,
+    scope_ref: str,
+    decision_ids: tuple[str, ...],
+    comparison_ids: tuple[str, ...],
+) -> str:
+    encoded = json.dumps(
+        {
+            "policy_digest": policy_digest,
+            "task_kind": task_kind.value,
+            "scope_ref": scope_ref,
+            "decision_ids": decision_ids,
+            "comparison_ids": comparison_ids,
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode()
+    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+
 
 def _required(context: Mapping[str, Any], key: str) -> str:
     value = context.get(key)
@@ -1965,6 +2178,8 @@ 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",)
@@ -2068,18 +2283,12 @@ def _decision_attempt_id(ledger: Any, task: Any, validation_id: str | None) -> s
 def _root_replacement(task: Any, value: Mapping[str, Any]) -> dict[str, Any]:
     """Allow a safe Root objective refinement without inventing a business contract."""
 
-    if set(value) != {"objective", "acceptance_criteria", "context_refs"}:
+    if set(value) != {"objective"}:
         raise TaskContractError(
             "TASK_CONTRACT_INVALID",
-            "Root replacement must contain objective, acceptance_criteria and context_refs",
+            "Root replacement may contain only the semantic objective",
         )
     objective = _bounded(str(value["objective"]), 2_000, "Root objective")
-    criteria = value["acceptance_criteria"]
-    refs = value["context_refs"]
-    if not isinstance(criteria, Sequence) or isinstance(criteria, (str, bytes)):
-        raise TaskContractError("TASK_CONTRACT_INVALID", "Root criteria must be an array")
-    if not isinstance(refs, Sequence) or isinstance(refs, (str, bytes)):
-        raise TaskContractError("TASK_CONTRACT_INVALID", "Root context_refs must be an array")
     expected_criteria = [
         {
             "criterion_id": item.criterion_id,
@@ -2089,11 +2298,6 @@ def _root_replacement(task: Any, value: Mapping[str, Any]) -> dict[str, Any]:
         for item in task.current_spec.acceptance_criteria
     ]
     expected_refs = list(task.current_spec.context_refs)
-    if list(criteria) != expected_criteria or list(refs) != expected_refs:
-        raise TaskContractError(
-            "TASK_CONTRACT_INVALID",
-            "Root REVISE may refine only objective and cannot expand its contract",
-        )
     return {
         "objective": objective,
         "acceptance_criteria": expected_criteria,

script_build_host/src/script_build_host/domain/phase_two_ports.py

@@ -57,6 +57,17 @@ class CandidateWorkspaceRepository(Protocol):
         self, context: CandidateWriteContext, paragraphs: Sequence[Mapping[str, Any]]
     ) -> Mapping[str, int]: ...
 
+    async def save_paragraphs(
+        self, context: CandidateWriteContext, paragraphs: Sequence[Mapping[str, Any]]
+    ) -> tuple[Mapping[str, int], tuple[int, ...]]: ...
+
+    async def create_elements(
+        self,
+        context: CandidateWriteContext,
+        elements: Sequence[Mapping[str, Any]],
+        links: Sequence[Mapping[str, Any]],
+    ) -> tuple[Mapping[str, int], int]: ...
+
     async def freeze(
         self,
         context: CandidateWriteContext,

script_build_host/src/script_build_host/domain/task_contracts.py

@@ -7,6 +7,7 @@ import re
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 from enum import StrEnum
+from hashlib import sha256
 from typing import Any
 
 from agent.orchestration import ArtifactRef
@@ -38,8 +39,15 @@ MAX_REFERENCES = 64
 
 
 class TaskContractError(ScriptBuildError):
-    def __init__(self, code: str, summary: str) -> None:
+    def __init__(
+        self,
+        code: str,
+        summary: str,
+        *,
+        details: Mapping[str, Any] | None = None,
+    ) -> None:
         super().__init__(code, summary)
+        self.details = dict(details or {})
 
 
 class ScriptTaskKind(StrEnum):
@@ -68,6 +76,70 @@ class ScriptIntentClass(StrEnum):
     DELIVER = "deliver"
 
 
+class ScopeAnchor(StrEnum):
+    MISSION = "mission"
+    PARENT = "parent"
+
+
+@dataclass(frozen=True, slots=True)
+class ScopeSelector:
+    anchor: ScopeAnchor
+    path: tuple[str, ...]
+
+    def __post_init__(self) -> None:
+        if len(self.path) > 4 or (self.anchor is ScopeAnchor.MISSION and not self.path):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "scope_selector path must contain 1-4 mission segments"
+            )
+        for segment in self.path:
+            _require_identifier(segment, "scope_selector path segment")
+
+    @classmethod
+    def from_payload(cls, value: object) -> ScopeSelector:
+        if not isinstance(value, Mapping):
+            raise TaskContractError("TASK_CONTRACT_INVALID", "scope_selector must be an object")
+        unknown = set(value) - {"anchor", "path"}
+        if unknown:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID",
+                f"scope_selector contains unknown fields: {sorted(unknown)}",
+            )
+        return cls(
+            anchor=_enum_value(ScopeAnchor, value.get("anchor"), "scope_selector.anchor"),
+            path=_strings(value.get("path"), "scope_selector.path"),
+        )
+
+    def to_payload(self) -> dict[str, Any]:
+        return {"anchor": self.anchor.value, "path": list(self.path)}
+
+
+@dataclass(frozen=True, slots=True)
+class PlannerCriterionInput:
+    client_key: str
+    description: str
+    hard: bool = True
+
+    def __post_init__(self) -> None:
+        _require_identifier(self.client_key, "criterion client_key")
+        _require_bounded_text(self.description, "criterion description", 500)
+        if not isinstance(self.hard, bool):
+            raise TaskContractError("TASK_CONTRACT_INVALID", "criterion hard must be boolean")
+
+    @classmethod
+    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"}:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "Planner criteria cannot provide criterion_id"
+            )
+        return cls(
+            client_key=str(value.get("client_key", "")),
+            description=str(value.get("description", "")),
+            hard=value.get("hard", True),
+        )
+
+
 _OUTPUT_SCHEMAS: dict[ScriptTaskKind, frozenset[str]] = {
     ScriptTaskKind.DIRECTION: frozenset({"script-direction/v1"}),
     ScriptTaskKind.PATTERN_RETRIEVAL: frozenset({"evidence-record/v1"}),
@@ -204,30 +276,27 @@ class PlannerTaskInput:
     """Planner-owned semantics; the Host derives every immutable reference."""
 
     task_kind: ScriptTaskKind
-    scope_ref: str
+    scope_selector: ScopeSelector
     intent_class: ScriptIntentClass
     objective: str
     goal_ids: tuple[str, ...]
-    criteria: tuple[ScriptCriterion, ...]
+    criteria: tuple[PlannerCriterionInput, ...] = ()
     input_decision_ids: tuple[str, ...] = ()
     base_decision_id: str | None = None
     comparison_decision_ids: tuple[str, ...] = ()
     supersedes_decision_ids: tuple[str, ...] = ()
-    gap_ref: str | None = None
+    gap_key: str | None = None
 
     def __post_init__(self) -> None:
         if self.task_kind is ScriptTaskKind.ROOT_DELIVERY:
-            raise TaskContractError(
-                "TASK_CONTRACT_INVALID", "Root Delivery is frozen by the Host"
-            )
-        _require_control_uri(self.scope_ref, "scope_ref")
+            raise TaskContractError("TASK_CONTRACT_INVALID", "Root Delivery is frozen by the Host")
         _require_bounded_text(self.objective, "objective", MAX_OBJECTIVE_CHARS)
-        if not self.criteria or len(self.criteria) > MAX_CRITERIA:
+        if len(self.criteria) > MAX_CRITERIA:
             raise TaskContractError(
                 "TASK_CONTRACT_INVALID",
-                f"criteria must contain between 1 and {MAX_CRITERIA} items",
+                f"criteria may contain at most {MAX_CRITERIA} business items",
             )
-        _require_unique((item.criterion_id for item in self.criteria), "criterion IDs")
+        _require_unique((item.client_key for item in self.criteria), "criterion client keys")
         if len(self.goal_ids) > MAX_DIRECTION_GOALS:
             raise TaskContractError(
                 "GOAL_SCOPE_INVALID",
@@ -250,8 +319,8 @@ class PlannerTaskInput:
             _require_identifier(value, "decision_id")
         if self.base_decision_id is not None:
             _require_identifier(self.base_decision_id, "base_decision_id")
-        if self.gap_ref is not None:
-            _require_control_uri(self.gap_ref, "gap_ref")
+        if self.gap_key is not None:
+            _require_identifier(self.gap_key, "gap_key")
         if self.task_kind is ScriptTaskKind.COMPARE and len(self.comparison_decision_ids) < 2:
             raise TaskContractError(
                 "TASK_CONTRACT_INVALID", "Compare requires at least two Decision IDs"
@@ -265,7 +334,7 @@ class PlannerTaskInput:
     def from_payload(cls, value: Mapping[str, Any]) -> PlannerTaskInput:
         allowed = {
             "task_kind",
-            "scope_ref",
+            "scope_selector",
             "intent_class",
             "objective",
             "goal_ids",
@@ -274,7 +343,7 @@ class PlannerTaskInput:
             "base_decision_id",
             "comparison_decision_ids",
             "supersedes_decision_ids",
-            "gap_ref",
+            "gap_key",
         }
         unknown = sorted(set(value) - allowed)
         if unknown:
@@ -284,16 +353,15 @@ class PlannerTaskInput:
             )
         return cls(
             task_kind=_enum_value(ScriptTaskKind, value.get("task_kind"), "task_kind"),
-            scope_ref=str(value.get("scope_ref", "")),
-            intent_class=_enum_value(
-                ScriptIntentClass, value.get("intent_class"), "intent_class"
-            ),
+            scope_selector=ScopeSelector.from_payload(value.get("scope_selector")),
+            intent_class=_enum_value(ScriptIntentClass, value.get("intent_class"), "intent_class"),
             objective=str(value.get("objective", "")),
             goal_ids=_strings(value.get("goal_ids"), "goal_ids"),
-            criteria=_criteria(value.get("criteria")),
-            input_decision_ids=_strings(
-                value.get("input_decision_ids"), "input_decision_ids"
+            criteria=tuple(
+                PlannerCriterionInput.from_payload(item)
+                for item in _objects(value.get("criteria"), "criteria")
             ),
+            input_decision_ids=_strings(value.get("input_decision_ids"), "input_decision_ids"),
             base_decision_id=(
                 str(value["base_decision_id"])
                 if value.get("base_decision_id") is not None
@@ -305,7 +373,7 @@ class PlannerTaskInput:
             supersedes_decision_ids=_strings(
                 value.get("supersedes_decision_ids"), "supersedes_decision_ids"
             ),
-            gap_ref=str(value["gap_ref"]) if value.get("gap_ref") is not None else None,
+            gap_key=str(value["gap_key"]) if value.get("gap_key") is not None else None,
         )
 
 
@@ -368,6 +436,7 @@ class ScriptTaskContractV1:
     held_or_rejected_decision_ids: tuple[str, ...] = ()
     compose_order: tuple[str, ...] = ()
     comparison_decision_refs: tuple[AcceptedDecisionRef, ...] = ()
+    compiler_manifest_digest: str = "sha256:" + "0" * 64
 
     def __post_init__(self) -> None:
         _require_control_uri(self.scope_ref, "scope_ref")
@@ -398,6 +467,7 @@ class ScriptTaskContractV1:
             _require_control_uri(self.gap_ref, "gap_ref")
         if self.base_artifact_ref is not None:
             _validate_artifact_ref(self.base_artifact_ref)
+        Sha256Digest.from_wire(self.compiler_manifest_digest)
         for label, values in (
             ("input decision IDs", (item.decision_id for item in self.input_decision_refs)),
             ("supersedes decision IDs", self.supersedes_decision_ids),
@@ -549,6 +619,7 @@ class ScriptTaskContractV1:
             "comparison_decision_refs": [
                 item.to_payload() for item in self.comparison_decision_refs
             ],
+            "compiler_manifest_digest": self.compiler_manifest_digest,
         }
 
     @classmethod
@@ -567,6 +638,8 @@ class ScriptTaskContractV1:
             raise TaskContractError("TASK_CONTRACT_INVALID", "task budget must be an object")
         if "goal_ids" not in value:
             raise TaskContractError("GOAL_SCOPE_INVALID", "goal_ids is required")
+        if "compiler_manifest_digest" not in value:
+            raise TaskContractError("TASK_CONTRACT_INVALID", "compiler_manifest_digest is required")
         return cls(
             task_kind=_enum_value(ScriptTaskKind, value.get("task_kind"), "task_kind"),
             scope_ref=str(value.get("scope_ref", "")),
@@ -595,9 +668,18 @@ class ScriptTaskContractV1:
             ),
             compose_order=_strings(value.get("compose_order"), "compose_order"),
             comparison_decision_refs=_decision_refs(value.get("comparison_decision_refs")),
+            compiler_manifest_digest=str(value.get("compiler_manifest_digest", "")),
         )
 
 
+def stable_semantic_id(task_id: str, kind: str, client_key: str) -> str:
+    """Generate an order-independent ID for model-owned semantic content."""
+
+    _require_identifier(client_key, f"{kind} client_key")
+    digest = sha256(f"{task_id}\0{kind}\0{client_key.strip().lower()}".encode()).hexdigest()
+    return f"{kind}_{digest[:16]}"
+
+
 @dataclass(frozen=True, slots=True)
 class FrozenTaskContract:
     uri: str
@@ -719,6 +801,14 @@ def _strings(value: Any, label: str) -> tuple[str, ...]:
     return tuple(value)
 
 
+def _objects(value: Any, label: str) -> tuple[Mapping[str, Any], ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, list) or any(not isinstance(item, Mapping) for item in value):
+        raise TaskContractError("TASK_CONTRACT_INVALID", f"{label} must be an object array")
+    return tuple(value)
+
+
 def _decision_refs(value: Any) -> tuple[AcceptedDecisionRef, ...]:
     if value is None:
         return ()
@@ -766,7 +856,10 @@ __all__ = [
     "AcceptedInputBundleV1",
     "FrozenTaskContract",
     "PhaseTwoLimits",
+    "PlannerCriterionInput",
     "PlannerTaskInput",
+    "ScopeAnchor",
+    "ScopeSelector",
     "ScriptCriterion",
     "ScriptIntentClass",
     "ScriptTaskBudget",
@@ -774,5 +867,6 @@ __all__ = [
     "ScriptTaskKind",
     "TaskContractError",
     "output_schema_for",
+    "stable_semantic_id",
     "task_kind_from_context_refs",
 ]

script_build_host/tests/test_phase_one_e2e.py

@@ -79,15 +79,18 @@ def _contract(
 ) -> dict[str, object]:
     return {
         "task_kind": kind,
-        "scope_ref": scope,
+        "scope_selector": {
+            "anchor": "mission",
+            "path": scope.removeprefix("script-build://scopes/").split("/")[:4],
+        },
         "intent_class": "explore",
         "objective": objective,
         "input_decision_ids": [],
         "base_decision_id": None,
-        "gap_ref": None,
+        "gap_key": None,
         "criteria": [
             {
-                "criterion_id": criterion_id,
+                "client_key": criterion_id,
                 "description": description,
                 "hard": True,
             }
@@ -244,15 +247,6 @@ class _ScriptedLLM:
                     "reason": "record accepted direction in root objective",
                     "replacement_contract": {
                         "objective": root.current_spec.objective + " using accepted direction",
-                        "acceptance_criteria": [
-                            {
-                                "criterion_id": item.criterion_id,
-                                "description": item.description,
-                                "hard": item.hard,
-                            }
-                            for item in root.current_spec.acceptance_criteria
-                        ],
-                        "context_refs": list(root.current_spec.context_refs),
                     },
                 },
             )
@@ -281,8 +275,7 @@ class _ScriptedLLM:
                 {},
             )
         if last is None or "artifact_ref" not in last:
-            accepted = prompt["accepted_child_results"]
-            evidence_uri = accepted[0]["submission"]["evidence_refs"][0]["uri"]
+            accepted = prompt["role_context"]["accepted_inputs"]
             return self.call(
                 "save_direction_candidate",
                 {
@@ -296,13 +289,14 @@ class _ScriptedLLM:
                     ],
                     "constraints": [
                         {
-                            "constraint_id": "c1",
+                            "client_key": "c1",
                             "statement": "remain grounded",
                             "rationale": "accepted evidence is mandatory",
                         }
                     ],
                     "preferences": [],
-                    "evidence_refs": [evidence_uri],
+                    "strategy_handles": [],
+                    "evidence_decision_ids": [accepted[0]["decision_id"]],
                 },
             )
         return self.call(
@@ -318,21 +312,14 @@ class _ScriptedLLM:
             if "script-build://task-kinds/direction" in spec["context_refs"]
             else "evidence"
         )
-        evidence = _last_tool_json(messages)
-        if evidence is None or "items" not in evidence:
-            return self.call(
-                "query_validation_evidence",
-                {"query": kind, "limit": 5},
-            )
-        assert evidence["items"], "validator must read the frozen artifact body"
-        artifact = evidence["items"][0]["artifact"]
+        role_context = prompt["role_context"]
+        assert role_context["artifact"]["items"], "validator must receive the frozen artifact body"
+        artifact = role_context["artifact"]["items"][0]["content"]
         assert isinstance(artifact, dict)
         self.validated_artifact_kinds.append(kind)
         failed = spec["objective"] == "decode evidence" and spec["version"] == 1
         verdict = "failed" if failed else "passed"
-        snapshot = prompt["artifact_snapshot"]
-        refs = [*snapshot["artifact_refs"], *snapshot["evidence_refs"]]
-        artifact_ref = refs[0]
+        evidence_handle = role_context["evidence_handles"][0]["evidence_handle"]
         return self.call(
             "submit_validation",
             {
@@ -342,7 +329,7 @@ class _ScriptedLLM:
                         "criterion_id": item["criterion_id"],
                         "verdict": verdict,
                         "reason": "checked frozen snapshot",
-                        "evidence_refs": [] if failed else [artifact_ref],
+                        "evidence_handle_ids": [] if failed else [evidence_handle],
                     }
                     for item in spec["acceptance_criteria"]
                 ],
@@ -351,9 +338,12 @@ class _ScriptedLLM:
                         {
                             "defect_code": "DECODE_QUERY_TOO_BROAD",
                             "criterion_id": spec["acceptance_criteria"][0]["criterion_id"],
-                            "scope_ref": "script-build://scopes/direction/decode",
+                            "scope_selector": {
+                                "anchor": "mission",
+                                "path": ["direction", "decode"],
+                            },
                             "observed_excerpt": "v1 decode evidence was too broad",
-                            "evidence_refs": [],
+                            "evidence_handle_ids": [],
                             "severity": "hard",
                             "invalidated_inputs": [],
                             "recommended_action_class": "revise",

script_build_host/tests/test_phase_two_contracts.py

@@ -31,6 +31,7 @@ from script_build_host.application.phase_two_planning import (
     PhaseTwoPlanningService,
     _execution_seconds,
 )
+from script_build_host.domain.artifacts import DirectionArtifact, DirectionGoal
 from script_build_host.domain.errors import ProtocolViolation
 from script_build_host.domain.records import BuildStatus, MissionBinding, Principal
 from script_build_host.domain.task_contracts import (
@@ -75,6 +76,7 @@ def _contract(
     }[kind]
     payload: dict[str, object] = {
         "schema_version": "script-task-contract/v1",
+        "compiler_manifest_digest": "sha256:" + "0" * 64,
         "task_kind": kind,
         "scope_ref": scope,
         "intent_class": (
@@ -104,9 +106,7 @@ def _contract(
             "max_external_queries": 40,
             "max_no_improvement": 3,
         },
-        "goal_ids": (
-            [] if kind in {"direction", "decode-retrieval"} else ["goal-1"]
-        ),
+        "goal_ids": ([] if kind in {"direction", "decode-retrieval"} else ["goal-1"]),
         "supersedes_decision_ids": [],
         "candidate_closure_decision_refs": [],
         "adopted_decision_ids": [],
@@ -134,17 +134,25 @@ def _contract(
 
 def _planner_input(payload: dict[str, object]) -> dict[str, object]:
     input_refs = cast(list[dict[str, Any]], payload.get("input_decision_refs") or [])
-    comparison_refs = cast(
-        list[dict[str, Any]], payload.get("comparison_decision_refs") or []
-    )
+    comparison_refs = cast(list[dict[str, Any]], payload.get("comparison_decision_refs") or [])
     base = cast(dict[str, Any] | None, payload.get("base_artifact_ref"))
+    scope = str(payload["scope_ref"])
+    prefix = "script-build://scopes/"
+    path = scope.removeprefix(prefix).split("/") if scope.startswith(prefix) else ["full"]
     return {
         "task_kind": payload["task_kind"],
-        "scope_ref": payload["scope_ref"],
+        "scope_selector": {"anchor": "mission", "path": path[:4]},
         "intent_class": payload["intent_class"],
         "objective": payload["objective"],
         "goal_ids": payload["goal_ids"],
-        "criteria": payload["criteria"],
+        "criteria": [
+            {
+                "client_key": item["criterion_id"],
+                "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(
             (
@@ -156,7 +164,11 @@ def _planner_input(payload: dict[str, object]) -> dict[str, object]:
         ),
         "comparison_decision_ids": [item["decision_id"] for item in comparison_refs],
         "supersedes_decision_ids": payload["supersedes_decision_ids"],
-        "gap_ref": payload["gap_ref"],
+        "gap_key": (
+            str(payload["gap_ref"]).rsplit("/", 1)[-1]
+            if payload.get("gap_ref") is not None
+            else None
+        ),
     }
 
 
@@ -184,9 +196,7 @@ def test_phase_policy_rejects_workspace_contract_without_capability_write_scope(
     )
     invalid = ScriptTaskContractV1.from_payload(_contract("structure"))
     with pytest.raises(TaskContractError, match="WRITE_SCOPE_VIOLATION"):
-        guard.contracts(
-            context={"phase": 2}, ledger=ledger, parent=parent, contracts=(invalid,)
-        )
+        guard.contracts(context={"phase": 2}, ledger=ledger, parent=parent, contracts=(invalid,))
 
     payload = _contract("structure")
     payload["write_scope"] = ["script-build://writes/paragraphs/main"]
@@ -202,9 +212,7 @@ def test_phase_policy_requires_broad_container_write_capability() -> None:
         _contract("candidate-portfolio", execution_ready=False)
     )
     with pytest.raises(TaskContractError, match="WRITE_SCOPE_VIOLATION"):
-        guard.contracts(
-            context={"phase": 2}, ledger=ledger, parent=root, contracts=(invalid,)
-        )
+        guard.contracts(context={"phase": 2}, ledger=ledger, parent=root, contracts=(invalid,))
 
     payload = _contract("candidate-portfolio", execution_ready=False)
     payload["write_scope"] = ["script-build://writes"]
@@ -322,6 +330,25 @@ class _Bindings:
         return self.binding
 
 
+class _DirectionArtifacts:
+    async def get_by_id(self, identifier: int, *, script_build_id: int) -> Any:
+        assert script_build_id == 7
+        return SimpleNamespace(
+            artifact_version_id=identifier,
+            artifact=DirectionArtifact(
+                goals=(
+                    DirectionGoal(
+                        goal_id="goal-1",
+                        statement="complete the script",
+                        rationale="the mission needs one complete result",
+                        success_criteria=("the result is complete",),
+                    ),
+                ),
+                evidence_refs=("script-build://artifact-versions/evidence",),
+            ),
+        )
+
+
 class _PlaceholderThenReplacementExecutor:
     """Exercise Coordinator state transitions without bypassing durable operations."""
 
@@ -372,7 +399,7 @@ class _PlaceholderThenReplacementExecutor:
         kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item)
         placeholder = kind == "compose" and context["spec_version"] == 2
         verdict = ValidationVerdict.FAILED if placeholder else ValidationVerdict.PASSED
-        criterion = cast(dict[str, Any], context["task_spec"])["acceptance_criteria"][0]
+        criteria = cast(dict[str, Any], context["task_spec"])["acceptance_criteria"]
         await self.coordinator.submit_validation(
             {
                 "role": AgentRole.VALIDATOR.value,
@@ -397,6 +424,7 @@ class _PlaceholderThenReplacementExecutor:
                         else "the replacement is concretely rendered"
                     ),
                 )
+                for criterion in criteria
             ],
             (
                 "hard_defects=1;defect_codes=REALIZATION_PLACEHOLDER"
@@ -439,7 +467,7 @@ async def _service(
         coordinator=coordinator,
         bindings=bindings,
         contracts=FileScriptTaskContractStore(tmp_path / "data"),
-        artifacts=cast(Any, SimpleNamespace()),
+        artifacts=cast(Any, _DirectionArtifacts()),
     )
     if active_direction:
         await _activate_direction(service, coordinator, bindings, root)
@@ -469,13 +497,14 @@ async def _activate_direction(
         task_id=direction_id,
         action=DecisionAction.ACCEPT.value,
         reason="fixture Direction passed",
-        validation_id=cast(str, cycle["validation_id"]),
+        validation_id=None,
         replacement_contract=None,
         child_contracts=(),
         context={"root_trace_id": root, "tool_call_id": "fixture:accept-direction"},
     )
     ledger = await coordinator.task_store.load(root)
     decision = ledger.decisions[cast(str, accepted["decision_id"])]
+    assert decision.validation_id == cycle["validation_id"]
     attempt = ledger.attempts[cast(str, decision.attempt_id)]
     assert attempt.submission is not None
     direction_ref = attempt.submission.artifact_refs[0]
@@ -485,6 +514,27 @@ async def _activate_direction(
     )
 
 
+@pytest.mark.asyncio
+async def test_plan_inspection_returns_active_frontier_and_compact_accept_index(
+    tmp_path: Path,
+) -> None:
+    service, _, root = await _service(tmp_path)
+    view = await service.inspect_script_plan(context={"root_trace_id": root})
+
+    assert "tasks" not in view
+    assert view["root"]["status"] == TaskStatus.NEEDS_REPLAN.value
+    assert [item["task_kind"] for item in view["active_tasks"]] == ["root"]
+    assert len(view["accepted_decisions"]) == 1
+    accepted = view["accepted_decisions"][0]
+    assert accepted["task_kind"] == "direction"
+    assert accepted["artifact_kind"] == "direction"
+    assert "digest" not in accepted
+    assert view["counts_by_status"] == {
+        TaskStatus.NEEDS_REPLAN.value: 1,
+        TaskStatus.COMPLETED.value: 1,
+    }
+
+
 @pytest.mark.asyncio
 async def test_dispatch_explains_that_blocked_children_do_not_close_parent(
     tmp_path: Path,
@@ -518,13 +568,35 @@ async def test_dispatch_explains_that_blocked_children_do_not_close_parent(
         )
 
 
+@pytest.mark.asyncio
+async def test_phase_one_direction_requires_an_accepted_retrieval_before_dispatch(
+    tmp_path: Path,
+) -> None:
+    service, _coordinator, root = await _service(tmp_path, active_direction=False)
+    context = {"root_trace_id": root, "phase": 1}
+    direction = await service.plan_script_tasks(
+        contract_payloads=[_planner_input(_contract("direction"))],
+        parent_task_id=None,
+        context={**context, "tool_call_id": "direction"},
+    )
+    await service.plan_script_tasks(
+        contract_payloads=[_planner_input(_contract("decode-retrieval"))],
+        parent_task_id=direction["task_ids"][0],
+        context={**context, "tool_call_id": "retrieval"},
+    )
+
+    with pytest.raises(TaskContractError, match="Retrieval ACCEPT"):
+        await service.dispatch_script_tasks(
+            task_ids=[direction["task_ids"][0]],
+            context={**context, "tool_call_id": "dispatch-direction"},
+        )
+
+
 @pytest.mark.asyncio
 async def test_unique_portfolio_container_cannot_be_cancelled(tmp_path: Path) -> None:
     service, _coordinator, root = await _service(tmp_path)
     portfolio = await service.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "portfolio"},
     )
@@ -582,8 +654,16 @@ async def test_accepted_input_scope_is_rejected_before_worker_dispatch(tmp_path:
 
     service.contract_for_task = frozen_contract  # type: ignore[method-assign]
 
-    with pytest.raises(TaskContractError, match="INPUT_SCOPE_MISMATCH"):
+    with pytest.raises(TaskContractError, match="INPUT_SCOPE_MISMATCH") as raised:
         await service._guard_accepted_input_scopes(root, ledger, (paragraph,))
+    assert raised.value.details == {
+        "decision_id": "accepted-structure",
+        "consumer_kind": "paragraph",
+        "consumer_scope": paragraph.scope_ref,
+        "producer_kind": "structure",
+        "producer_scope": structure.scope_ref,
+        "source": "explicit",
+    }
 
 
 class _BlockingExecutor:
@@ -650,13 +730,11 @@ async def test_real_operation_stop_converges_and_forbids_new_dispatch(tmp_path:
         coordinator=coordinator,
         bindings=bindings,
         contracts=FileScriptTaskContractStore(tmp_path / "stop-contracts"),
-        artifacts=cast(Any, SimpleNamespace()),
+        artifacts=cast(Any, _DirectionArtifacts()),
     )
     await _activate_direction(planning, coordinator, bindings, root)
     portfolio = await planning.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "stop:portfolio"},
     )
@@ -743,9 +821,7 @@ async def test_contract_store_is_content_addressed_and_detects_tampering(tmp_pat
 async def test_orphan_contract_reclamation_uses_only_ledger_references(tmp_path: Path) -> None:
     service, _, root = await _service(tmp_path)
     referenced = await service.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "portfolio"},
     )
@@ -855,9 +931,7 @@ def test_comparison_refs_are_kind_specific_and_limits_cannot_be_raised() -> None
 async def test_planning_freezes_contract_and_creates_only_one_portfolio(tmp_path: Path) -> None:
     service, coordinator, root = await _service(tmp_path)
     result = await service.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "plan-portfolio"},
     )
@@ -897,6 +971,24 @@ async def test_protected_phase_rejects_cross_phase_root_contracts(tmp_path: Path
     assert set(ledger.tasks) == existing_task_ids
 
 
+@pytest.mark.asyncio
+async def test_phase_two_root_violation_describes_attempted_and_required_kind(
+    tmp_path: Path,
+) -> None:
+    service, _, root = await _service(tmp_path)
+    with pytest.raises(TaskContractError) as caught:
+        await service.plan_script_tasks(
+            contract_payloads=[_planner_input(_contract("structure"))],
+            parent_task_id=None,
+            context={"root_trace_id": root, "phase": 2},
+        )
+
+    error = caught.value
+    assert "must first plan exactly one candidate-portfolio" in error.summary
+    assert error.details["attempted_task_kinds"] == ["structure"]
+    assert error.details["allowed_task_kinds"] == ["candidate-portfolio"]
+
+
 @pytest.mark.asyncio
 async def test_invalid_contract_batch_creates_zero_tasks(tmp_path: Path) -> None:
     service, coordinator, root = await _service(tmp_path)
@@ -921,9 +1013,7 @@ async def test_invalid_contract_batch_creates_zero_tasks(tmp_path: Path) -> None
 async def test_planning_rejects_unresolved_supersession_before_freeze(tmp_path: Path) -> None:
     service, coordinator, root = await _service(tmp_path)
     portfolio = await service.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "portfolio"},
     )
@@ -952,9 +1042,7 @@ async def test_planning_rejects_unresolved_supersession_before_freeze(tmp_path:
 async def test_open_compose_container_cannot_dispatch(tmp_path: Path) -> None:
     service, coordinator, root = await _service(tmp_path)
     portfolio = await service.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "p"},
     )
@@ -980,9 +1068,7 @@ async def test_compose_placeholder_split_replacement_then_revised_contract_passe
     service, coordinator, root = await _service(tmp_path)
     coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator))
     portfolio = await service.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "portfolio"},
     )
@@ -1141,9 +1227,7 @@ async def test_exploration_order_and_completion_order_do_not_choose_compose_orde
     service, coordinator, root = await _service(tmp_path)
     coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator))
     portfolio = await service.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "portfolio"},
     )
@@ -1214,9 +1298,7 @@ async def test_phase_two_retrieval_counts_toward_task_and_depth_limits(tmp_path:
     service, _, root = await _service(tmp_path)
     service.limits = PhaseTwoLimits(max_tasks=2)
     portfolio = await service.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "portfolio"},
     )
@@ -1236,9 +1318,7 @@ async def test_phase_two_retrieval_counts_toward_task_and_depth_limits(tmp_path:
     with pytest.raises(TaskContractError, match="depth"):
         await service.plan_script_tasks(
             contract_payloads=[
-                _planner_input(
-                    _contract("structure", scope="script-build://scopes/full/local")
-                )
+                _planner_input(_contract("structure", scope="script-build://scopes/full/local"))
             ],
             parent_task_id=compose["task_ids"][0],
             context={"root_trace_id": root, "tool_call_id": "too-deep"},
@@ -1249,9 +1329,7 @@ async def test_phase_two_retrieval_counts_toward_task_and_depth_limits(tmp_path:
 async def test_task_contract_cannot_be_silently_changed_in_task_spec(tmp_path: Path) -> None:
     service, coordinator, root = await _service(tmp_path)
     result = await service.plan_script_tasks(
-        contract_payloads=[
-            _planner_input(_contract("candidate-portfolio", execution_ready=False))
-        ],
+        contract_payloads=[_planner_input(_contract("candidate-portfolio", execution_ready=False))],
         parent_task_id=None,
         context={"root_trace_id": root},
     )