主机：由 Decision ID 确定性构造 Phase2 任务合同

Planner 只选择语义输入与 Decision ID，Host 解析 accepted Decision、Direction、scope、Artifact digest 和预算，自动冻结 adoption closure，并阻断等价任务与非法 Goal 组合。

script_build_host/src/script_build_host/application/phase_two_planning.py

@@ -7,8 +7,9 @@ checks script-build invariants before calling Coordinator operations.
 
 from __future__ import annotations
 
+import json
 from collections.abc import Mapping, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import UTC, datetime
 from typing import Any, Protocol, cast
 
@@ -16,6 +17,7 @@ from agent.orchestration import (
     DecisionAction,
     OperationStatus,
     TaskStatus,
+    ValidationVerdict,
 )
 
 from script_build_host.domain.artifacts import DirectionArtifact
@@ -32,11 +34,15 @@ from script_build_host.domain.ports import (
     ScriptBusinessArtifactRepository,
 )
 from script_build_host.domain.task_contracts import (
+    AcceptedDecisionRef,
     FrozenTaskContract,
     PhaseTwoLimits,
+    PlannerTaskInput,
+    ScriptTaskBudget,
     ScriptTaskContractV1,
     ScriptTaskKind,
     TaskContractError,
+    output_schema_for,
     task_kind_from_context_refs,
 )
 
@@ -83,6 +89,16 @@ _CANDIDATE_KINDS = frozenset(
         ScriptTaskKind.COMPOSE,
     }
 )
+_DIRECTION_INPUT_KINDS = frozenset(
+    {
+        ScriptTaskKind.STRUCTURE,
+        ScriptTaskKind.PARAGRAPH,
+        ScriptTaskKind.ELEMENT_SET,
+        ScriptTaskKind.COMPARE,
+        ScriptTaskKind.COMPOSE,
+        ScriptTaskKind.CANDIDATE_PORTFOLIO,
+    }
+)
 RETRIEVAL_KINDS = frozenset(
     {
         ScriptTaskKind.PATTERN_RETRIEVAL.value,
@@ -191,6 +207,50 @@ class PhasePolicyGuard:
         if phase == 2:
             self._phase_two_contracts(contracts)
 
+    def planner_inputs(
+        self,
+        *,
+        context: Mapping[str, Any],
+        ledger: Any,
+        parent: Any,
+        inputs: Sequence[PlannerTaskInput],
+    ) -> None:
+        """Reject cross-phase semantics before resolving accepted Decisions."""
+
+        phase = self.phase(context)
+        if phase is None:
+            return
+        parent_is_root = parent.task_id == ledger.root_task_id
+        parent_kind = task_kind_from_context_refs(parent.current_spec.context_refs)
+        kinds = {item.task_kind for item in inputs}
+        if phase == 1:
+            allowed = (
+                {ScriptTaskKind.DIRECTION}
+                if parent_is_root
+                else (
+                    set(ScriptTaskKind(item) for item in RETRIEVAL_KINDS)
+                    if parent_kind == ScriptTaskKind.DIRECTION.value
+                    else set()
+                )
+            )
+        elif phase == 2:
+            allowed = (
+                {ScriptTaskKind.CANDIDATE_PORTFOLIO}
+                if parent_is_root
+                else set(_PHASE_TWO_KINDS)
+            )
+        else:
+            allowed = set()
+        if not kinds <= allowed:
+            raise TaskContractError(
+                "PHASE_POLICY_VIOLATION",
+                f"phase {phase} cannot plan {sorted(item.value for item in kinds)} here",
+            )
+        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"
+            )
+
     def dispatch(self, *, context: Mapping[str, Any], kind: ScriptTaskKind) -> None:
         phase = self.phase(context)
         if phase is None:
@@ -401,8 +461,25 @@ class PhaseTwoPlanningService:
         if not supplied:
             raise TaskContractError("TASK_CONTRACT_INVALID", "at least one contract is required")
 
-        # Parse every value before writing any content-addressed blob.
-        parsed_contracts = tuple(ScriptTaskContractV1.from_payload(value) for value in supplied)
+        # Resolve model-owned semantics before writing any content-addressed blob.
+        planner_inputs = tuple(PlannerTaskInput.from_payload(value) for value in supplied)
+        self.phase_policy.planner_inputs(
+            context=context,
+            ledger=ledger,
+            parent=parent,
+            inputs=planner_inputs,
+        )
+        parsed_contracts = tuple(
+            [
+                await self._resolve_planner_input(
+                    value,
+                    binding=binding,
+                    ledger=ledger,
+                    root_trace_id=root_trace_id,
+                )
+                for value in planner_inputs
+            ]
+        )
         self.phase_policy.contracts(
             context=context, ledger=ledger, parent=parent, contracts=parsed_contracts
         )
@@ -421,6 +498,12 @@ class PhaseTwoPlanningService:
         await self._guard_hierarchy(root_trace_id, ledger, parent, parsed_contracts)
         await self._guard_supersessions(root_trace_id, ledger, parsed_contracts)
         await self._guard_accepted_input_scopes(root_trace_id, ledger, parsed_contracts)
+        await self._guard_duplicate_contracts(
+            root_trace_id=root_trace_id,
+            ledger=ledger,
+            parent=parent,
+            contracts=parsed_contracts,
+        )
         planned = await self._freeze_all(
             root_trace_id, str(binding.input_snapshot_id), parsed_contracts
         )
@@ -455,6 +538,7 @@ class PhaseTwoPlanningService:
         validation_id: str | None,
         replacement_contract: Mapping[str, Any] | None,
         child_contracts: Sequence[Mapping[str, Any]],
+        selected_decision_ids: Sequence[str] = (),
         context: Mapping[str, Any],
     ) -> dict[str, Any]:
         root_trace_id = _required(context, "root_trace_id")
