主机：定义 Planner 精简输入与冻结任务合同

新增 PlannerTaskInput 和 Goal 驱动的合同字段，由 Host 保留完整 ScriptTaskContract 冻结格式；收紧各 TaskKind 的 goal_ids、输入类型与输出约束，并同步合同参数化测试。

script_build_host/src/script_build_host/domain/task_contracts.py

@@ -199,6 +199,116 @@ class ScriptTaskBudget:
             ) from exc
 
 
+@dataclass(frozen=True, slots=True)
+class PlannerTaskInput:
+    """Planner-owned semantics; the Host derives every immutable reference."""
+
+    task_kind: ScriptTaskKind
+    scope_ref: str
+    intent_class: ScriptIntentClass
+    objective: str
+    goal_ids: tuple[str, ...]
+    criteria: tuple[ScriptCriterion, ...]
+    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
+
+    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")
+        _require_bounded_text(self.objective, "objective", MAX_OBJECTIVE_CHARS)
+        if not self.criteria or len(self.criteria) > MAX_CRITERIA:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID",
+                f"criteria must contain between 1 and {MAX_CRITERIA} items",
+            )
+        _require_unique((item.criterion_id for item in self.criteria), "criterion IDs")
+        if len(self.goal_ids) > MAX_DIRECTION_GOALS:
+            raise TaskContractError(
+                "GOAL_SCOPE_INVALID",
+                f"goal_ids may contain at most {MAX_DIRECTION_GOALS} values",
+            )
+        if any(not item.strip() for item in self.goal_ids):
+            raise TaskContractError("GOAL_SCOPE_INVALID", "goal_ids must not contain blanks")
+        for label, values in (
+            ("goal_ids", self.goal_ids),
+            ("input_decision_ids", self.input_decision_ids),
+            ("comparison_decision_ids", self.comparison_decision_ids),
+            ("supersedes_decision_ids", self.supersedes_decision_ids),
+        ):
+            _require_unique(values, label)
+        for value in (
+            *self.input_decision_ids,
+            *self.comparison_decision_ids,
+            *self.supersedes_decision_ids,
+        ):
+            _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.task_kind is ScriptTaskKind.COMPARE and len(self.comparison_decision_ids) < 2:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "Compare requires at least two Decision IDs"
+            )
+        if self.task_kind is not ScriptTaskKind.COMPARE and self.comparison_decision_ids:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "comparison_decision_ids require a Compare Task"
+            )
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> PlannerTaskInput:
+        allowed = {
+            "task_kind",
+            "scope_ref",
+            "intent_class",
+            "objective",
+            "goal_ids",
+            "criteria",
+            "input_decision_ids",
+            "base_decision_id",
+            "comparison_decision_ids",
+            "supersedes_decision_ids",
+            "gap_ref",
+        }
+        unknown = sorted(set(value) - allowed)
+        if unknown:
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID",
+                f"PlannerTaskInput contains Host-owned fields: {unknown}",
+            )
+        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"
+            ),
+            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"
+            ),
+            base_decision_id=(
+                str(value["base_decision_id"])
+                if value.get("base_decision_id") is not None
+                else None
+            ),
+            comparison_decision_ids=_strings(
+                value.get("comparison_decision_ids"), "comparison_decision_ids"
+            ),
+            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,
+        )
+
+
 @dataclass(frozen=True, slots=True)
 class AcceptedDecisionRef:
     decision_id: str
@@ -632,6 +742,14 @@ def _enum_value(enum_type: type[Any], value: Any, label: str) -> Any:
         raise TaskContractError("TASK_CONTRACT_INVALID", f"unsupported {label}") from exc
 
 
