script-build-and-framework-tmp/script_build_host/tests/test_phase_two_contracts.py

from __future__ import annotations

import asyncio
from dataclasses import replace
from datetime import UTC, datetime, timedelta
from hashlib import sha256
from pathlib import Path
from types import SimpleNamespace
from typing import Any, cast

import pytest
from agent import AgentRunner, FileSystemArtifactStore, FileSystemTaskStore, FileSystemTraceStore
from agent.orchestration import (
    AgentRole,
    ArtifactRef,
    AttemptSubmission,
    CriterionResult,
    DecisionAction,
    OperationStatus,
    OrchestrationConfig,
    TaskCoordinator,
    TaskStatus,
    ValidationVerdict,
)
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,
)
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 (
    PhaseTwoLimits,
    ScriptTaskContractV1,
    TaskContractError,
)
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,
    *,
    scope: str = "script-build://scopes/full",
    execution_ready: bool = True,
) -> dict[str, object]:
    output = {
        "direction": "script-direction/v1",
        "candidate-portfolio": "candidate-portfolio/v1",
        "compose": "structured-script/v1",
        "paragraph": "paragraph-artifact/v1",
        "element-set": "element-set-artifact/v1",
        "structure": "structure-artifact/v1",
        "compare": "comparison-artifact/v1",
        "decode-retrieval": "evidence-record/v1",
    }[kind]
    payload: dict[str, object] = {
        "schema_version": "script-task-contract/v1",
        "task_kind": kind,
        "scope_ref": scope,
        "intent_class": (
            "portfolio"
            if kind == "candidate-portfolio"
            else "compose"
            if kind == "compose"
            else "explore"
        ),
        "objective": f"produce one bounded {kind} increment",
        "input_decision_refs": [],
        "base_artifact_ref": None,
        "write_scope": [scope],
        "gap_ref": None,
        "output_schema": output,
        "criteria": [
            {
                "criterion_id": "closed",
                "description": "the bounded increment is independently verifiable",
                "hard": True,
            }
        ],
        "budget": {
            "max_attempts": 4,
            "max_tokens": 32000,
            "max_seconds": 900,
            "max_external_queries": 40,
            "max_no_improvement": 3,
        },
        "goal_ids": (
            [] if kind in {"direction", "decode-retrieval"} else ["goal-1"]
        ),
        "supersedes_decision_ids": [],
        "candidate_closure_decision_refs": [],
        "adopted_decision_ids": [],
        "held_or_rejected_decision_ids": [],
        "compose_order": [],
        "comparison_decision_refs": [],
    }
    if execution_ready and kind in {"compose", "candidate-portfolio"}:
        ref = {
            "decision_id": "decision-1",
            "artifact_ref": {
                "uri": "script-build://artifact-versions/1",
                "kind": "structured_script" if kind == "candidate-portfolio" else "paragraph",
                "version": "1",
                "digest": "sha256:" + "a" * 64,
            },
            "scope_ref": scope,
            "expected_task_kind": "compose" if kind == "candidate-portfolio" else "paragraph",
        }
        payload["candidate_closure_decision_refs"] = [ref]
        payload["adopted_decision_ids"] = ["decision-1"]
        payload["compose_order"] = ["decision-1"]
    return payload


def _planner_input(payload: dict[str, object]) -> dict[str, object]:
    input_refs = cast(list[dict[str, Any]], payload.get("input_decision_refs") or [])
    comparison_refs = cast(
        list[dict[str, Any]], payload.get("comparison_decision_refs") or []
    )
    base = cast(dict[str, Any] | None, payload.get("base_artifact_ref"))
    return {
        "task_kind": payload["task_kind"],
        "scope_ref": payload["scope_ref"],
        "intent_class": payload["intent_class"],
        "objective": payload["objective"],
        "goal_ids": payload["goal_ids"],
        "criteria": payload["criteria"],
        "input_decision_ids": [item["decision_id"] for item in input_refs],
        "base_decision_id": next(
            (
                item["decision_id"]
                for item in input_refs
                if base is not None and item["artifact_ref"]["uri"] == base["uri"]
            ),
            None,
        ),
        "comparison_decision_ids": [item["decision_id"] for item in comparison_refs],
        "supersedes_decision_ids": payload["supersedes_decision_ids"],
        "gap_ref": payload["gap_ref"],
    }


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)
    guard.revisions(context={"phase": 2}, contracts=(missing_structure_input,))

    paragraph["base_artifact_ref"] = structure_ref["artifact_ref"]
    invalid_paragraph = ScriptTaskContractV1.from_payload(paragraph)
    with pytest.raises(TaskContractError, match="accepted Structure decision"):
        guard.revisions(context={"phase": 2}, contracts=(invalid_paragraph,))

    paragraph["input_decision_refs"] = [structure_ref]
    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(
            binding_id=1,
            script_build_id=7,
            root_trace_id=root,
            input_snapshot_id=11,
            active_direction_artifact_version_id=None,
            accepted_root_artifact_version_id=None,
            engine_version="test",
            schema_version="v1",
            created_at=datetime.now(UTC),
            updated_at=datetime.now(UTC),
        )

    async def get_by_root(self, root: str) -> MissionBinding:
        assert root == self.binding.root_trace_id
        return self.binding

    async def get_by_build(self, script_build_id: int) -> MissionBinding:
        assert script_build_id == self.binding.script_build_id
        return self.binding


