fix(Phase2编排): 前置阻断无效闭包与错误容器状态

在 PLAN、REVISE、SPLIT、DISPATCH 四个边界校验 accepted input 的 Task kind 与 scope，避免必失败 Worker 被启动。

补齐 BLOCKED 子任务提示、唯一 Portfolio 保护、accepted_decision_ref 投影，并按实际 Worker/Validator 区间统计执行预算。

script_build_host/src/script_build_host/application/phase_two_planning.py

@@ -184,6 +184,7 @@ class PhaseTwoPlanningService:
         await self._guard_task_budget(root_trace_id, ledger, parsed_contracts, parent)
         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)
         planned = await self._freeze_all(
             root_trace_id, str(binding.input_snapshot_id), parsed_contracts
         )
@@ -252,11 +253,33 @@ class PhaseTwoPlanningService:
                 "accepted history is replaced by a new Task with supersedes_decision_ids",
             )
         if not is_root:
+            current_contract = await self.contract_for_task(root_trace_id, task)
+            if (
+                decision_action is DecisionAction.CANCEL
+                and current_contract.contract.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO
+            ):
+                raise TaskContractError(
+                    "TASK_CONTRACT_INVALID",
+                    "the unique CandidatePortfolio container cannot be cancelled; "
+                    "complete or revise its Compose children",
+                )
             await self._guard_repair_and_improvement(
                 root_trace_id, ledger, task, decision_action, replacement_contract
             )
 
-        payload: dict[str, Any] = {"reason": _bounded(reason, 1_000, "reason")}
+        bounded_reason = _bounded(reason, 1_000, "reason")
+        phase_one_ready = is_root and _phase_one_boundary_ready(ledger, task)
+        if (
+            decision_action is DecisionAction.BLOCK
+            and phase_one_ready
+            and bounded_reason
+            not in {
+                PHASE_ONE_CAPABILITY_BOUNDARY,
+                PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
+            }
+        ):
+            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:
                 raise TaskContractError(
@@ -276,6 +299,7 @@ class PhaseTwoPlanningService:
                     "TASK_KIND_PRESET_MISMATCH", "REVISE cannot change the Task kind"
                 )
             await self._guard_supersessions(root_trace_id, ledger, (contract,))
+            await self._guard_accepted_input_scopes(root_trace_id, ledger, (contract,))
             planned_replacement = (
                 await self._freeze_all(root_trace_id, str(binding.input_snapshot_id), (contract,))
             )[0]
@@ -289,6 +313,7 @@ class PhaseTwoPlanningService:
             await self._guard_task_budget(root_trace_id, ledger, contracts, task)
             await self._guard_hierarchy(root_trace_id, ledger, task, contracts)
             await self._guard_supersessions(root_trace_id, ledger, contracts)
+            await self._guard_accepted_input_scopes(root_trace_id, ledger, contracts)
             planned_children = await self._freeze_all(
                 root_trace_id, str(binding.input_snapshot_id), contracts
             )
@@ -299,9 +324,13 @@ class PhaseTwoPlanningService:
             )
 
         if decision_action is DecisionAction.BLOCK and is_root:
