规划守卫：在 Worker 前校验 Phase 2 Goal 闭包

PhaseTwoPlanningService 读取 active Direction，并在 PLAN、REVISE、SPLIT 与 DISPATCH 前统一校验空值、未知 Goal、父子 Task 子集、Compare 并集和 Direction 引用。

允许无 base 的 Paragraph-first 探索；一旦作为 Structure patch，仍必须固定对应的已接受 Structure。

script_build_host/src/script_build_host/application/phase_two_planning.py

@@ -18,14 +18,19 @@ from agent.orchestration import (
     TaskStatus,
 )
 
+from script_build_host.domain.artifacts import DirectionArtifact
 from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.goal_coverage import validate_goal_ids
 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.ports import (
+    MissionBindingRepository,
+    ScriptBusinessArtifactRepository,
+)
 from script_build_host.domain.task_contracts import (
     FrozenTaskContract,
     PhaseTwoLimits,
@@ -179,6 +184,10 @@ class PhasePolicyGuard:
                 "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 contracts):
+            raise TaskContractError(
+                "GOAL_SCOPE_INVALID", "Phase1 contracts must use an empty goal_ids array"
+            )
         if phase == 2:
             self._phase_two_contracts(contracts)
 
@@ -214,6 +223,10 @@ class PhasePolicyGuard:
                 "PHASE_POLICY_VIOLATION",
                 f"phase {phase} cannot revise {sorted(item.value for item in kinds)}",
             )
+        if phase == 1 and any(item.goal_ids for item in contracts):
+            raise TaskContractError(
+                "GOAL_SCOPE_INVALID", "Phase1 contracts must use an empty goal_ids array"
+            )
         if phase == 2:
             self._phase_two_contracts(contracts)
 
@@ -270,22 +283,22 @@ class PhasePolicyGuard:
                     for item in contract.input_decision_refs
                     if item.expected_task_kind is ScriptTaskKind.STRUCTURE
                 )
-                if not structures:
+                if contract.base_artifact_ref is not None and not structures:
                     raise TaskContractError(
                         "INPUT_SCOPE_MISMATCH",
-                        "Paragraph requires the accepted Structure decision in "
+                        "Paragraph patch requires the accepted Structure decision in "
                         "input_decision_refs",
                     )
                 if (
-                    contract.base_artifact_ref is None
-                    or all(
+                    contract.base_artifact_ref is not None
+                    and all(
                         contract.base_artifact_ref != item.artifact_ref
                         for item in structures
                     )
                 ):
                     raise TaskContractError(
                         "INPUT_SCOPE_MISMATCH",
-                        "Paragraph must copy the accepted Structure artifact_ref unchanged "
+                        "Paragraph patch must copy the accepted Structure artifact_ref unchanged "
                         "into base_artifact_ref",
                     )
             if contract.task_kind is ScriptTaskKind.COMPOSE:
@@ -340,6 +353,7 @@ class PhaseTwoPlanningService:
         coordinator: Any,
         bindings: MissionBindingRepository,
         contracts: ScriptTaskContractStore,
+        artifacts: ScriptBusinessArtifactRepository,
         dispatch_guard: DispatchGuard | None = None,
         closure_gate: PhaseTwoClosureGate | None = None,
         workspace_lifecycle: AttemptWorkspaceLifecycle | None = None,
@@ -348,6 +362,7 @@ class PhaseTwoPlanningService:
         self.coordinator = coordinator
         self.bindings = bindings
         self.contracts = contracts
+        self.artifacts = artifacts
         self.dispatch_guard = dispatch_guard
         self.closure_gate = closure_gate
         self.workspace_lifecycle = workspace_lifecycle
@@ -394,6 +409,14 @@ class PhaseTwoPlanningService:
         await self._guard_phase_two_direction_scope(
             context, root_trace_id, ledger, parsed_contracts
         )
+        await self._guard_phase_two_goals(
+            context=context,
+            binding=binding,
+            root_trace_id=root_trace_id,
+            ledger=ledger,
+            contracts=parsed_contracts,
+            parent=parent,
+        )
         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)
@@ -515,6 +538,15 @@ class PhaseTwoPlanningService:
             await self._guard_phase_two_direction_scope(
                 context, root_trace_id, ledger, (contract,)
             )
+            parent = ledger.tasks.get(task.parent_task_id or "")
+            await self._guard_phase_two_goals(
+                context=context,
+                binding=binding,
+                root_trace_id=root_trace_id,
+                ledger=ledger,
+                contracts=(contract,),
+                parent=parent,
+            )
             await self._guard_supersessions(root_trace_id, ledger, (contract,))
             await self._guard_accepted_input_scopes(root_trace_id, ledger, (contract,))
             planned_replacement = (
@@ -533,6 +565,14 @@ class PhaseTwoPlanningService:
             await self._guard_phase_two_direction_scope(
                 context, root_trace_id, ledger, contracts
             )
+            await self._guard_phase_two_goals(
+                context=context,
+                binding=binding,
+                root_trace_id=root_trace_id,
+                ledger=ledger,
+                contracts=contracts,
+                parent=task,
+            )
             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)