@@ -500,7 +584,12 @@ class PhaseTwoPlanningService:
                     "complete or revise its Compose children",
                 )
             await self._guard_repair_and_improvement(
-                root_trace_id, ledger, task, decision_action, replacement_contract
+                root_trace_id,
+                ledger,
+                task,
+                decision_action,
+                replacement_contract,
+                selected_decision_ids,
             )
 
         bounded_reason = _bounded(reason, 1_000, "reason")
@@ -517,20 +606,59 @@ class PhaseTwoPlanningService:
             bounded_reason = PHASE_ONE_CAPABILITY_BOUNDARY
         payload: dict[str, Any] = {"reason": bounded_reason}
         if decision_action is DecisionAction.REVISE and is_root:
-            if replacement_contract is None or child_contracts:
+            if replacement_contract is None or child_contracts or selected_decision_ids:
                 raise TaskContractError(
                     "TASK_CONTRACT_INVALID",
                     "Root REVISE requires one complete bounded Root specification",
                 )
             payload.update(_root_replacement(task, replacement_contract))
         elif decision_action is DecisionAction.REVISE:
-            if replacement_contract is None or child_contracts:
-                raise TaskContractError(
-                    "TASK_CONTRACT_INVALID", "REVISE requires one complete replacement contract"
+            current = await self.contract_for_task(root_trace_id, task)
+            if current.contract.task_kind in {
+                ScriptTaskKind.COMPOSE,
+                ScriptTaskKind.CANDIDATE_PORTFOLIO,
+            }:
+                if replacement_contract is not None or child_contracts or not selected_decision_ids:
+                    raise TaskContractError(
+                        "TASK_CONTRACT_INVALID",
+                        "container REVISE requires selected_decision_ids only",
+                    )
+                self.phase_policy.revisions(
+                    context=context, contracts=(current.contract,)
+                )
+                contract = await self._adoption_contract(
+                    binding=binding,
+                    ledger=ledger,
+                    root_trace_id=root_trace_id,
+                    task=task,
+                    current=current.contract,
+                    selected_decision_ids=selected_decision_ids,
+                )
+            else:
+                if replacement_contract is None or child_contracts or selected_decision_ids:
+                    raise TaskContractError(
+                        "TASK_CONTRACT_INVALID",
+                        "REVISE requires one PlannerTaskInput replacement",
+                    )
+                planner_input = PlannerTaskInput.from_payload(replacement_contract)
+                parent = ledger.tasks.get(task.parent_task_id or "")
+                if parent is None:
+                    raise TaskContractError(
+                        "TASK_CONTRACT_INVALID", "Task parent is outside this Root"
+                    )
+                self.phase_policy.planner_inputs(
+                    context=context,
+                    ledger=ledger,
+                    parent=parent,
+                    inputs=(planner_input,),
+                )
+                contract = await self._resolve_planner_input(
+                    planner_input,
+                    binding=binding,
+                    ledger=ledger,
+                    root_trace_id=root_trace_id,
                 )
-            contract = ScriptTaskContractV1.from_payload(replacement_contract)
             self.phase_policy.revisions(context=context, contracts=(contract,))
-            current = await self.contract_for_task(root_trace_id, task)
             if contract.task_kind is not current.contract.task_kind:
                 raise TaskContractError(
                     "TASK_KIND_PRESET_MISMATCH", "REVISE cannot change the Task kind"
@@ -554,11 +682,30 @@ class PhaseTwoPlanningService:
             )[0]
             payload.update(planned_replacement.draft)
         elif decision_action is DecisionAction.SPLIT:
-            if replacement_contract is not None or not child_contracts:
+            if replacement_contract is not None or not child_contracts or selected_decision_ids:
                 raise TaskContractError(
                     "TASK_CONTRACT_INVALID", "SPLIT requires complete child contracts"
                 )
-            contracts = tuple(ScriptTaskContractV1.from_payload(value) for value in child_contracts)
+            planner_inputs = tuple(
+                PlannerTaskInput.from_payload(value) for value in child_contracts
+            )
+            self.phase_policy.planner_inputs(
+                context=context,
+                ledger=ledger,
+                parent=task,
+                inputs=planner_inputs,
+            )
+            contracts = tuple(
+                [
+                    await self._resolve_planner_input(
+                        value,
+                        binding=binding,
+                        ledger=ledger,
+                        root_trace_id=root_trace_id,
+                    )
+                    for value in planner_inputs
+                ]
+            )
             self.phase_policy.contracts(
                 context=context, ledger=ledger, parent=task, contracts=contracts
             )
@@ -581,7 +728,7 @@ class PhaseTwoPlanningService:
                 root_trace_id, str(binding.input_snapshot_id), contracts
             )
             payload["tasks"] = [item.draft for item in planned_children]
