test(阶段合同): 参数化覆盖阶段守卫与 accepted input 规则

新增 PLAN/REVISE/SPLIT/DISPATCH/BLOCK 的阶段不匹配用例，并覆盖同作用域、祖先后代、Structure/Paragraph、显式 ElementSet placement 与 direct-child 拒绝组合。同步验证违规发生在 Worker 派发前。

script_build_host/tests/test_phase_two_contracts.py

@@ -27,6 +27,7 @@ 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 (
+    PhasePolicyGuard,
     PhaseTwoPlanningService,
     _execution_seconds,
 )
@@ -128,6 +129,144 @@ def _contract(
     return payload
 
 
+def test_deep_control_uri_is_not_mistaken_for_base64_content() -> None:
+    scope = (
+        "script-build://mission/900032/direction/main/portfolio/main/compose/main/"
+        "structure/main/paragraph/slot1"
+    )
+    contract = ScriptTaskContractV1.from_payload(_contract("paragraph", scope=scope))
+    assert contract.scope_ref == scope
+
+    encoded = _contract("paragraph", scope="script-build://scopes/" + "A" * 80)
+    with pytest.raises(TaskContractError, match="encoded candidate content"):
+        ScriptTaskContractV1.from_payload(encoded)
+
+
+def test_phase_policy_rejects_workspace_contract_without_capability_write_scope() -> None:
+    guard = PhasePolicyGuard()
+    ledger = SimpleNamespace(root_task_id="root")
+    parent = SimpleNamespace(
+        task_id="portfolio",
+        current_spec=SimpleNamespace(
+            context_refs=("script-build://task-kinds/candidate-portfolio",)
+        ),
+    )
+    invalid = ScriptTaskContractV1.from_payload(_contract("structure"))
+    with pytest.raises(TaskContractError, match="WRITE_SCOPE_VIOLATION"):
+        guard.contracts(
+            context={"phase": 2}, ledger=ledger, parent=parent, contracts=(invalid,)
+        )
+
+    payload = _contract("structure")
+    payload["write_scope"] = ["script-build://writes/paragraphs/main"]
+    valid = ScriptTaskContractV1.from_payload(payload)
+    guard.contracts(context={"phase": 2}, ledger=ledger, parent=parent, contracts=(valid,))
+
+
+def test_phase_policy_requires_broad_container_write_capability() -> None:
+    guard = PhasePolicyGuard()
+    ledger = SimpleNamespace(root_task_id="root")
+    root = SimpleNamespace(task_id="root", current_spec=SimpleNamespace(context_refs=()))
+    invalid = ScriptTaskContractV1.from_payload(
+        _contract("candidate-portfolio", execution_ready=False)
+    )
+    with pytest.raises(TaskContractError, match="WRITE_SCOPE_VIOLATION"):
+        guard.contracts(
+            context={"phase": 2}, ledger=ledger, parent=root, contracts=(invalid,)
+        )
+
+    payload = _contract("candidate-portfolio", execution_ready=False)
+    payload["write_scope"] = ["script-build://writes"]
+    valid = ScriptTaskContractV1.from_payload(payload)
+    guard.contracts(context={"phase": 2}, ledger=ledger, parent=root, contracts=(valid,))
+
+
+def test_phase_policy_requires_structure_base_and_compose_coverage() -> None:
+    guard = PhasePolicyGuard()
+    structure_ref = {
+        "decision_id": "structure-decision",
+        "artifact_ref": {
+            "uri": "script-build://artifact-versions/10",
+            "kind": "structure",
+            "version": "10",
+            "digest": "sha256:" + "a" * 64,
+        },
+        "scope_ref": "script-build://scopes/full/structure",
+        "expected_task_kind": "structure",
+    }
+    paragraph_ref = {
+        "decision_id": "paragraph-decision",
+        "artifact_ref": {
+            "uri": "script-build://artifact-versions/11",
+            "kind": "paragraph",
+            "version": "11",
+            "digest": "sha256:" + "b" * 64,
+        },
+        "scope_ref": "script-build://scopes/full/structure/paragraph",
+        "expected_task_kind": "paragraph",
+    }
+
+    paragraph = _contract("paragraph")
+    paragraph["write_scope"] = ["script-build://writes/paragraphs/main"]
+
+    missing_structure_input = ScriptTaskContractV1.from_payload(paragraph)
+    with pytest.raises(TaskContractError, match="accepted Structure decision"):
+        guard.revisions(context={"phase": 2}, contracts=(missing_structure_input,))
+
+    paragraph["input_decision_refs"] = [structure_ref]
+    invalid_paragraph = ScriptTaskContractV1.from_payload(paragraph)
+    with pytest.raises(TaskContractError, match="base_artifact_ref"):
+        guard.revisions(context={"phase": 2}, contracts=(invalid_paragraph,))
+
+    paragraph["base_artifact_ref"] = {
+        **structure_ref["artifact_ref"],
+        "digest": "sha256:" + "f" * 64,
+    }
+    reconstructed_paragraph = ScriptTaskContractV1.from_payload(paragraph)
+    with pytest.raises(TaskContractError, match="unchanged"):
+        guard.revisions(context={"phase": 2}, contracts=(reconstructed_paragraph,))
+
+    paragraph["base_artifact_ref"] = structure_ref["artifact_ref"]
+    valid_paragraph = ScriptTaskContractV1.from_payload(paragraph)
+    guard.revisions(context={"phase": 2}, contracts=(valid_paragraph,))
+
+    compose = _contract("compose", execution_ready=False)
+    compose["write_scope"] = ["script-build://writes"]
+    guard.revisions(
+        context={"phase": 2},
+        contracts=(ScriptTaskContractV1.from_payload(compose),),
+    )
+    compose["candidate_closure_decision_refs"] = [paragraph_ref]
+    compose["adopted_decision_ids"] = ["paragraph-decision"]
+    compose["compose_order"] = ["paragraph-decision"]
+    with pytest.raises(TaskContractError, match="accepted Direction"):
+        guard.revisions(
+            context={"phase": 2},
+            contracts=(ScriptTaskContractV1.from_payload(compose),),
+        )
+    direction_ref = {
+        "decision_id": "direction-decision",
+        "artifact_ref": {
+            "uri": "script-build://artifact-versions/9",
+            "kind": "direction",
+            "version": "9",
+            "digest": "sha256:" + "d" * 64,
+        },
+        "scope_ref": "script-build://scopes/full",
+        "expected_task_kind": "direction",
+    }
+    compose["input_decision_refs"] = [direction_ref]
+    invalid_compose = ScriptTaskContractV1.from_payload(compose)
+    with pytest.raises(TaskContractError, match="covering Structure"):
+        guard.revisions(context={"phase": 2}, contracts=(invalid_compose,))
+
+    compose["candidate_closure_decision_refs"] = [structure_ref, paragraph_ref]
+    compose["adopted_decision_ids"] = ["structure-decision", "paragraph-decision"]
+    compose["compose_order"] = ["structure-decision", "paragraph-decision"]
+    valid_compose = ScriptTaskContractV1.from_payload(compose)
+    guard.revisions(context={"phase": 2}, contracts=(valid_compose,))
+
+
 class _Bindings:
     def __init__(self, root: str) -> None:
         self.binding = MissionBinding(
@@ -364,7 +503,7 @@ async def test_accepted_input_scope_is_rejected_before_worker_dispatch(tmp_path:
 
     service.contract_for_task = frozen_contract  # type: ignore[method-assign]
 
-    with pytest.raises(TaskContractError, match="paragraph scope must equal or nest within"):
+    with pytest.raises(TaskContractError, match="INPUT_SCOPE_MISMATCH"):
         await service._guard_accepted_input_scopes(root, ledger, (paragraph,))
 
 
@@ -652,6 +791,19 @@ 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)
+    with pytest.raises(TaskContractError, match="PHASE_POLICY_VIOLATION"):
+        await service.plan_script_tasks(
+            contract_payloads=[_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]
+
+
 @pytest.mark.asyncio
 async def test_invalid_contract_batch_creates_zero_tasks(tmp_path: Path) -> None:
     service, coordinator, root = await _service(tmp_path)