-            if reason == PHASE_ONE_CAPABILITY_BOUNDARY:
-                pass
-            elif reason != PHASE_TWO_CANDIDATE_PORTFOLIO_READY:
+            if bounded_reason == PHASE_ONE_CAPABILITY_BOUNDARY:
+                if not phase_one_ready:
+                    raise TaskContractError(
+                        "PHASE_TWO_BOUNDARY_NOT_READY",
+                        "Phase1 requires one current accepted Direction and no Portfolio",
+                    )
+            elif bounded_reason != PHASE_TWO_CANDIDATE_PORTFOLIO_READY:
                 raise TaskContractError(
                     "PHASE_TWO_BOUNDARY_NOT_READY",
                     "Root may only block at an explicit phase capability boundary",
@@ -346,11 +375,12 @@ class PhaseTwoPlanningService:
         if (
             is_root
             and decision_action is DecisionAction.BLOCK
-            and reason in {PHASE_ONE_CAPABILITY_BOUNDARY, PHASE_TWO_CANDIDATE_PORTFOLIO_READY}
+            and bounded_reason
+            in {PHASE_ONE_CAPABILITY_BOUNDARY, PHASE_TWO_CANDIDATE_PORTFOLIO_READY}
             and result.get("status") == "blocked"
         ):
             result["terminal_boundary"] = True
-            result["phase_boundary"] = reason
+            result["phase_boundary"] = bounded_reason
         return result
 
     async def dispatch_script_tasks(
@@ -377,6 +407,17 @@ class PhaseTwoPlanningService:
             task = ledger.tasks.get(task_id)
             if task is None:
                 raise TaskContractError("TASK_CONTRACT_INVALID", "Task is outside this Root")
+            blocked_children = [
+                child_id
+                for child_id in task.child_task_ids
+                if ledger.tasks[child_id].status is TaskStatus.BLOCKED
+            ]
+            if task.status is TaskStatus.WAITING_CHILDREN and blocked_children:
+                raise TaskContractError(
+                    "TASK_CHILD_BLOCKED",
+                    "BLOCKED is resumable and does not close a parent; CANCEL each unusable "
+                    f"blocked child or UNBLOCK it before dispatch: {blocked_children}",
+                )
             frozen = await self.contract_for_task(root_trace_id, task)
             contract = frozen.contract
             if len(task.attempt_ids) >= min(
@@ -385,6 +426,7 @@ class PhaseTwoPlanningService:
                 raise TaskContractError("TASK_BUDGET_EXCEEDED", "Task attempt budget is exhausted")
             self._guard_task_usage(ledger, task, contract)
             await self._guard_supersessions(root_trace_id, ledger, (contract,))
+            await self._guard_accepted_input_scopes(root_trace_id, ledger, (contract,))
             contract.require_execution_ready()
             self._guard_retrieval_parent(ledger, task, contract.task_kind)
             preset = TASK_PRESET_BY_KIND[contract.task_kind]
@@ -479,6 +521,39 @@ class PhaseTwoPlanningService:
                 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)
+                attempt = (
+                    ledger.attempts.get(decision.attempt_id)
+                    if decision is not None and decision.attempt_id is not None
+                    else None
+                )
+                refs = (
+                    [*attempt.submission.artifact_refs, *attempt.submission.evidence_refs]
+                    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
+                ):
+                    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,
+                        },
+                        "scope_ref": scope_ref,
+                        "expected_task_kind": kind_value,
+                    }
             tasks.append(
                 {
                     "task_id": task.task_id,
@@ -492,6 +567,7 @@ class PhaseTwoPlanningService:
                     "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}
@@ -666,6 +742,64 @@ class PhaseTwoPlanningService:
                         "replacement and superseded scopes are incompatible",
                     )
 
+    async def _guard_accepted_input_scopes(
+        self,
+        root_trace_id: str,
+        ledger: Any,
+        contracts: Sequence[ScriptTaskContractV1],
+    ) -> None:
+        """Reject accepted-input scopes that can never form a valid frozen closure.
+
+        Full accepted-input resolution still happens when an Attempt is frozen.  This
+        planning guard exists so a malformed contract cannot launch a Worker only to
+        fail deterministically at the Attempt boundary.
+        """
+
+        for consumer in contracts:
+            imported = (
+                *consumer.input_decision_refs,
+                *consumer.candidate_closure_decision_refs,
+                *consumer.comparison_decision_refs,
+            )
+            for imported_ref in imported:
+                decision = ledger.decisions.get(imported_ref.decision_id)
+                if decision is None:
+                    # The authoritative resolver reports forged/missing Decisions at
+                    # Attempt freeze.  Keeping that responsibility there also permits
+                    # contract-only unit fixtures without a fully populated ledger.
+                    continue
+                if decision.action is not DecisionAction.ACCEPT or decision.attempt_id is None:
+                    raise TaskContractError(
+                        "CHILD_DECISION_INVALID",
+                        "an immutable input must reference an ACCEPT decision",
+                    )
+                producer_task = ledger.tasks.get(decision.task_id)
+                if producer_task is None:
+                    raise TaskContractError(
+                        "CHILD_DECISION_INVALID",
+                        "an immutable input references a Task outside this Root",
+                    )
+                producer = (await self.contract_for_task(root_trace_id, producer_task)).contract
+                if imported_ref.expected_task_kind is not producer.task_kind:
+                    raise TaskContractError(
+                        "CHILD_DECISION_INVALID",
+                        "an immutable input's expected Task kind differs from its producer",
+                    )
+                if imported_ref.scope_ref != producer.scope_ref:
+                    raise TaskContractError(
+                        "INPUT_SCOPE_MISMATCH",
+                        "an immutable input's declared scope differs from its producer scope",
+                    )
+                if not (
+                    _scope_contains(consumer.scope_ref, producer.scope_ref)
+                    or _scope_contains(producer.scope_ref, consumer.scope_ref)
+                ):
+                    raise TaskContractError(
+                        "INPUT_SCOPE_MISMATCH",
+                        f"{consumer.task_kind.value} scope must equal or nest within accepted "
+                        f"{producer.task_kind.value} scope {producer.scope_ref}",
+                    )
+
     async def _guard_runtime_budget(
         self, ledger: Any, *, external_query_mode: str = "strict"
     ) -> None:
@@ -683,15 +817,12 @@ class PhaseTwoPlanningService:
         )
         if total_tokens >= self.limits.max_total_tokens:
             raise TaskContractError("TASK_BUDGET_EXCEEDED", "Phase2 token budget is exhausted")