-        elif replacement_contract is not None or child_contracts:
+        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"
             )
@@ -879,6 +1026,292 @@ class PhaseTwoPlanningService:
             )
         return frozen
 
+    async def _resolve_planner_input(
+        self,
+        value: PlannerTaskInput,
+        *,
+        binding: Any,
+        ledger: Any,
+        root_trace_id: str,
+    ) -> ScriptTaskContractV1:
+        decision_ids = list(value.input_decision_ids)
+        for decision_id in (value.base_decision_id, *value.supersedes_decision_ids):
+            if decision_id is not None and decision_id not in decision_ids:
+                decision_ids.append(decision_id)
+        refs = [
+            await self._resolve_decision_ref(
+                root_trace_id=root_trace_id,
+                ledger=ledger,
+                decision_id=decision_id,
+            )
+            for decision_id in decision_ids
+        ]
+        if value.task_kind in _DIRECTION_INPUT_KINDS:
+            direction = await self._active_direction_decision_ref(
+                binding=binding,
+                ledger=ledger,
+                root_trace_id=root_trace_id,
+            )
+            refs = [
+                item
+                for item in refs
+                if item.expected_task_kind is not ScriptTaskKind.DIRECTION
+            ]
+            refs.insert(0, direction)
+
+        comparison_refs = tuple(
+            [
+                await self._resolve_decision_ref(
+                    root_trace_id=root_trace_id,
+                    ledger=ledger,
+                    decision_id=decision_id,
+                )
+                for decision_id in value.comparison_decision_ids
+            ]
+        )
+        base_ref = next(
+            (
+                item.artifact_ref
+                for item in refs
+                if item.decision_id == value.base_decision_id
+            ),
+            None,
+        )
+        return ScriptTaskContractV1(
+            task_kind=value.task_kind,
+            scope_ref=value.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,
+            output_schema=output_schema_for(
+                value.task_kind, patched=value.base_decision_id is not None
+            ),
+            criteria=value.criteria,
+            budget=ScriptTaskBudget(
+                max_attempts=self.limits.max_attempts_per_task,
+                max_tokens=self.limits.max_task_tokens,
+                max_seconds=self.limits.max_task_seconds,
+                max_external_queries=self.limits.max_external_queries,
+                max_no_improvement=self.limits.max_no_improvement,
+            ),
+            goal_ids=value.goal_ids,
+            supersedes_decision_ids=value.supersedes_decision_ids,
+            comparison_decision_refs=comparison_refs,
+        )
+
+    async def _resolve_decision_ref(
+        self,
+        *,
+        root_trace_id: str,
+        ledger: Any,
+        decision_id: str,
+    ) -> AcceptedDecisionRef:
+        decision = ledger.decisions.get(decision_id)
+        if (
+            decision is None
+            or decision.action is not DecisionAction.ACCEPT
+            or decision.attempt_id is None
+            or decision.validation_id is None
+        ):
+            raise TaskContractError(
+                "CHILD_DECISION_INVALID", f"Decision {decision_id} is not an ACCEPT decision"
+            )
+        task = ledger.tasks.get(decision.task_id)
+        attempt = ledger.attempts.get(decision.attempt_id)
+        validation = ledger.validations.get(decision.validation_id)
+        if (
+            task is None
+            or not task.decision_ids
+            or task.decision_ids[-1] != decision_id
+            or attempt is None
+            or attempt.task_id != task.task_id
+            or attempt.submission is None
+            or validation is None
+            or validation.attempt_id != attempt.attempt_id
+            or validation.verdict is not ValidationVerdict.PASSED
+        ):
+            raise TaskContractError(
+                "CHILD_DECISION_INVALID",
+                f"Decision {decision_id} is stale or is not closed over a passed Attempt",
+            )
+        refs = [*attempt.submission.artifact_refs, *attempt.submission.evidence_refs]
+        if len(refs) != 1:
+            raise TaskContractError(
+                "CHILD_DECISION_INVALID", f"Decision {decision_id} does not own one Artifact"
+            )
+        producer = await self.contract_for_task(root_trace_id, task)
+        return AcceptedDecisionRef(
+            decision_id=decision_id,
+            artifact_ref=refs[0],
+            scope_ref=producer.contract.scope_ref,
+            expected_task_kind=producer.contract.task_kind,
+        )
+
+    async def _active_direction_decision_ref(
+        self, *, binding: Any, ledger: Any, root_trace_id: str
+    ) -> AcceptedDecisionRef:
+        version_id = binding.active_direction_artifact_version_id
+        if version_id is None:
+            raise TaskContractError("GOAL_SCOPE_INVALID", "Phase2 has no active Direction")
+        expected_uri = f"script-build://artifact-versions/{version_id}"
+        for task in ledger.tasks.values():
+            if not task.decision_ids:
+                continue
+            decision_id = task.decision_ids[-1]
+            try:
+                resolved = await self._resolve_decision_ref(
+                    root_trace_id=root_trace_id,
+                    ledger=ledger,
+                    decision_id=decision_id,
+                )
+            except TaskContractError:
+                continue
+            if (
+                resolved.expected_task_kind is ScriptTaskKind.DIRECTION
+                and resolved.artifact_ref.uri == expected_uri
+            ):
+                return resolved
+        raise TaskContractError(
+            "GOAL_SCOPE_INVALID", "active Direction has no current ACCEPT decision"
+        )
+
+    async def _adoption_contract(
+        self,
+        *,
+        binding: Any,
+        ledger: Any,
+        root_trace_id: str,
+        task: Any,
+        current: ScriptTaskContractV1,
+        selected_decision_ids: Sequence[str],
+    ) -> ScriptTaskContractV1:
+        selected = tuple(selected_decision_ids)
+        if not selected or len(set(selected)) != len(selected):
+            raise TaskContractError(
+                "TASK_CONTRACT_INVALID", "selected_decision_ids must be unique and non-empty"
+            )
+        expected_kinds = (
+            {ScriptTaskKind.COMPOSE}
+            if current.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO
+            else {
+                ScriptTaskKind.STRUCTURE,
+                ScriptTaskKind.PARAGRAPH,
+                ScriptTaskKind.ELEMENT_SET,
+            }
+        )
+        closure: list[AcceptedDecisionRef] = []
+        comparisons: list[AcceptedDecisionRef] = []
+        support: list[AcceptedDecisionRef] = []
+        for child_id in task.child_task_ids:
+            child = ledger.tasks.get(child_id)
+            if child is None or not child.decision_ids:
+                continue
+            decision_id = child.decision_ids[-1]
+            try:
+                ref = await self._resolve_decision_ref(
+                    root_trace_id=root_trace_id,
+                    ledger=ledger,
+                    decision_id=decision_id,
+                )
+            except TaskContractError:
+                continue
+            if ref.expected_task_kind in expected_kinds:
+                closure.append(ref)
+            elif ref.expected_task_kind is ScriptTaskKind.COMPARE:
+                comparisons.append(ref)
+            elif ref.expected_task_kind.value in RETRIEVAL_KINDS:
+                support.append(ref)
+        closure_by_id = {item.decision_id: item for item in closure}
+        unknown = set(selected) - set(closure_by_id)
+        if unknown:
+            raise TaskContractError(
+                "CHILD_DECISION_INVALID",
+                f"selected Decisions are outside the current child closure: {sorted(unknown)}",
+            )
+        if current.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO and len(selected) != 1:
+            raise TaskContractError(
+                "TASK_CONTRACT_NOT_EXECUTABLE", "Portfolio must select exactly one Compose"
+            )
+        if current.task_kind is ScriptTaskKind.COMPOSE:
+            selected_kinds = {closure_by_id[item].expected_task_kind for item in selected}
+            missing = expected_kinds - selected_kinds
+            if missing:
+                raise TaskContractError(
+                    "TASK_CONTRACT_NOT_EXECUTABLE",
+                    "Compose selection requires Structure, Paragraph and ElementSet",
+                )
+        input_refs = list(current.input_decision_refs)
+        for ref in support:
+            if ref.decision_id not in {item.decision_id for item in input_refs}:
+                input_refs.append(ref)
+        if current.task_kind in _DIRECTION_INPUT_KINDS:
+            direction = await self._active_direction_decision_ref(
+                binding=binding,
+                ledger=ledger,
+                root_trace_id=root_trace_id,
+            )
+            input_refs = [
+                item
+                for item in input_refs
+                if item.expected_task_kind is not ScriptTaskKind.DIRECTION
+            ]
+            input_refs.insert(0, direction)
+        held = tuple(item.decision_id for item in closure if item.decision_id not in set(selected))
+        return replace(
+            current,
+            input_decision_refs=tuple(input_refs),
+            candidate_closure_decision_refs=tuple(closure),
+            adopted_decision_ids=selected,
+            held_or_rejected_decision_ids=held,
+            compose_order=selected,
+            comparison_decision_refs=tuple(comparisons),
+        )
+
+    async def _guard_duplicate_contracts(
+        self,
+        *,
+        root_trace_id: str,
+        ledger: Any,
+        parent: Any,
+        contracts: Sequence[ScriptTaskContractV1],
+    ) -> None:
+        signatures: dict[str, str] = {}
+        for contract in contracts:
+            signature = _contract_signature(contract)
+            existing = signatures.get(signature)
+            if existing is not None:
+                raise TaskContractError(
+                    "TASK_CONTRACT_DUPLICATE", "one planning batch repeats an equivalent Task"
+                )
+            signatures[signature] = "batch"
+        for child_id in parent.child_task_ids:
+            child = ledger.tasks.get(child_id)
+            if child is None or child.status in {TaskStatus.CANCELLED, TaskStatus.SUPERSEDED}:
+                continue
+            existing_contract = (await self.contract_for_task(root_trace_id, child)).contract
+            signature = _contract_signature(existing_contract)
+            if signature in signatures:
+                raise TaskContractError(
+                    "TASK_CONTRACT_DUPLICATE",
+                    f"equivalent Task already exists: {child.task_id}",
+                )
+            if (
+                existing_contract.task_kind is ScriptTaskKind.COMPOSE
+                and any(item.task_kind is ScriptTaskKind.COMPOSE for item in contracts)
+                and not any(
+                    decision_id in contract.supersedes_decision_ids
+                    for decision_id in child.decision_ids[-1:]
+                    for contract in contracts
+                )
+            ):
+                raise TaskContractError(
+                    "TASK_CONTRACT_DUPLICATE",
+                    f"Portfolio already has a current Compose Task: {child.task_id}",
+                )
+
     async def _freeze_all(
         self,
         root_trace_id: str,
@@ -1303,6 +1736,7 @@ class PhaseTwoPlanningService:
         task: Any,
         action: DecisionAction,
         replacement_contract: Mapping[str, Any] | None,
+        selected_decision_ids: Sequence[str],
     ) -> None:
         if action is DecisionAction.REPAIR:
             repair_count = sum(
@@ -1328,16 +1762,17 @@ class PhaseTwoPlanningService:
         )
         if _non_improvement_streak(validations) < no_improvement_limit:
             return
