feat(阶段守卫): 集中约束 PLAN REVISE SPLIT DISPATCH 与边界

新增 PhasePolicyGuard，以 Host 注入的 phase 为唯一阶段依据。在创建任务、修订、拆分、派发和 Root BLOCK 前统一阻断跨阶段行为，并补充 Paragraph 基础 Structure、Compose Direction/闭包、Portfolio Direction 作用域等执行前校验。

script_build_host/src/script_build_host/application/phase_two_planning.py

@@ -19,6 +19,11 @@ from agent.orchestration import (
 )
 
 from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.input_compatibility import (
+    AcceptedInputCompatibilityError,
+    AcceptedInputSource,
+    validate_accepted_input,
+)
 from script_build_host.domain.phase_two_ports import ScriptTaskContractStore
 from script_build_host.domain.ports import MissionBindingRepository
 from script_build_host.domain.task_contracts import (
@@ -125,6 +130,207 @@ class PlannedContract:
     draft: dict[str, Any]
 
 
+class PhasePolicyGuard:
+    """Enforce Host-protected phase ownership before framework mutations."""
+
+    @staticmethod
+    def phase(context: Mapping[str, Any]) -> int | None:
+        value = context.get("phase")
+        if value is None:
+            # Direct application-service callers created before protected phase
+            # context remain supported. Agent runs always carry this value.
+            return None
+        if isinstance(value, bool) or value not in {1, 2, 3}:
+            raise TaskContractError("PHASE_POLICY_VIOLATION", "protected phase is invalid")
+        return int(value)
+
+    def contracts(
+        self,
+        *,
+        context: Mapping[str, Any],
+        ledger: Any,
+        parent: Any,
+        contracts: Sequence[ScriptTaskContractV1],
+    ) -> None:
+        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 contracts}
+        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 == 2:
+            self._phase_two_contracts(contracts)
+
+    def dispatch(self, *, context: Mapping[str, Any], kind: ScriptTaskKind) -> None:
+        phase = self.phase(context)
+        if phase is None:
+            return
+        allowed = {
+            1: {ScriptTaskKind.DIRECTION, *(ScriptTaskKind(item) for item in RETRIEVAL_KINDS)},
+            2: set(_PHASE_TWO_KINDS),
+            3: {ScriptTaskKind.ROOT_DELIVERY},
+        }[phase]
+        if kind not in allowed:
+            raise TaskContractError(
+                "PHASE_POLICY_VIOLATION",
+                f"phase {phase} cannot dispatch {kind.value}",
+            )
+
+    def revisions(
+        self, *, context: Mapping[str, Any], contracts: Sequence[ScriptTaskContractV1]
+    ) -> None:
+        phase = self.phase(context)
+        if phase is None:
+            return
+        allowed = {
+            1: {ScriptTaskKind.DIRECTION, *(ScriptTaskKind(item) for item in RETRIEVAL_KINDS)},
+            2: set(_PHASE_TWO_KINDS),
+            3: set(),
+        }[phase]
+        kinds = {item.task_kind for item in contracts}
+        if not kinds <= allowed:
+            raise TaskContractError(
+                "PHASE_POLICY_VIOLATION",
+                f"phase {phase} cannot revise {sorted(item.value for item in kinds)}",
+            )
+        if phase == 2:
+            self._phase_two_contracts(contracts)
+
+    @staticmethod
+    def _phase_two_contracts(contracts: Sequence[ScriptTaskContractV1]) -> None:
+        container_kinds = {
+            ScriptTaskKind.CANDIDATE_PORTFOLIO,
+            ScriptTaskKind.COMPOSE,
+        }
+        required_write_prefixes = {
+            ScriptTaskKind.STRUCTURE: "script-build://writes/paragraphs/",
+            ScriptTaskKind.PARAGRAPH: "script-build://writes/paragraphs/",
+            ScriptTaskKind.ELEMENT_SET: "script-build://writes/elements/",
+        }
+        for contract in contracts:
+            adoption_started = bool(
+                contract.candidate_closure_decision_refs
+                or contract.adopted_decision_ids
+                or contract.held_or_rejected_decision_ids
+                or contract.compose_order
+            )
+            if (
+                contract.task_kind
+                in {ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO}
+                and adoption_started
+                and not contract.execution_ready
+            ):
+                raise TaskContractError(
+                    "TASK_CONTRACT_NOT_EXECUTABLE",
+                    "an adoption revision must close every candidate and set compose_order "
+                    "to the adopted Decision IDs",
+                )
+            if (
+                contract.task_kind in container_kinds
+                and "script-build://writes" not in contract.write_scope
+            ):
+                raise TaskContractError(
+                    "WRITE_SCOPE_VIOLATION",
+                    f"{contract.task_kind.value} requires write_scope "
+                    "script-build://writes so child capabilities remain contained",
+                )
+            prefix = required_write_prefixes.get(contract.task_kind)
+            if prefix is not None and not any(
+                value == prefix.removesuffix("/") or value.startswith(prefix)
+                for value in contract.write_scope
+            ):
+                raise TaskContractError(
+                    "WRITE_SCOPE_VIOLATION",
+                    f"{contract.task_kind.value} requires write_scope under {prefix}",
+                )
+            if contract.task_kind is ScriptTaskKind.PARAGRAPH:
+                structures = tuple(
+                    item
+                    for item in contract.input_decision_refs
+                    if item.expected_task_kind is ScriptTaskKind.STRUCTURE
+                )
+                if not structures:
+                    raise TaskContractError(
+                        "INPUT_SCOPE_MISMATCH",
+                        "Paragraph requires the accepted Structure decision in "
+                        "input_decision_refs",
+                    )
+                if (
+                    contract.base_artifact_ref is None
+                    or all(
+                        contract.base_artifact_ref != item.artifact_ref
+                        for item in structures
+                    )
+                ):
+                    raise TaskContractError(
+                        "INPUT_SCOPE_MISMATCH",
+                        "Paragraph must copy the accepted Structure artifact_ref unchanged "
+                        "into base_artifact_ref",
+                    )
+            if contract.task_kind is ScriptTaskKind.COMPOSE:
+                directions = tuple(
+                    item
+                    for item in contract.input_decision_refs
+                    if item.expected_task_kind is ScriptTaskKind.DIRECTION
+                )
+                if contract.execution_ready and len(directions) != 1:
+                    raise TaskContractError(
+                        "CHILD_DECISION_INVALID",
+                        "Compose requires exactly one accepted Direction in "
+                        "input_decision_refs",
+                    )
+                by_id = {
+                    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
+                )
+                if any(
+                    item.expected_task_kind is ScriptTaskKind.PARAGRAPH for item in adopted
+                ) and not any(
+                    item.expected_task_kind is ScriptTaskKind.STRUCTURE for item in adopted
+                ):
+                    raise TaskContractError(
+                        "INPUT_SCOPE_MISMATCH",
+                        "Compose adopting Paragraph must also adopt its covering Structure",
+                    )
+
+    def boundary(self, *, context: Mapping[str, Any], reason: str) -> None:
+        phase = self.phase(context)
+        if phase is None:
+            return
+        expected = {1: PHASE_ONE_CAPABILITY_BOUNDARY, 2: PHASE_TWO_CANDIDATE_PORTFOLIO_READY}
+        if phase not in expected or reason != expected[phase]:
+            raise TaskContractError(
+                "PHASE_POLICY_VIOLATION",
+                f"phase {phase} cannot submit Root boundary {reason}",
+            )
+
+
 class PhaseTwoPlanningService:
     """Controlled Planner surface for Phase1 and Phase2 Tasks."""
 