class _PlaceholderThenReplacementExecutor:
    """Exercise Coordinator state transitions without bypassing durable operations."""

    def __init__(self, coordinator: TaskCoordinator) -> None:
        self.coordinator = coordinator
        self.artifact_version = 100

    async def run_worker(self, context: dict[str, Any]) -> WorkerRunResult:
        refs = cast(dict[str, Any], context["task_spec"])["context_refs"]
        kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item)
        artifact_kind = {
            "direction": "direction",
            "compose": "structured_script",
            "structure": "structure",
            "paragraph": "paragraph",
            "element-set": "element_set",
        }[kind]
        digest = "sha256:" + sha256(context["attempt_id"].encode()).hexdigest()
        version = str(self.artifact_version)
        self.artifact_version += 1
        ref = ArtifactRef(
            f"script-build://artifact-versions/{version}",
            artifact_kind,
            version,
            digest,
        )
        await self.coordinator.submit_attempt(
            {
                "role": AgentRole.WORKER.value,
                "root_trace_id": context["root_trace_id"],
                "task_id": context["task_id"],
                "attempt_id": context["attempt_id"],
                "spec_version": context["spec_version"],
                "trace_id": context["worker_trace_id"],
                "tool_call_id": f"submit:{context['attempt_id']}",
                "operation_id": context["operation_id"],
                "execution_epoch": context["execution_epoch"],
            },
            AttemptSubmission(
                summary=f"kind={artifact_kind};status=frozen",
                artifact_refs=[ref],
            ),
        )
        return WorkerRunResult(context["worker_trace_id"], "completed")

    async def run_validator(self, context: dict[str, Any]) -> ValidatorRunResult:
        refs = cast(dict[str, Any], context["task_spec"])["context_refs"]
        kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item)
        placeholder = kind == "compose" and context["spec_version"] == 2
        verdict = ValidationVerdict.FAILED if placeholder else ValidationVerdict.PASSED
        criterion = cast(dict[str, Any], context["task_spec"])["acceptance_criteria"][0]
        await self.coordinator.submit_validation(
            {
                "role": AgentRole.VALIDATOR.value,
                "root_trace_id": context["root_trace_id"],
                "task_id": context["task_id"],
                "attempt_id": context["attempt_id"],
                "validation_id": context["validation_id"],
                "snapshot_id": context["snapshot_id"],
                "trace_id": context["validator_trace_id"],
                "tool_call_id": f"validate:{context['validation_id']}",
                "operation_id": context["operation_id"],
                "execution_epoch": context["execution_epoch"],
            },
            verdict,
            [
                CriterionResult(
                    criterion_id=criterion["criterion_id"],
                    verdict=verdict,
                    reason=(
                        "REALIZATION_PLACEHOLDER at artifact.paragraphs[0].description"
                        if placeholder
                        else "the replacement is concretely rendered"
                    ),
                )
            ],
            (
                "hard_defects=1;defect_codes=REALIZATION_PLACEHOLDER"
                if placeholder
                else "hard_defects=0"
            ),
            (),
            ("artifact.paragraphs[0].description",) if placeholder else (),
            ("REALIZATION_PLACEHOLDER",) if placeholder else (),
            "split" if placeholder else "accept",
        )
        return ValidatorRunResult(context["validator_trace_id"], "completed")

    async def stop(self, trace_id: str) -> bool:
        del trace_id
        return True


async def _service(
    tmp_path: Path, *, active_direction: bool = True
) -> tuple[PhaseTwoPlanningService, TaskCoordinator, str]:
    root = "root-contracts"
    coordinator = TaskCoordinator(
        FileSystemTaskStore(str(tmp_path / "ledger")),
        FileSystemArtifactStore(str(tmp_path / "snapshots")),
        FileSystemTraceStore(str(tmp_path / "traces")),
    )
    await coordinator.ensure_ledger(
        root,
        {
            "objective": "build a script",
            "acceptance_criteria": [
                {"criterion_id": "ready", "description": "ready", "hard": True}
            ],
            "context_refs": ["script-build://inputs/11"],
        },
    )
    bindings = _Bindings(root)
    service = PhaseTwoPlanningService(
        coordinator=coordinator,
        bindings=bindings,
        contracts=FileScriptTaskContractStore(tmp_path / "data"),
        artifacts=cast(Any, SimpleNamespace()),
    )
    if active_direction:
        await _activate_direction(service, coordinator, bindings, root)
    return service, coordinator, root