-        started = [
-            _parse_time(ledger.tasks[task_id].created_at)
-            for task_id in phase_two_ids
-            if ledger.tasks[task_id].created_at
+        phase_two_stages = [
+            value
+            for value in (*ledger.attempts.values(), *ledger.validations.values())
+            if value.task_id in phase_two_ids
         ]
-        if (
-            started
-            and (datetime.now(UTC) - min(started)).total_seconds() >= self.limits.max_total_seconds
-        ):
+        if _execution_seconds(phase_two_stages) >= self.limits.max_total_seconds:
             raise TaskContractError("TASK_BUDGET_EXCEEDED", "Phase2 time budget is exhausted")
         retrieval_attempts = sum(
             1
@@ -737,8 +868,7 @@ class PhaseTwoPlanningService:
         )
         if tokens >= min(contract.budget.max_tokens, self.limits.max_task_tokens):
             raise TaskContractError("TASK_BUDGET_EXCEEDED", "Task token budget is exhausted")
-        started = [_parse_time(item.started_at or item.created_at) for item in attempts]
-        if started and (datetime.now(UTC) - min(started)).total_seconds() >= min(
+        if _execution_seconds((*attempts, *validations)) >= min(
             contract.budget.max_seconds, self.limits.max_task_seconds
         ):
             raise TaskContractError("TASK_BUDGET_EXCEEDED", "Task time budget is exhausted")
@@ -944,6 +1074,27 @@ def _required(context: Mapping[str, Any], key: str) -> str:
     return value
 
 
+def _phase_one_boundary_ready(ledger: Any, root: Any) -> bool:
+    accepted_direction = False
+    for child_id in root.child_task_ids:
+        child = ledger.tasks.get(child_id)
+        if child is None:
+            continue
+        kind = task_kind_from_context_refs(child.current_spec.context_refs)
+        if kind == ScriptTaskKind.CANDIDATE_PORTFOLIO.value:
+            return False
+        if kind != ScriptTaskKind.DIRECTION.value 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
+            and child.status is TaskStatus.COMPLETED
+        ):
+            accepted_direction = True
+    return accepted_direction
+
+
 def _optional(context: Mapping[str, Any], key: str) -> str | None:
     value = context.get(key)
     return value if isinstance(value, str) and value else None
@@ -986,6 +1137,22 @@ def _parse_time(value: str) -> datetime:
     return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
 
 
+def _execution_seconds(values: Sequence[Any]) -> float:
+    """Sum actual Worker/Validator stage intervals, excluding Planner idle time."""
+
+    now = datetime.now(UTC)
+    total = 0.0
+    for value in values:
+        started_at = getattr(value, "started_at", None) or getattr(value, "created_at", None)
+        if not started_at:
+            continue
+        completed_at = getattr(value, "completed_at", None)
+        start = _parse_time(started_at)
+        end = _parse_time(completed_at) if completed_at else now
+        total += max(0.0, (end - start).total_seconds())
+    return total
+
+
 def _validation_vector(value: Any) -> tuple[int, int, int]:
     failed = sum(item.verdict.value == "failed" for item in value.criterion_results)
     inconclusive = sum(item.verdict.value == "inconclusive" for item in value.criterion_results)

script_build_host/src/script_build_host/domain/task_contracts.py

@@ -24,8 +24,8 @@ MAX_PHASE_TWO_TASKS = 64
 MAX_TASK_DEPTH = 8
 MAX_CANDIDATES_PER_SCOPE = 4
 MAX_ATTEMPTS_PER_TASK = 4
-MAX_TASK_TOKENS = 32_000
-MAX_PHASE_TWO_TOKENS = 250_000
+MAX_TASK_TOKENS = 1_000_000
+MAX_PHASE_TWO_TOKENS = 10_000_000
 MAX_TASK_SECONDS = 900
 MAX_PHASE_TWO_SECONDS = 3_600
 MAX_EXTERNAL_QUERIES = 40
@@ -395,7 +395,8 @@ class ScriptTaskContractV1:
         if not self.execution_ready:
             raise TaskContractError(
                 "TASK_CONTRACT_NOT_EXECUTABLE",
-                "compose or portfolio contract has not closed its candidate disposition",
+                "first accept child candidates, then REVISE this compose or portfolio with "
+                "their exact accepted_decision_ref values and Decision IDs",
             )
 
     def to_payload(self) -> dict[str, Any]:

script_build_host/tests/test_phase_two_contracts.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import asyncio
 from dataclasses import replace
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from hashlib import sha256
 from pathlib import Path
 from types import SimpleNamespace
@@ -26,7 +26,10 @@ from agent.orchestration.protocols import ValidatorRunResult, WorkerRunResult
 
 from script_build_host.application.mission_factory import ScriptMissionFactory
 from script_build_host.application.mission_service import ScriptMissionService
-from script_build_host.application.phase_two_planning import PhaseTwoPlanningService
+from script_build_host.application.phase_two_planning import (
+    PhaseTwoPlanningService,
+    _execution_seconds,
+)
 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 (
@@ -37,6 +40,22 @@ from script_build_host.domain.task_contracts import (
 from script_build_host.infrastructure.task_contract_store import FileScriptTaskContractStore
 
 
+def test_execution_budget_counts_stage_intervals_not_planner_idle_time() -> None:
+    now = datetime.now(UTC)
+    stages = (
+        SimpleNamespace(
+            started_at=(now - timedelta(hours=2)).isoformat(),
+            completed_at=(now - timedelta(hours=2) + timedelta(seconds=12)).isoformat(),
+        ),
+        SimpleNamespace(
+            started_at=(now - timedelta(minutes=1)).isoformat(),
+            completed_at=(now - timedelta(minutes=1) + timedelta(seconds=8)).isoformat(),
+        ),
+    )
+
+    assert _execution_seconds(stages) == pytest.approx(20.0)
+
+
 def _contract(
     kind: str,
     *,
@@ -250,6 +269,105 @@ async def _service(tmp_path: Path) -> tuple[PhaseTwoPlanningService, TaskCoordin
     return service, coordinator, root
 
 
+@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)
+    direction = await service.plan_script_tasks(
+        contract_payloads=[_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")],
+        parent_task_id=direction_id,
+        context={"root_trace_id": root, "tool_call_id": "retrieval"},
+    )
+    child_id = retrieval["task_ids"][0]
+    await coordinator.decide_task(
+        root,
+        child_id,
+        None,
+        DecisionAction.BLOCK,
+        {"reason": "unusable evidence"},
+        "block-child",
+    )
+
+    with pytest.raises(TaskContractError, match="BLOCKED is resumable"):
+        await service.dispatch_script_tasks(
+            task_ids=[direction_id],
+            context={"root_trace_id": root, "tool_call_id": "dispatch-parent"},
+        )
+
+
+@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=[_contract("candidate-portfolio", execution_ready=False)],
+        parent_task_id=None,
+        context={"root_trace_id": root, "tool_call_id": "portfolio"},
+    )
+
+    with pytest.raises(TaskContractError, match="cannot be cancelled"):
+        await service.decide_script_task(
+            task_id=portfolio["task_ids"][0],
+            action="cancel",
+            reason="wrong structural recovery",
+            validation_id=None,
+            replacement_contract=None,
+            child_contracts=(),
+            context={"root_trace_id": root, "tool_call_id": "cancel-portfolio"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_accepted_input_scope_is_rejected_before_worker_dispatch(tmp_path: Path) -> None:
+    service, _coordinator, root = await _service(tmp_path)
+    structure_scope = "script-build://direction/main/compose/candidate-1/structure/main"
+    structure = ScriptTaskContractV1.from_payload(_contract("structure", scope=structure_scope))
+    paragraph_payload = _contract(
+        "paragraph",
+        scope="script-build://direction/main/compose/candidate-1/paragraph/p1",
+    )
+    accepted_ref = {
+        "decision_id": "accepted-structure",
+        "artifact_ref": {
+            "uri": "script-build://artifact-versions/1",
+            "kind": "structure",
+            "version": "1",
+            "digest": "sha256:" + "a" * 64,
+        },
+        "scope_ref": structure_scope,
+        "expected_task_kind": "structure",
+    }
+    paragraph_payload["input_decision_refs"] = [accepted_ref]
+    paragraph_payload["base_artifact_ref"] = accepted_ref["artifact_ref"]
+    paragraph = ScriptTaskContractV1.from_payload(paragraph_payload)
+    producer_task = SimpleNamespace(task_id="structure-task")
+    ledger = SimpleNamespace(
+        decisions={
+            "accepted-structure": SimpleNamespace(
+                action=DecisionAction.ACCEPT,
+                attempt_id="structure-attempt",
+                task_id=producer_task.task_id,
+            )
+        },
+        tasks={producer_task.task_id: producer_task},
+    )
+
+    async def frozen_contract(_root_trace_id: str, task: Any) -> Any:
+        assert task is producer_task
+        return SimpleNamespace(contract=structure)
+
+    service.contract_for_task = frozen_contract  # type: ignore[method-assign]
+
+    with pytest.raises(TaskContractError, match="paragraph scope must equal or nest within"):
+        await service._guard_accepted_input_scopes(root, ledger, (paragraph,))
+
+
 class _BlockingExecutor:
     def __init__(self) -> None:
         self.started = asyncio.Event()
@@ -593,7 +711,7 @@ async def test_open_compose_container_cannot_dispatch(tmp_path: Path) -> None:
         parent_task_id=portfolio["task_ids"][0],
         context={"root_trace_id": root, "tool_call_id": "c"},
     )
-    with pytest.raises(TaskContractError, match="not closed"):
+    with pytest.raises(TaskContractError, match="first accept child candidates"):
         await service.dispatch_script_tasks(
             task_ids=compose["task_ids"], context={"root_trace_id": root}
         )