+        if action is DecisionAction.REVISE and selected_decision_ids:
+            return
         if action is DecisionAction.REVISE and replacement_contract is not None:
-            new_contract = ScriptTaskContractV1.from_payload(replacement_contract)
+            replacement = PlannerTaskInput.from_payload(replacement_contract)
             old_contract = current_contract.contract
             if (
-                new_contract.intent_class != old_contract.intent_class
+                replacement.intent_class != old_contract.intent_class
                 or (
-                    _scope_contains(old_contract.scope_ref, new_contract.scope_ref)
-                    and new_contract.scope_ref != old_contract.scope_ref
+                    _scope_contains(old_contract.scope_ref, replacement.scope_ref)
+                    and replacement.scope_ref != old_contract.scope_ref
                 )
-                or set(new_contract.write_scope) < set(old_contract.write_scope)
             ):
                 return
         raise TaskContractError(
@@ -1526,6 +1961,22 @@ def _scope_contains(parent: str, child: str) -> bool:
     return child == parent or child.startswith(f"{parent.rstrip('/')}/")
 
 
+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:
+        return ("script-build://writes/elements",)
+    if kind in {ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO}:
+        return ("script-build://writes",)
+    return ()
+
+
+def _contract_signature(contract: ScriptTaskContractV1) -> str:
+    payload = contract.to_payload()
+    payload.pop("budget", None)
+    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
 def _task_depth(ledger: Any, task: Any) -> int:
     depth = 0
     current = task

script_build_host/tests/test_phase_two_lifecycle.py

@@ -845,17 +845,20 @@ def test_mission_observation_dto_bounds_text_and_omits_internal_payloads() -> No
         version="1",
         digest="sha256:" + "a" * 64,
     )