async def _activate_direction(
    service: PhaseTwoPlanningService,
    coordinator: TaskCoordinator,
    bindings: _Bindings,
    root: str,
) -> None:
    coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator))
    planned = await service.plan_script_tasks(
        contract_payloads=[_planner_input(_contract("direction"))],
        parent_task_id=None,
        context={"root_trace_id": root, "tool_call_id": "fixture:direction"},
    )
    direction_id = cast(str, planned["task_ids"][0])
    cycle = (
        await service.dispatch_script_tasks(
            task_ids=(direction_id,),
            context={"root_trace_id": root, "tool_call_id": "fixture:dispatch-direction"},
        )
    )[0]
    accepted = await service.decide_script_task(
        task_id=direction_id,
        action=DecisionAction.ACCEPT.value,
        reason="fixture Direction passed",
        validation_id=cast(str, cycle["validation_id"]),
        replacement_contract=None,
        child_contracts=(),
        context={"root_trace_id": root, "tool_call_id": "fixture:accept-direction"},
    )
    ledger = await coordinator.task_store.load(root)
    decision = ledger.decisions[cast(str, accepted["decision_id"])]
    attempt = ledger.attempts[cast(str, decision.attempt_id)]
    assert attempt.submission is not None
    direction_ref = attempt.submission.artifact_refs[0]
    bindings.binding = replace(
        bindings.binding,
        active_direction_artifact_version_id=int(direction_ref.version),
    )