@@ -146,6 +352,7 @@ class PhaseTwoPlanningService:
         self.closure_gate = closure_gate
         self.workspace_lifecycle = workspace_lifecycle
         self.limits = limits or PhaseTwoLimits()
+        self.phase_policy = PhasePolicyGuard()
 
     async def plan_script_tasks(
         self,
@@ -181,6 +388,12 @@ class PhaseTwoPlanningService:
 
         # Parse every value before writing any content-addressed blob.
         parsed_contracts = tuple(ScriptTaskContractV1.from_payload(value) for value in supplied)
+        self.phase_policy.contracts(
+            context=context, ledger=ledger, parent=parent, contracts=parsed_contracts
+        )
+        await self._guard_phase_two_direction_scope(
+            context, root_trace_id, ledger, parsed_contracts
+        )
         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)
@@ -293,11 +506,15 @@ class PhaseTwoPlanningService:
                     "TASK_CONTRACT_INVALID", "REVISE requires one complete replacement contract"
                 )
             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"
                 )
+            await self._guard_phase_two_direction_scope(
+                context, root_trace_id, ledger, (contract,)
+            )
             await self._guard_supersessions(root_trace_id, ledger, (contract,))
             await self._guard_accepted_input_scopes(root_trace_id, ledger, (contract,))
             planned_replacement = (
@@ -310,6 +527,12 @@ class PhaseTwoPlanningService:
                     "TASK_CONTRACT_INVALID", "SPLIT requires complete child contracts"
                 )
             contracts = tuple(ScriptTaskContractV1.from_payload(value) for value in child_contracts)