+    spec = SimpleNamespace(
+        version=1,
+        objective="candidate body " * 500,
+        acceptance_criteria=[],
+        context_refs=("protected://" + secret,),
+        created_at="2026-07-19T00:00:00+00:00",
+    )
     task = SimpleNamespace(
         task_id="task-1",
         parent_task_id="root",
         display_path="root/task-1",
         status=TaskStatus.COMPLETED,
-        current_spec=SimpleNamespace(
-            version=1,
-            objective="candidate body " * 500,
-            acceptance_criteria=[],
-            context_refs=("protected://" + secret,),
-        ),
+        current_spec=spec,
+        specs=[spec],
         child_task_ids=[],
         attempt_ids=["attempt-1"],
         validation_ids=[],
@@ -871,6 +874,7 @@ def test_mission_observation_dto_bounds_text_and_omits_internal_payloads() -> No
         spec_version=1,
         worker_trace_id="worker",
         worker_preset="script_paragraph_worker",
+        execution_mode="new",
         status="submitted",
         operation_id="operation-1",
         snapshot_id="snapshot-1",
@@ -881,6 +885,14 @@ def test_mission_observation_dto_bounds_text_and_omits_internal_payloads() -> No
             evidence_refs=[],
         ),
         error=None,
+        execution_stats=SimpleNamespace(
+            primary_model="qwen-test",
+            total_tokens=321,
+            total_cost=0.25,
+            failure_code=None,
+        ),
+        started_at="2026-07-19T00:00:00+00:00",
+        completed_at="2026-07-19T00:00:02+00:00",
         created_at="now",
         updated_at="now",
         protected_context={"token": secret},