@@ -645,6 +685,15 @@ 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)
+            parent = ledger.tasks.get(task.parent_task_id or "")
+            await self._guard_phase_two_goals(
+                context=context,
+                binding=binding,
+                root_trace_id=root_trace_id,
+                ledger=ledger,
+                contracts=(contract,),
+                parent=parent,
+            )
             if len(task.attempt_ids) >= min(
                 contract.budget.max_attempts, self.limits.max_attempts_per_task
             ):
@@ -1011,6 +1060,97 @@ class PhaseTwoPlanningService:
                 "CandidatePortfolio scope must stay below the accepted Direction scope",
             )
 
+    async def _guard_phase_two_goals(
+        self,
+        *,
+        context: Mapping[str, Any],
+        binding: Any,
+        root_trace_id: str,
+        ledger: Any,
+        contracts: Sequence[ScriptTaskContractV1],
+        parent: Any | None,
+    ) -> None:
+        """Validate Goal scope before any contract is frozen or Worker is dispatched."""
+
+        if self.phase_policy.phase(context) != 2:
+            return
+        direction_version_id = binding.active_direction_artifact_version_id
+        if direction_version_id is None:
+            raise TaskContractError("GOAL_SCOPE_INVALID", "Phase2 has no active Direction")
+        version = await self.artifacts.get_by_id(
+            direction_version_id, script_build_id=binding.script_build_id
+        )
+        direction = version.artifact
+        if not isinstance(direction, DirectionArtifact):
+            raise TaskContractError(
+                "GOAL_SCOPE_INVALID", "active Direction artifact has the wrong type"
+            )
+        direction_goal_ids = tuple(item.goal_id for item in direction.goals)
+        direction_uri = f"script-build://artifact-versions/{direction_version_id}"
+
+        parent_goal_ids: tuple[str, ...] | None = None
+        if parent is not None and parent.task_id != ledger.root_task_id:
+            parent_goal_ids = (
+                await self.contract_for_task(root_trace_id, parent)
+            ).contract.goal_ids
+
+        for contract in contracts:
+            require_all = contract.task_kind in {
+                ScriptTaskKind.COMPOSE,
+                ScriptTaskKind.CANDIDATE_PORTFOLIO,
+            }
+            validate_goal_ids(
+                contract.goal_ids,
+                allowed_goal_ids=direction_goal_ids,
+                require_all=require_all,
+            )
+            if parent_goal_ids is not None and not set(contract.goal_ids) <= set(parent_goal_ids):
+                raise TaskContractError(
+                    "GOAL_SCOPE_INVALID", "child Task goal_ids must be a subset of its parent"
+                )
+
+            if contract.task_kind not in {
+                ScriptTaskKind.PATTERN_RETRIEVAL,
+                ScriptTaskKind.DECODE_RETRIEVAL,
+                ScriptTaskKind.EXTERNAL_RETRIEVAL,
+                ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
+            }:
+                direction_refs = tuple(
+                    item
+                    for item in contract.input_decision_refs
+                    if item.expected_task_kind is ScriptTaskKind.DIRECTION
+                )
+                if len(direction_refs) != 1 or direction_refs[0].artifact_ref.uri != direction_uri:
+                    raise TaskContractError(
+                        "GOAL_SCOPE_INVALID",
+                        "Phase2 creative Tasks must pin the active accepted Direction",
+                    )
+
+            if contract.task_kind is ScriptTaskKind.COMPARE:
+                compared_goal_ids: set[str] = set()
+                for item in contract.comparison_decision_refs:
+                    decision = ledger.decisions.get(item.decision_id)
+                    attempt = (
+                        ledger.attempts.get(decision.attempt_id)
+                        if decision is not None and decision.attempt_id is not None
+                        else None
+                    )
+                    producer = (
+                        ledger.tasks.get(attempt.task_id) if attempt is not None else None
+                    )
+                    if producer is None:
+                        raise TaskContractError(
+                            "GOAL_SCOPE_INVALID", "Compare references an unknown candidate Task"
+                        )
+                    compared_goal_ids.update(
+                        (await self.contract_for_task(root_trace_id, producer)).contract.goal_ids
+                    )
+                if set(contract.goal_ids) != compared_goal_ids:
+                    raise TaskContractError(
+                        "GOAL_SCOPE_INVALID",
+                        "Compare goal_ids must equal the union of compared candidates",
+                    )
+
     async def _guard_accepted_input_scopes(
         self,
         root_trace_id: str,

script_build_host/src/script_build_host/composition.py

@@ -181,6 +181,7 @@ def compose_host(dependencies: HostDependencies) -> HostComposition:
         coordinator=coordinator,
         bindings=dependencies.bindings,
         contracts=contract_store,
+        artifacts=dependencies.business_artifacts,
         closure_gate=boundary_verifier,
         limits=dependencies.phase_two_limits,
     )