+            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_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)
@@ -324,6 +547,7 @@ class PhaseTwoPlanningService:
             )
 
         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:
                 if not phase_one_ready:
                     raise TaskContractError(
@@ -420,6 +644,7 @@ 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)
             if len(task.attempt_ids) >= min(
                 contract.budget.max_attempts, self.limits.max_attempts_per_task
             ):
@@ -742,6 +967,50 @@ class PhaseTwoPlanningService:
                         "replacement and superseded scopes are incompatible",
                     )
 
+    async def _guard_phase_two_direction_scope(
+        self,
+        context: Mapping[str, Any],
+        root_trace_id: str,
+        ledger: Any,
+        contracts: Sequence[ScriptTaskContractV1],
+    ) -> None:
+        """Keep the Phase2 business tree inside its accepted Direction scope."""
+
+        if self.phase_policy.phase(context) != 2 or not any(
+            item.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO for item in contracts
+        ):
+            return
+        direction_scopes: list[str] = []
+        for task in ledger.tasks.values():
+            if (
+                task_kind_from_context_refs(task.current_spec.context_refs)
+                != ScriptTaskKind.DIRECTION.value
+                or task.status is not TaskStatus.COMPLETED
+                or not task.decision_ids
+            ):
+                continue
+            decision = ledger.decisions.get(task.decision_ids[-1])
+            if decision is None or decision.action is not DecisionAction.ACCEPT:
+                continue
+            direction_scopes.append(
+                (await self.contract_for_task(root_trace_id, task)).contract.scope_ref
+            )
+        if len(direction_scopes) != 1:
+            raise TaskContractError(
+                "PHASE_POLICY_VIOLATION",
+                "Phase2 requires exactly one accepted Direction scope",
+            )
+        direction_scope = direction_scopes[0]
+        if any(
+            item.task_kind is ScriptTaskKind.CANDIDATE_PORTFOLIO
+            and not _scope_contains(direction_scope, item.scope_ref)
+            for item in contracts
+        ):
+            raise TaskContractError(
+                "INPUT_SCOPE_MISMATCH",
+                "CandidatePortfolio scope must stay below the accepted Direction scope",
+            )
+
     async def _guard_accepted_input_scopes(
         self,
         root_trace_id: str,
@@ -790,15 +1059,19 @@ class PhaseTwoPlanningService:
                         "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)
-                ):
+                try:
+                    validate_accepted_input(
+                        consumer_kind=consumer.task_kind,
+                        consumer_scope=consumer.scope_ref,
+                        producer_kind=producer.task_kind,
+                        producer_scope=producer.scope_ref,
+                        source=AcceptedInputSource.EXPLICIT,
+                    )
+                except AcceptedInputCompatibilityError as exc:
                     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}",
-                    )
+                        str(exc),
+                    ) from exc
 
     async def _guard_runtime_budget(
         self, ledger: Any, *, external_query_mode: str = "strict"