@@ -897,6 +909,35 @@ def test_mission_observation_dto_bounds_text_and_omits_internal_payloads() -> No
         created_at="now",
         payload={"tool_arguments": secret},
     )
+    validation = SimpleNamespace(
+        validation_id="validation-1",
+        task_id="task-1",
+        attempt_id="attempt-1",
+        spec_version=1,
+        snapshot_id="snapshot-1",
+        validator_trace_id="validator-1",
+        validator_preset="script_candidate_validator",
+        validation_plan=SimpleNamespace(
+            mode="deterministic",
+            preflight_rule_ids=("artifact-digest", "goal-coverage"),
+        ),
+        status="completed",
+        verdict="passed",
+        criterion_results=[],
+        summary="safe",
+        recommendation="accept",
+        error=None,
+        execution_stats=SimpleNamespace(
+            primary_model=None,
+            total_tokens=0,
+            total_cost=0.0,
+            failure_code=None,
+        ),
+        started_at="2026-07-19T00:00:02+00:00",
+        completed_at="2026-07-19T00:00:03+00:00",
+        created_at="now",
+        updated_at="now",
+    )
     operation = SimpleNamespace(
         operation_id="operation-1",
         kind="dispatch",
@@ -906,6 +947,8 @@ def test_mission_observation_dto_bounds_text_and_omits_internal_payloads() -> No
         validation_ids=[],
         deadline_at=None,
         error=None,
+        started_at="2026-07-19T00:00:00+00:00",
+        completed_at="2026-07-19T00:00:03+00:00",
         created_at="now",
         updated_at="now",
         request={"arguments": secret},
@@ -918,7 +961,7 @@ def test_mission_observation_dto_bounds_text_and_omits_internal_payloads() -> No
         focused_task_id=None,
         tasks={"task-1": task},
         attempts={"attempt-1": attempt},
-        validations={},
+        validations={"validation-1": validation},
         decisions={"decision-1": decision},
         operations={"operation-1": operation},
         protected_context={"token": secret},
@@ -929,4 +972,13 @@ def test_mission_observation_dto_bounds_text_and_omits_internal_payloads() -> No
     assert secret not in rendered
     assert view["tasks"][0]["current_spec"]["context_refs"] == ["[internal-ref]"]
     assert len(view["tasks"][0]["current_spec"]["objective"]) == 2_000
+    assert view["tasks"][0]["specs"][0]["version"] == 1
     assert len(view["attempts"][0]["submission"]["summary"]) == 500
+    assert view["attempts"][0]["duration_ms"] == 2_000
+    assert view["attempts"][0]["execution_stats"]["primary_model"] == "qwen-test"
+    assert view["validations"][0]["validation_plan"] == {
+        "mode": "deterministic",
+        "preflight_rule_ids": ["artifact-digest", "goal-coverage"],
+    }
+    assert view["validations"][0]["duration_ms"] == 1_000
+    assert view["operations"][0]["duration_ms"] == 3_000