+def output_schema_for(kind: ScriptTaskKind, *, patched: bool = False) -> str:
+    if kind is ScriptTaskKind.PARAGRAPH:
+        return "paragraph-patch/v1" if patched else "paragraph-artifact/v1"
+    if kind is ScriptTaskKind.ELEMENT_SET:
+        return "element-set-patch/v1" if patched else "element-set-artifact/v1"
+    return next(iter(_OUTPUT_SCHEMAS[kind]))
+
+
 __all__ = [
     "MAX_ATTEMPTS_PER_TASK",
     "MAX_CANDIDATES_PER_SCOPE",
@@ -648,11 +766,13 @@ __all__ = [
     "AcceptedInputBundleV1",
     "FrozenTaskContract",
     "PhaseTwoLimits",
+    "PlannerTaskInput",
     "ScriptCriterion",
     "ScriptIntentClass",
     "ScriptTaskBudget",
     "ScriptTaskContractV1",
     "ScriptTaskKind",
     "TaskContractError",
+    "output_schema_for",
     "task_kind_from_context_refs",
 ]

script_build_host/tests/test_phase_two_contracts.py

@@ -132,6 +132,34 @@ def _contract(
     return payload
 
 
+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 []
+    )
+    base = cast(dict[str, Any] | None, payload.get("base_artifact_ref"))
+    return {
+        "task_kind": payload["task_kind"],
+        "scope_ref": payload["scope_ref"],
+        "intent_class": payload["intent_class"],
+        "objective": payload["objective"],
+        "goal_ids": payload["goal_ids"],
+        "criteria": payload["criteria"],
+        "input_decision_ids": [item["decision_id"] for item in input_refs],
+        "base_decision_id": next(
+            (
+                item["decision_id"]
+                for item in input_refs
+                if base is not None and item["artifact_ref"]["uri"] == base["uri"]
+            ),
+            None,
+        ),
+        "comparison_decision_ids": [item["decision_id"] for item in comparison_refs],
+        "supersedes_decision_ids": payload["supersedes_decision_ids"],
+        "gap_ref": payload["gap_ref"],
+    }
+
+
 def test_deep_control_uri_is_not_mistaken_for_base64_content() -> None:
     scope = (
         "script-build://mission/900032/direction/main/portfolio/main/compose/main/"
@@ -305,6 +333,7 @@ class _PlaceholderThenReplacementExecutor:
         refs = cast(dict[str, Any], context["task_spec"])["context_refs"]
         kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item)
         artifact_kind = {
+            "direction": "direction",
             "compose": "structured_script",
             "structure": "structure",
             "paragraph": "paragraph",
@@ -341,7 +370,7 @@ class _PlaceholderThenReplacementExecutor:
     async def run_validator(self, context: dict[str, Any]) -> ValidatorRunResult:
         refs = cast(dict[str, Any], context["task_spec"])["context_refs"]
         kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item)
-        placeholder = kind == "compose" and context["spec_version"] == 1
+        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]
         await self.coordinator.submit_validation(
@@ -386,7 +415,9 @@ class _PlaceholderThenReplacementExecutor:
         return True
 
 
-async def _service(tmp_path: Path) -> tuple[PhaseTwoPlanningService, TaskCoordinator, str]:
+async def _service(
+    tmp_path: Path, *, active_direction: bool = True
+) -> tuple[PhaseTwoPlanningService, TaskCoordinator, str]:
     root = "root-contracts"
     coordinator = TaskCoordinator(
         FileSystemTaskStore(str(tmp_path / "ledger")),
@@ -403,28 +434,70 @@ async def _service(tmp_path: Path) -> tuple[PhaseTwoPlanningService, TaskCoordin
             "context_refs": ["script-build://inputs/11"],
         },
     )
+    bindings = _Bindings(root)
     service = PhaseTwoPlanningService(
         coordinator=coordinator,
-        bindings=_Bindings(root),
+        bindings=bindings,
         contracts=FileScriptTaskContractStore(tmp_path / "data"),
         artifacts=cast(Any, SimpleNamespace()),
     )
+    if active_direction:
+        await _activate_direction(service, coordinator, bindings, root)
     return service, coordinator, root
 
 
+async def _activate_direction(
+    service: PhaseTwoPlanningService,
+    coordinator: TaskCoordinator,
+    bindings: _Bindings,
+    root: str,
+) -> None:
+    coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator))
+    planned = await service.plan_script_tasks(
+        contract_payloads=[_planner_input(_contract("direction"))],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "fixture:direction"},
+    )
+    direction_id = cast(str, planned["task_ids"][0])
+    cycle = (
+        await service.dispatch_script_tasks(
+            task_ids=(direction_id,),
+            context={"root_trace_id": root, "tool_call_id": "fixture:dispatch-direction"},
+        )
+    )[0]
+    accepted = await service.decide_script_task(
+        task_id=direction_id,
+        action=DecisionAction.ACCEPT.value,
+        reason="fixture Direction passed",
+        validation_id=cast(str, cycle["validation_id"]),
+        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"])]
+    attempt = ledger.attempts[cast(str, decision.attempt_id)]
+    assert attempt.submission is not None
+    direction_ref = attempt.submission.artifact_refs[0]
+    bindings.binding = replace(
+        bindings.binding,
+        active_direction_artifact_version_id=int(direction_ref.version),
+    )
+
+
 @pytest.mark.asyncio
 async def test_dispatch_explains_that_blocked_children_do_not_close_parent(
     tmp_path: Path,
 ) -> None:
-    service, coordinator, root = await _service(tmp_path)
+    service, coordinator, root = await _service(tmp_path, active_direction=False)
     direction = await service.plan_script_tasks(
-        contract_payloads=[_contract("direction")],
+        contract_payloads=[_planner_input(_contract("direction"))],
         parent_task_id=None,
         context={"root_trace_id": root, "tool_call_id": "direction"},
     )
     direction_id = direction["task_ids"][0]
     retrieval = await service.plan_script_tasks(
-        contract_payloads=[_contract("decode-retrieval")],
+        contract_payloads=[_planner_input(_contract("decode-retrieval"))],
         parent_task_id=direction_id,
         context={"root_trace_id": root, "tool_call_id": "retrieval"},
     )
@@ -449,7 +522,9 @@ async def test_dispatch_explains_that_blocked_children_do_not_close_parent(
 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=[_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"},
     )
@@ -577,18 +652,21 @@ async def test_real_operation_stop_converges_and_forbids_new_dispatch(tmp_path:
         contracts=FileScriptTaskContractStore(tmp_path / "stop-contracts"),
         artifacts=cast(Any, SimpleNamespace()),
     )
+    await _activate_direction(planning, coordinator, bindings, root)
     portfolio = await planning.plan_script_tasks(
-        contract_payloads=[_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"},
     )
     compose = await planning.plan_script_tasks(
-        contract_payloads=[_contract("compose", execution_ready=False)],
+        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
         parent_task_id=portfolio["task_ids"][0],
         context={"root_trace_id": root, "tool_call_id": "stop:compose"},
     )
     paragraph = await planning.plan_script_tasks(
-        contract_payloads=[_contract("paragraph")],
+        contract_payloads=[_planner_input(_contract("paragraph"))],
         parent_task_id=compose["task_ids"][0],
         context={"root_trace_id": root, "tool_call_id": "stop:paragraph"},
     )
@@ -665,7 +743,9 @@ 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=[_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"},
     )
@@ -775,7 +855,9 @@ 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=[_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"},
     )
@@ -790,7 +872,9 @@ async def test_planning_freezes_contract_and_creates_only_one_portfolio(tmp_path
 
     with pytest.raises(TaskContractError, match="one Portfolio"):
         await service.plan_script_tasks(
-            contract_payloads=[_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-2"},
         )
@@ -799,101 +883,151 @@ async def test_planning_freezes_contract_and_creates_only_one_portfolio(tmp_path
 @pytest.mark.asyncio
 async def test_protected_phase_rejects_cross_phase_root_contracts(tmp_path: Path) -> None:
     service, coordinator, root = await _service(tmp_path)
+    before = await coordinator.task_store.load(root)
+    existing_task_ids = set(before.tasks)
     with pytest.raises(TaskContractError, match="PHASE_POLICY_VIOLATION"):
         await service.plan_script_tasks(
-            contract_payloads=[_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, "phase": 1},
         )
     ledger = await coordinator.task_store.load(root)
-    assert list(ledger.tasks) == [ledger.root_task_id]
+    assert set(ledger.tasks) == existing_task_ids
 
 
 @pytest.mark.asyncio
 async def test_invalid_contract_batch_creates_zero_tasks(tmp_path: Path) -> None:
     service, coordinator, root = await _service(tmp_path)
-    invalid = _contract("paragraph")
+    before = await coordinator.task_store.load(root)
+    existing_task_ids = set(before.tasks)
+    invalid = _planner_input(_contract("paragraph"))
     invalid["output_schema"] = "structured-script/v1"
-    with pytest.raises(TaskContractError, match="output_schema"):
+    with pytest.raises(TaskContractError, match="Host-owned fields"):
         await service.plan_script_tasks(
-            contract_payloads=[_contract("candidate-portfolio"), invalid],
+            contract_payloads=[
+                _planner_input(_contract("candidate-portfolio")),
+                invalid,
+            ],
             parent_task_id=None,
             context={"root_trace_id": root},
         )
     ledger = await coordinator.task_store.load(root)
-    assert list(ledger.tasks) == [ledger.root_task_id]
+    assert set(ledger.tasks) == existing_task_ids
 
 
 @pytest.mark.asyncio
 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=[_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"},
     )
     compose = await service.plan_script_tasks(
-        contract_payloads=[_contract("compose", execution_ready=False)],
+        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
         parent_task_id=portfolio["task_ids"][0],
         context={"root_trace_id": root, "tool_call_id": "compose"},
     )
     replacement = _contract("paragraph")
     replacement["supersedes_decision_ids"] = ["missing-accept"]
+    before = await coordinator.task_store.load(root)
+    existing_task_ids = set(before.tasks)
 
-    with pytest.raises(TaskContractError, match="import every superseded"):
+    with pytest.raises(TaskContractError, match="not an ACCEPT decision"):
         await service.plan_script_tasks(
-            contract_payloads=[replacement],
+            contract_payloads=[_planner_input(replacement)],
             parent_task_id=compose["task_ids"][0],
             context={"root_trace_id": root, "tool_call_id": "invalid-replacement"},
         )
 
     ledger = await coordinator.task_store.load(root)
-    assert set(ledger.tasks) == {
-        ledger.root_task_id,
-        portfolio["task_ids"][0],
-        compose["task_ids"][0],
-    }
+    assert set(ledger.tasks) == existing_task_ids
 
 
 @pytest.mark.asyncio
 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=[_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"},
     )
     compose = await service.plan_script_tasks(
-        contract_payloads=[_contract("compose", execution_ready=False)],
+        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
         parent_task_id=portfolio["task_ids"][0],
         context={"root_trace_id": root, "tool_call_id": "c"},
     )
+    before = await coordinator.task_store.load(root)
+    operation_ids = set(before.operations)
     with pytest.raises(TaskContractError, match="first accept child candidates"):
         await service.dispatch_script_tasks(
             task_ids=compose["task_ids"], context={"root_trace_id": root}
         )
     ledger = await coordinator.task_store.load(root)
-    assert not ledger.operations
+    assert set(ledger.operations) == operation_ids
 
 
 @pytest.mark.asyncio
-async def test_compose_placeholder_split_replacement_then_revised_v2_passes(
+async def test_compose_placeholder_split_replacement_then_revised_contract_passes(
     tmp_path: Path,
 ) -> None:
     service, coordinator, root = await _service(tmp_path)
     coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator))
     portfolio = await service.plan_script_tasks(
-        contract_payloads=[_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"},
     )
     compose = await service.plan_script_tasks(
-        contract_payloads=[_contract("compose")],
+        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
         parent_task_id=portfolio["task_ids"][0],
         context={"root_trace_id": root, "tool_call_id": "compose-v1"},
     )
     compose_id = compose["task_ids"][0]
 
+    accepted_children: dict[str, str] = {}
+    for kind in ("structure", "paragraph", "element-set"):
+        planned = await service.plan_script_tasks(
+            contract_payloads=[_planner_input(_contract(kind))],
+            parent_task_id=compose_id,
+            context={"root_trace_id": root, "tool_call_id": f"plan:{kind}"},
+        )
+        child_id = cast(str, planned["task_ids"][0])
+        child_cycle = (
+            await service.dispatch_script_tasks(
+                task_ids=(child_id,),
+                context={"root_trace_id": root, "tool_call_id": f"dispatch:{kind}"},
+            )
+        )[0]
+        accepted = await service.decide_script_task(
+            task_id=child_id,
+            action=DecisionAction.ACCEPT.value,
+            reason=f"{kind} passed",
+            validation_id=cast(str, child_cycle["validation_id"]),
+            replacement_contract=None,
+            child_contracts=(),
+            context={"root_trace_id": root, "tool_call_id": f"accept:{kind}"},
+        )
+        accepted_children[kind] = cast(str, accepted["decision_id"])
+    await service.decide_script_task(
+        task_id=compose_id,
+        action=DecisionAction.REVISE.value,
+        reason="freeze the initial complete frontier",
+        validation_id=None,
+        replacement_contract=None,
+        child_contracts=(),
+        selected_decision_ids=tuple(accepted_children.values()),
+        context={"root_trace_id": root, "tool_call_id": "revise-compose-v1"},
+    )
+
     first_cycle = (
         await service.dispatch_script_tasks(
             task_ids=(compose_id,),
@@ -913,7 +1047,14 @@ async def test_compose_placeholder_split_replacement_then_revised_v2_passes(
         reason="replace only the unrealized opening scope",
         validation_id=first_validation.validation_id,
         replacement_contract=None,
-        child_contracts=(_contract("paragraph"),),
+        child_contracts=(
+            _planner_input(
+                _contract(
+                    "paragraph",
+                    scope="script-build://scopes/full/replacement",
+                )
+            ),
+        ),
         context={"root_trace_id": root, "tool_call_id": "split-placeholder"},
     )
     replacement_id = split["payload"]["child_task_ids"][0]
@@ -936,33 +1077,18 @@ async def test_compose_placeholder_split_replacement_then_revised_v2_passes(
     ledger = await coordinator.task_store.load(root)
     assert ledger.tasks[compose_id].status is TaskStatus.NEEDS_REPLAN
     replacement_decision_id = replacement_accept["decision_id"]
-    replacement_attempt = ledger.attempts[
-        cast(str, ledger.decisions[replacement_decision_id].attempt_id)
-    ]
-    assert replacement_attempt.submission is not None
-    replacement_ref = replacement_attempt.submission.artifact_refs[0]
-    frozen_replacement = {
-        "decision_id": replacement_decision_id,
-        "artifact_ref": {
-            "uri": replacement_ref.uri,
-            "kind": replacement_ref.kind,
-            "version": replacement_ref.version,
-            "digest": replacement_ref.digest,
-        },
-        "scope_ref": "script-build://scopes/full",
-        "expected_task_kind": "paragraph",
-    }
-    compose_v2 = _contract("compose", execution_ready=False)
-    compose_v2["candidate_closure_decision_refs"] = [frozen_replacement]
-    compose_v2["adopted_decision_ids"] = [replacement_decision_id]
-    compose_v2["compose_order"] = [replacement_decision_id]
     await service.decide_script_task(
         task_id=compose_id,
         action=DecisionAction.REVISE.value,
         reason="pin the accepted replacement and retry the complete render",
         validation_id=None,
-        replacement_contract=compose_v2,
+        replacement_contract=None,
         child_contracts=(),
+        selected_decision_ids=(
+            accepted_children["structure"],
+            replacement_decision_id,
+            accepted_children["element-set"],
+        ),
         context={"root_trace_id": root, "tool_call_id": "revise-compose-v2"},
     )
     second_cycle = (
@@ -984,12 +1110,13 @@ async def test_compose_placeholder_split_replacement_then_revised_v2_passes(
     reloaded = await FileSystemTaskStore(str(tmp_path / "ledger")).load(root)
     compose_task = reloaded.tasks[compose_id]
     assert compose_task.status is TaskStatus.COMPLETED
-    assert compose_task.current_spec_version == 2
+    assert compose_task.current_spec_version == 3
     assert len(compose_task.attempt_ids) == 2
     assert [
         reloaded.validations[validation_id].verdict for validation_id in compose_task.validation_ids
     ] == [ValidationVerdict.FAILED, ValidationVerdict.PASSED]
     assert [reloaded.decisions[item].action for item in compose_task.decision_ids] == [
+        DecisionAction.REVISE,
         DecisionAction.SPLIT,
         DecisionAction.REVISE,
         DecisionAction.ACCEPT,
@@ -1014,12 +1141,14 @@ 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=[_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"},
     )
     compose = await service.plan_script_tasks(
-        contract_payloads=[_contract("compose", execution_ready=False)],
+        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
         parent_task_id=portfolio["task_ids"][0],
         context={"root_trace_id": root, "tool_call_id": "compose"},
     )
@@ -1027,14 +1156,13 @@ async def test_exploration_order_and_completion_order_do_not_choose_compose_orde
     task_by_kind: dict[str, str] = {}
     for kind in creation_order:
         planned = await service.plan_script_tasks(
-            contract_payloads=[_contract(kind)],
+            contract_payloads=[_planner_input(_contract(kind))],
             parent_task_id=compose_id,
             context={"root_trace_id": root, "tool_call_id": f"plan:{kind}"},
         )
         task_by_kind[kind] = planned["task_ids"][0]
 
     decision_by_kind: dict[str, str] = {}
-    ref_by_kind: dict[str, ArtifactRef] = {}
     completion_order = tuple(reversed(creation_order))
     for kind in completion_order:
         task_id = task_by_kind[kind]
@@ -1055,47 +1183,25 @@ async def test_exploration_order_and_completion_order_do_not_choose_compose_orde
         )
         decision_id = accepted["decision_id"]
         decision_by_kind[kind] = decision_id
-        ledger = await coordinator.task_store.load(root)
-        attempt = ledger.attempts[cast(str, ledger.decisions[decision_id].attempt_id)]
-        assert attempt.submission is not None
-        ref_by_kind[kind] = attempt.submission.artifact_refs[0]
 
     ledger = await coordinator.task_store.load(root)
     assert ledger.tasks[compose_id].status is TaskStatus.NEEDS_REPLAN
     adoption_kinds = ("paragraph", "structure", "element-set")
-    closure = []
-    for kind in adoption_kinds:
-        ref = ref_by_kind[kind]
-        closure.append(
-            {
-                "decision_id": decision_by_kind[kind],
-                "artifact_ref": {
-                    "uri": ref.uri,
-                    "kind": ref.kind,
-                    "version": ref.version,
-                    "digest": ref.digest,
-                },
-                "scope_ref": "script-build://scopes/full",
-                "expected_task_kind": kind,
-            }
-        )
-    execution_ready = _contract("compose", execution_ready=False)
-    execution_ready["candidate_closure_decision_refs"] = closure
-    execution_ready["adopted_decision_ids"] = [decision_by_kind[kind] for kind in adoption_kinds]
-    execution_ready["compose_order"] = [decision_by_kind[kind] for kind in adoption_kinds]
+    selected = tuple(decision_by_kind[kind] for kind in adoption_kinds)
     await service.decide_script_task(
         task_id=compose_id,
         action=DecisionAction.REVISE.value,
         reason="formal adoption is based on scope and intent, not timing",
         validation_id=None,
-        replacement_contract=execution_ready,
+        replacement_contract=None,
         child_contracts=(),
+        selected_decision_ids=selected,
         context={"root_trace_id": root, "tool_call_id": "close-compose"},
     )
 
     reloaded = await FileSystemTaskStore(str(tmp_path / "ledger")).load(root)
     frozen = await service.contract_for_task(root, reloaded.tasks[compose_id])
-    assert frozen.contract.compose_order == tuple(decision_by_kind[kind] for kind in adoption_kinds)
+    assert frozen.contract.compose_order == selected
     assert [
         next(kind for kind, task_id in task_by_kind.items() if task_id == child_id)
         for child_id in reloaded.tasks[compose_id].child_task_ids
@@ -1108,18 +1214,20 @@ 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=[_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"},
     )
     compose = await service.plan_script_tasks(
-        contract_payloads=[_contract("compose", execution_ready=False)],
+        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
         parent_task_id=portfolio["task_ids"][0],
         context={"root_trace_id": root, "tool_call_id": "compose"},
     )
     with pytest.raises(TaskContractError, match="Task budget"):
         await service.plan_script_tasks(
-            contract_payloads=[_contract("decode-retrieval")],
+            contract_payloads=[_planner_input(_contract("decode-retrieval"))],
             parent_task_id=compose["task_ids"][0],
             context={"root_trace_id": root, "tool_call_id": "retrieval"},
         )
@@ -1127,7 +1235,11 @@ async def test_phase_two_retrieval_counts_toward_task_and_depth_limits(tmp_path:
     service.limits = PhaseTwoLimits(max_depth=1)
     with pytest.raises(TaskContractError, match="depth"):
         await service.plan_script_tasks(
-            contract_payloads=[_contract("structure", scope="script-build://scopes/full/local")],
+            contract_payloads=[
+                _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"},
         )
@@ -1137,7 +1249,9 @@ 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=[_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},
     )