@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, active_direction=False)
    direction = await service.plan_script_tasks(
        contract_payloads=[_planner_input(_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=[_planner_input(_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=[
            _planner_input(_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="INPUT_SCOPE_MISMATCH"):
        await service._guard_accepted_input_scopes(root, ledger, (paragraph,))


class _BlockingExecutor:
    def __init__(self) -> None:
        self.started = asyncio.Event()
        self.stopped_trace_ids: list[str] = []

    async def run_worker(self, context: dict[str, Any]) -> WorkerRunResult:
        self.started.set()
        await asyncio.Event().wait()
        raise AssertionError(f"worker unexpectedly resumed: {context['attempt_id']}")

    async def run_validator(self, context: dict[str, Any]) -> ValidatorRunResult:
        raise AssertionError(f"validator must not start: {context['validation_id']}")

    async def stop(self, trace_id: str) -> bool:
        self.stopped_trace_ids.append(trace_id)
        return True


class _StopState:
    def __init__(self) -> None:
        self.status = BuildStatus.RUNNING
        self.error_summary: str | None = None

    async def get_status(self, script_build_id: int) -> BuildStatus:
        assert script_build_id == 7
        return self.status

    async def set_status(
        self,
        script_build_id: int,
        status: BuildStatus,
        *,
        error_summary: str | None = None,
    ) -> None:
        assert script_build_id == 7
        self.status = status
        self.error_summary = error_summary


@pytest.mark.asyncio
async def test_real_operation_stop_converges_and_forbids_new_dispatch(tmp_path: Path) -> None:
    root = "root-stop-operation"
    trace_store = FileSystemTraceStore(str(tmp_path / "stop-traces"))
    coordinator = TaskCoordinator(
        FileSystemTaskStore(str(tmp_path / "stop-ledger")),
        FileSystemArtifactStore(str(tmp_path / "stop-artifacts")),
        trace_store,
        config=OrchestrationConfig(stop_grace_seconds=0.01),
    )
    await coordinator.ensure_ledger(
        root,
        {
            "objective": "build a script",
            "acceptance_criteria": [
                {"criterion_id": "ready", "description": "ready", "hard": True}
            ],
            "context_refs": ["script-build://inputs/11"],
        },
    )
    bindings = _Bindings(root)
    planning = PhaseTwoPlanningService(
        coordinator=coordinator,
        bindings=bindings,
        contracts=FileScriptTaskContractStore(tmp_path / "stop-contracts"),
        artifacts=cast(Any, SimpleNamespace()),
    )
    await _activate_direction(planning, coordinator, bindings, root)
    portfolio = await planning.plan_script_tasks(
        contract_payloads=[
            _planner_input(_contract("candidate-portfolio", execution_ready=False))
        ],
        parent_task_id=None,
        context={"root_trace_id": root, "tool_call_id": "stop:portfolio"},
    )
    compose = await planning.plan_script_tasks(
        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
        parent_task_id=portfolio["task_ids"][0],
        context={"root_trace_id": root, "tool_call_id": "stop:compose"},
    )
    paragraph = await planning.plan_script_tasks(
        contract_payloads=[_planner_input(_contract("paragraph"))],
        parent_task_id=compose["task_ids"][0],
        context={"root_trace_id": root, "tool_call_id": "stop:paragraph"},
    )
    executor = _BlockingExecutor()
    coordinator.set_executor(executor)

    async def unused_llm(**_kwargs: Any) -> Any:
        raise AssertionError("Planner LLM must not run during operation stop")

    runner = AgentRunner(trace_store=trace_store, llm_call=unused_llm)
    state = _StopState()

    class Authorizer:
        async def require_access(self, principal: Principal, script_build_id: int) -> None:
            assert principal.subject == "owner" and script_build_id == 7

    mission = ScriptMissionService(
        runner=runner,
        coordinator=coordinator,
        factory=ScriptMissionFactory(),
        input_snapshots=SimpleNamespace(),
        bindings=bindings,
        legacy_state=state,
        authorizer=Authorizer(),
        direction_reconciler=SimpleNamespace(),
        stop_timeout_seconds=1,
    )
    planning.dispatch_guard = mission
    dispatch = asyncio.create_task(
        planning.dispatch_script_tasks(
            task_ids=paragraph["task_ids"],
            context={"root_trace_id": root, "tool_call_id": "stop:dispatch"},
        )
    )
    await asyncio.wait_for(executor.started.wait(), timeout=1)

    stopped = await mission.stop(7, Principal("owner"))
    dispatch_result = await asyncio.gather(dispatch, return_exceptions=True)

    assert stopped.status is BuildStatus.STOPPED
    assert len(stopped.stopped_operation_ids) == 1
    assert executor.stopped_trace_ids
    assert not isinstance(dispatch_result[0], asyncio.CancelledError)
    ledger = await coordinator.task_store.load(root)
    operation = ledger.operations[stopped.stopped_operation_ids[0]]
    assert operation.status is OperationStatus.STOPPED
    assert ledger.tasks[paragraph["task_ids"][0]].status is TaskStatus.NEEDS_REPLAN
    assert (await mission.stop(7, Principal("owner"))).status is BuildStatus.STOPPED

    with pytest.raises(ProtocolViolation, match="forbidden after stop intent"):
        await planning.dispatch_script_tasks(
            task_ids=paragraph["task_ids"],
            context={"root_trace_id": root, "tool_call_id": "stop:late-dispatch"},
        )


@pytest.mark.asyncio
async def test_contract_store_is_content_addressed_and_detects_tampering(tmp_path: Path) -> None:
    store = FileScriptTaskContractStore(tmp_path)
    from script_build_host.domain.task_contracts import ScriptTaskContractV1

    contract = ScriptTaskContractV1.from_payload(_contract("paragraph"))
    first = await store.freeze("root-a", contract)
    second = await store.freeze("root-a", contract)
    assert first == second
    path = tmp_path / "script-task-contracts" / "root-a" / "sha256" / first.uri.rsplit("/", 1)[-1]
    path.write_bytes(b"{}")
    with pytest.raises(TaskContractError, match="TASK_CONTRACT_DIGEST_MISMATCH"):
        await store.read("root-a", first.uri)
    assert not await store.verify("root-a", first.uri, first.digest)


@pytest.mark.asyncio
async def test_orphan_contract_reclamation_uses_only_ledger_references(tmp_path: Path) -> None:
    service, _, root = await _service(tmp_path)
    referenced = await service.plan_script_tasks(
        contract_payloads=[
            _planner_input(_contract("candidate-portfolio", execution_ready=False))
        ],
        parent_task_id=None,
        context={"root_trace_id": root, "tool_call_id": "portfolio"},
    )
    orphan = await service.contracts.freeze(
        root, ScriptTaskContractV1.from_payload(_contract("paragraph"))
    )
    removed = await service.reclaim_orphan_contracts(root_trace_id=root)
    assert removed == (orphan.uri,)
    assert await service.contracts.verify(
        root,
        referenced["contracts"][0]["contract_ref"],
        referenced["contracts"][0]["contract_digest"],
    )
    assert not await service.contracts.verify(root, orphan.uri, orphan.digest)


@pytest.mark.asyncio
async def test_failed_operation_discards_workspace_but_recovery_required_preserves_it(
    tmp_path: Path,
) -> None:
    service, _, root = await _service(tmp_path)

    class Lifecycle:
        def __init__(self) -> None:
            self.calls: list[dict[str, str]] = []

        async def discard_attempt(self, **values: str) -> None:
            self.calls.append(values)

    lifecycle = Lifecycle()
    service.workspace_lifecycle = lifecycle
    attempts = {
        "failed-attempt": SimpleNamespace(attempt_id="failed-attempt", task_id="paragraph-task"),
        "recovery-attempt": SimpleNamespace(
            attempt_id="recovery-attempt", task_id="paragraph-task"
        ),
    }
    ledger = SimpleNamespace(
        attempts=attempts,
        operations={
            "failed": SimpleNamespace(
                status=OperationStatus.FAILED,
                error="worker failed before submit",
                attempt_ids=["failed-attempt"],
            ),
            "recovery": SimpleNamespace(
                status=OperationStatus.FAILED,
                error="MISSION_RECOVERY_REQUIRED",
                attempt_ids=["recovery-attempt"],
            ),
        },
    )
    await service._reconcile_terminal_workspaces(root, ledger)
    assert lifecycle.calls == [
        {
            "root_trace_id": root,
            "task_id": "paragraph-task",
            "attempt_id": "failed-attempt",
            "reason": "worker failed before submit",
        }
    ]


def test_comparison_refs_are_kind_specific_and_limits_cannot_be_raised() -> None:
    candidate_refs = []
    for index, kind in ((1, "paragraph"), (2, "structure")):
        candidate_refs.append(
            {
                "decision_id": f"decision-{index}",
                "artifact_ref": {
                    "uri": f"script-build://artifact-versions/{index}",
                    "kind": kind,
                    "version": str(index),
                    "digest": "sha256:" + str(index) * 64,
                },
                "scope_ref": "script-build://scopes/full",
                "expected_task_kind": kind,
            }
        )
    compare = _contract("compare")
    compare["comparison_decision_refs"] = candidate_refs
    assert ScriptTaskContractV1.from_payload(compare).task_kind.value == "compare"

    comparison_ref = {
        "decision_id": "decision-compare",
        "artifact_ref": {
            "uri": "script-build://artifact-versions/3",
            "kind": "comparison",
            "version": "3",
            "digest": "sha256:" + "3" * 64,
        },
        "scope_ref": "script-build://scopes/full",
        "expected_task_kind": "compare",
    }
    compose = _contract("compose")
    compose["comparison_decision_refs"] = [comparison_ref]
    ScriptTaskContractV1.from_payload(compose)
    portfolio = _contract("candidate-portfolio")
    portfolio["comparison_decision_refs"] = [comparison_ref]
    with pytest.raises(TaskContractError, match="comparison_decision_refs"):
        ScriptTaskContractV1.from_payload(portfolio)
    with pytest.raises(TaskContractError, match="TASK_BUDGET_EXCEEDED"):
        PhaseTwoLimits(max_tasks=65)


@pytest.mark.asyncio
async def test_planning_freezes_contract_and_creates_only_one_portfolio(tmp_path: Path) -> None:
    service, coordinator, root = await _service(tmp_path)
    result = await service.plan_script_tasks(
        contract_payloads=[
            _planner_input(_contract("candidate-portfolio", execution_ready=False))
        ],
        parent_task_id=None,
        context={"root_trace_id": root, "tool_call_id": "plan-portfolio"},
    )
    task_id = result["task_ids"][0]
    ledger = await coordinator.task_store.load(root)
    task = ledger.tasks[task_id]
    assert task.current_spec.context_refs[0].endswith("candidate-portfolio")
    assert task.current_spec.context_refs[1].startswith("script-build://task-contracts/sha256/")
    assert task.current_spec.context_refs[2] == "script-build://inputs/11"
    frozen = await service.contract_for_task(root, task)
    assert frozen.contract.scope_ref == "script-build://scopes/full"

    with pytest.raises(TaskContractError, match="one Portfolio"):
        await service.plan_script_tasks(
            contract_payloads=[
                _planner_input(_contract("candidate-portfolio", execution_ready=False))
            ],
            parent_task_id=None,
            context={"root_trace_id": root, "tool_call_id": "plan-portfolio-2"},
        )


@pytest.mark.asyncio
async def test_protected_phase_rejects_cross_phase_root_contracts(tmp_path: Path) -> None:
    service, coordinator, root = await _service(tmp_path)
    before = await coordinator.task_store.load(root)
    existing_task_ids = set(before.tasks)
    with pytest.raises(TaskContractError, match="PHASE_POLICY_VIOLATION"):
        await service.plan_script_tasks(
            contract_payloads=[
                _planner_input(_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 set(ledger.tasks) == existing_task_ids


@pytest.mark.asyncio
async def test_invalid_contract_batch_creates_zero_tasks(tmp_path: Path) -> None:
    service, coordinator, root = await _service(tmp_path)
    before = await coordinator.task_store.load(root)
    existing_task_ids = set(before.tasks)
    invalid = _planner_input(_contract("paragraph"))
    invalid["output_schema"] = "structured-script/v1"
    with pytest.raises(TaskContractError, match="Host-owned fields"):
        await service.plan_script_tasks(
            contract_payloads=[
                _planner_input(_contract("candidate-portfolio")),
                invalid,
            ],
            parent_task_id=None,
            context={"root_trace_id": root},
        )
    ledger = await coordinator.task_store.load(root)
    assert set(ledger.tasks) == existing_task_ids


@pytest.mark.asyncio
async def test_planning_rejects_unresolved_supersession_before_freeze(tmp_path: Path) -> None:
    service, coordinator, root = await _service(tmp_path)
    portfolio = await service.plan_script_tasks(
        contract_payloads=[
            _planner_input(_contract("candidate-portfolio", execution_ready=False))
        ],
        parent_task_id=None,
        context={"root_trace_id": root, "tool_call_id": "portfolio"},
    )
    compose = await service.plan_script_tasks(
        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
        parent_task_id=portfolio["task_ids"][0],
        context={"root_trace_id": root, "tool_call_id": "compose"},
    )
    replacement = _contract("paragraph")
    replacement["supersedes_decision_ids"] = ["missing-accept"]
    before = await coordinator.task_store.load(root)
    existing_task_ids = set(before.tasks)

    with pytest.raises(TaskContractError, match="not an ACCEPT decision"):
        await service.plan_script_tasks(
            contract_payloads=[_planner_input(replacement)],
            parent_task_id=compose["task_ids"][0],
            context={"root_trace_id": root, "tool_call_id": "invalid-replacement"},
        )

    ledger = await coordinator.task_store.load(root)
    assert set(ledger.tasks) == existing_task_ids


@pytest.mark.asyncio
async def test_open_compose_container_cannot_dispatch(tmp_path: Path) -> None:
    service, coordinator, root = await _service(tmp_path)
    portfolio = await service.plan_script_tasks(
        contract_payloads=[
            _planner_input(_contract("candidate-portfolio", execution_ready=False))
        ],
        parent_task_id=None,
        context={"root_trace_id": root, "tool_call_id": "p"},
    )
    compose = await service.plan_script_tasks(
        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
        parent_task_id=portfolio["task_ids"][0],
        context={"root_trace_id": root, "tool_call_id": "c"},
    )
    before = await coordinator.task_store.load(root)
    operation_ids = set(before.operations)
    with pytest.raises(TaskContractError, match="first accept child candidates"):
        await service.dispatch_script_tasks(
            task_ids=compose["task_ids"], context={"root_trace_id": root}
        )
    ledger = await coordinator.task_store.load(root)
    assert set(ledger.operations) == operation_ids


@pytest.mark.asyncio
async def test_compose_placeholder_split_replacement_then_revised_contract_passes(
    tmp_path: Path,
) -> None:
    service, coordinator, root = await _service(tmp_path)
    coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator))
    portfolio = await service.plan_script_tasks(
        contract_payloads=[
            _planner_input(_contract("candidate-portfolio", execution_ready=False))
        ],
        parent_task_id=None,
        context={"root_trace_id": root, "tool_call_id": "portfolio"},
    )
    compose = await service.plan_script_tasks(
        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
        parent_task_id=portfolio["task_ids"][0],
        context={"root_trace_id": root, "tool_call_id": "compose-v1"},
    )
    compose_id = compose["task_ids"][0]

    accepted_children: dict[str, str] = {}
    for kind in ("structure", "paragraph", "element-set"):
        planned = await service.plan_script_tasks(
            contract_payloads=[_planner_input(_contract(kind))],
            parent_task_id=compose_id,
            context={"root_trace_id": root, "tool_call_id": f"plan:{kind}"},
        )
        child_id = cast(str, planned["task_ids"][0])
        child_cycle = (
            await service.dispatch_script_tasks(
                task_ids=(child_id,),
                context={"root_trace_id": root, "tool_call_id": f"dispatch:{kind}"},
            )
        )[0]
        accepted = await service.decide_script_task(
            task_id=child_id,
            action=DecisionAction.ACCEPT.value,
            reason=f"{kind} passed",
            validation_id=cast(str, child_cycle["validation_id"]),
            replacement_contract=None,
            child_contracts=(),
            context={"root_trace_id": root, "tool_call_id": f"accept:{kind}"},
        )
        accepted_children[kind] = cast(str, accepted["decision_id"])
    await service.decide_script_task(
        task_id=compose_id,
        action=DecisionAction.REVISE.value,
        reason="freeze the initial complete frontier",
        validation_id=None,
        replacement_contract=None,
        child_contracts=(),
        selected_decision_ids=tuple(accepted_children.values()),
        context={"root_trace_id": root, "tool_call_id": "revise-compose-v1"},
    )

    first_cycle = (
        await service.dispatch_script_tasks(
            task_ids=(compose_id,),
            context={"root_trace_id": root, "tool_call_id": "dispatch-compose-v1"},
        )
    )[0]
    assert first_cycle["error"] is None
    ledger = await coordinator.task_store.load(root)
    first_validation = ledger.validations[first_cycle["validation_id"]]
    assert first_validation.verdict is ValidationVerdict.FAILED
    assert first_validation.unverified_claims == ["artifact.paragraphs[0].description"]
    assert first_validation.risks == ["REALIZATION_PLACEHOLDER"]

    split = await service.decide_script_task(
        task_id=compose_id,
        action=DecisionAction.SPLIT.value,
        reason="replace only the unrealized opening scope",
        validation_id=first_validation.validation_id,
        replacement_contract=None,
        child_contracts=(
            _planner_input(
                _contract(
                    "paragraph",
                    scope="script-build://scopes/full/replacement",
                )
            ),
        ),
        context={"root_trace_id": root, "tool_call_id": "split-placeholder"},
    )
    replacement_id = split["payload"]["child_task_ids"][0]
    replacement_cycle = (
        await service.dispatch_script_tasks(
            task_ids=(replacement_id,),
            context={"root_trace_id": root, "tool_call_id": "dispatch-replacement"},
        )
    )[0]
    replacement_accept = await service.decide_script_task(
        task_id=replacement_id,
        action=DecisionAction.ACCEPT.value,
        reason="the local replacement passed independent validation",
        validation_id=replacement_cycle["validation_id"],
        replacement_contract=None,
        child_contracts=(),
        context={"root_trace_id": root, "tool_call_id": "accept-replacement"},
    )

    ledger = await coordinator.task_store.load(root)
    assert ledger.tasks[compose_id].status is TaskStatus.NEEDS_REPLAN
    replacement_decision_id = replacement_accept["decision_id"]
    await service.decide_script_task(
        task_id=compose_id,
        action=DecisionAction.REVISE.value,
        reason="pin the accepted replacement and retry the complete render",
        validation_id=None,
        replacement_contract=None,
        child_contracts=(),
        selected_decision_ids=(
            accepted_children["structure"],
            replacement_decision_id,
            accepted_children["element-set"],
        ),
        context={"root_trace_id": root, "tool_call_id": "revise-compose-v2"},
    )
    second_cycle = (
        await service.dispatch_script_tasks(
            task_ids=(compose_id,),
            context={"root_trace_id": root, "tool_call_id": "dispatch-compose-v2"},
        )
    )[0]
    compose_accept = await service.decide_script_task(
        task_id=compose_id,
        action=DecisionAction.ACCEPT.value,
        reason="the revised full render passed independent validation",
        validation_id=second_cycle["validation_id"],
        replacement_contract=None,
        child_contracts=(),
        context={"root_trace_id": root, "tool_call_id": "accept-compose-v2"},
    )

    reloaded = await FileSystemTaskStore(str(tmp_path / "ledger")).load(root)
    compose_task = reloaded.tasks[compose_id]
    assert compose_task.status is TaskStatus.COMPLETED
    assert compose_task.current_spec_version == 3
    assert len(compose_task.attempt_ids) == 2
    assert [
        reloaded.validations[validation_id].verdict for validation_id in compose_task.validation_ids
    ] == [ValidationVerdict.FAILED, ValidationVerdict.PASSED]
    assert [reloaded.decisions[item].action for item in compose_task.decision_ids] == [
        DecisionAction.REVISE,
        DecisionAction.SPLIT,
        DecisionAction.REVISE,
        DecisionAction.ACCEPT,
    ]
    assert compose_accept["status"] == TaskStatus.COMPLETED.value
    assert reloaded.tasks[replacement_id].status is TaskStatus.COMPLETED


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "creation_order",
    [
        ("element-set", "paragraph", "structure"),
        ("paragraph", "element-set", "structure"),
        ("structure", "paragraph", "element-set"),
    ],
)
async def test_exploration_order_and_completion_order_do_not_choose_compose_order(
    tmp_path: Path,
    creation_order: tuple[str, str, str],
) -> None:
    service, coordinator, root = await _service(tmp_path)
    coordinator.set_executor(_PlaceholderThenReplacementExecutor(coordinator))
    portfolio = await service.plan_script_tasks(
        contract_payloads=[
            _planner_input(_contract("candidate-portfolio", execution_ready=False))
        ],
        parent_task_id=None,
        context={"root_trace_id": root, "tool_call_id": "portfolio"},
    )
    compose = await service.plan_script_tasks(
        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
        parent_task_id=portfolio["task_ids"][0],
        context={"root_trace_id": root, "tool_call_id": "compose"},
    )
    compose_id = compose["task_ids"][0]
    task_by_kind: dict[str, str] = {}
    for kind in creation_order:
        planned = await service.plan_script_tasks(
            contract_payloads=[_planner_input(_contract(kind))],
            parent_task_id=compose_id,
            context={"root_trace_id": root, "tool_call_id": f"plan:{kind}"},
        )
        task_by_kind[kind] = planned["task_ids"][0]

    decision_by_kind: dict[str, str] = {}
    completion_order = tuple(reversed(creation_order))
    for kind in completion_order:
        task_id = task_by_kind[kind]
        cycle = (
            await service.dispatch_script_tasks(
                task_ids=(task_id,),
                context={"root_trace_id": root, "tool_call_id": f"dispatch:{kind}"},
            )
        )[0]
        accepted = await service.decide_script_task(
            task_id=task_id,
            action=DecisionAction.ACCEPT.value,
            reason=f"the bounded {kind} increment passed validation",
            validation_id=cycle["validation_id"],
            replacement_contract=None,
            child_contracts=(),
            context={"root_trace_id": root, "tool_call_id": f"accept:{kind}"},
        )
        decision_id = accepted["decision_id"]
        decision_by_kind[kind] = decision_id

    ledger = await coordinator.task_store.load(root)
    assert ledger.tasks[compose_id].status is TaskStatus.NEEDS_REPLAN
    adoption_kinds = ("paragraph", "structure", "element-set")
    selected = tuple(decision_by_kind[kind] for kind in adoption_kinds)
    await service.decide_script_task(
        task_id=compose_id,
        action=DecisionAction.REVISE.value,
        reason="formal adoption is based on scope and intent, not timing",
        validation_id=None,
        replacement_contract=None,
        child_contracts=(),
        selected_decision_ids=selected,
        context={"root_trace_id": root, "tool_call_id": "close-compose"},
    )

    reloaded = await FileSystemTaskStore(str(tmp_path / "ledger")).load(root)
    frozen = await service.contract_for_task(root, reloaded.tasks[compose_id])
    assert frozen.contract.compose_order == selected
    assert [
        next(kind for kind, task_id in task_by_kind.items() if task_id == child_id)
        for child_id in reloaded.tasks[compose_id].child_task_ids
    ] == list(creation_order)
    assert completion_order == tuple(reversed(creation_order))


@pytest.mark.asyncio
async def test_phase_two_retrieval_counts_toward_task_and_depth_limits(tmp_path: Path) -> None:
    service, _, root = await _service(tmp_path)
    service.limits = PhaseTwoLimits(max_tasks=2)
    portfolio = await service.plan_script_tasks(
        contract_payloads=[
            _planner_input(_contract("candidate-portfolio", execution_ready=False))
        ],
        parent_task_id=None,
        context={"root_trace_id": root, "tool_call_id": "portfolio"},
    )
    compose = await service.plan_script_tasks(
        contract_payloads=[_planner_input(_contract("compose", execution_ready=False))],
        parent_task_id=portfolio["task_ids"][0],
        context={"root_trace_id": root, "tool_call_id": "compose"},
    )
    with pytest.raises(TaskContractError, match="Task budget"):
        await service.plan_script_tasks(
            contract_payloads=[_planner_input(_contract("decode-retrieval"))],
            parent_task_id=compose["task_ids"][0],
            context={"root_trace_id": root, "tool_call_id": "retrieval"},
        )

    service.limits = PhaseTwoLimits(max_depth=1)
    with pytest.raises(TaskContractError, match="depth"):
        await service.plan_script_tasks(
            contract_payloads=[
                _planner_input(
                    _contract("structure", scope="script-build://scopes/full/local")
                )
            ],
            parent_task_id=compose["task_ids"][0],
            context={"root_trace_id": root, "tool_call_id": "too-deep"},
        )


@pytest.mark.asyncio
async def test_task_contract_cannot_be_silently_changed_in_task_spec(tmp_path: Path) -> None:
    service, coordinator, root = await _service(tmp_path)
    result = await service.plan_script_tasks(
        contract_payloads=[
            _planner_input(_contract("candidate-portfolio", execution_ready=False))
        ],
        parent_task_id=None,
        context={"root_trace_id": root},
    )
    ledger = await coordinator.task_store.load(root)
    task = ledger.tasks[result["task_ids"][0]]
    task.specs[-1] = replace(task.specs[-1], objective="tampered")
    await coordinator.task_store.commit(ledger, expected_revision=ledger.revision)
    with pytest.raises(TaskContractError, match="DIGEST_MISMATCH"):
        await service.contract_for_task(root, task)