from __future__ import annotations

import asyncio
import re
from collections.abc import Mapping, Sequence
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, cast

import pytest
from agent import FileSystemArtifactStore, FileSystemTaskStore, FileSystemTraceStore
from agent.orchestration import (
    AgentRole,
    ArtifactRef,
    AttemptSubmission,
    CriterionResult,
    DecisionAction,
    TaskCoordinator,
    TaskStatus,
    ValidationVerdict,
)
from agent.orchestration.protocols import ValidatorRunResult, WorkerRunResult
from sqlalchemy import event, func, select

from script_build_host.application.phase_two_boundary import ScriptPhaseTwoBoundaryVerifier
from script_build_host.application.phase_two_candidates import PhaseTwoCandidateService
from script_build_host.application.phase_two_inputs import (
    AcceptedInputResolver,
    ActiveFrontierResolver,
    StoredTaskContractReader,
)
from script_build_host.application.phase_two_planning import (
    PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
    PhaseTwoPlanningService,
)
from script_build_host.domain.artifacts import (
    ArtifactKind,
    DirectionGoal,
    EvidenceRecordV1,
    ScriptDirectionArtifactV1,
)
from script_build_host.domain.errors import PhaseTwoBoundaryNotReady
from script_build_host.domain.phase_two_artifacts import (
    CandidatePortfolioArtifactV1,
    StructuredScriptArtifactV1,
)
from script_build_host.domain.records import BuildStatus, MissionBinding
from script_build_host.domain.task_contracts import ScriptTaskKind
from script_build_host.infrastructure.legacy_tables import (
    script_build_element,
    script_build_paragraph,
    script_build_paragraph_element,
    script_build_record,
)
from script_build_host.infrastructure.tables import (
    artifact_version_table,
    publication_table,
)
from script_build_host.infrastructure.task_contract_store import FileScriptTaskContractStore
from script_build_host.repositories.legacy_state import SqlAlchemyLegacyBuildStateRepository
from script_build_host.repositories.sqlalchemy import (
    SqlAlchemyMissionBindingRepository,
    SqlAlchemyScriptBusinessArtifactRepository,
)
from script_build_host.repositories.workspace import SqlAlchemyCandidateWorkspaceRepository

ROOT = "phase-two-sql-root"
SNAPSHOT_ID = 11
FULL_SCOPE = "script-build://scopes/full"
FULL_WRITE = "script-build://writes/full"


def _contract(
    kind: ScriptTaskKind,
    *,
    scope: str = FULL_SCOPE,
    write_scope: Sequence[str] = (FULL_WRITE,),
    input_refs: Sequence[dict[str, Any]] = (),
    base_ref: ArtifactRef | None = None,
    supersedes: Sequence[str] = (),
    closure: Sequence[dict[str, Any]] = (),
    adopted: Sequence[str] = (),
    held: Sequence[str] = (),
    order: Sequence[str] = (),
    comparison_refs: Sequence[dict[str, Any]] = (),
    execution_ready: bool = True,
) -> dict[str, Any]:
    output_schema = {
        ScriptTaskKind.DIRECTION: "script-direction/v1",
        ScriptTaskKind.DECODE_RETRIEVAL: "evidence-record/v1",
        ScriptTaskKind.STRUCTURE: "structure-artifact/v1",
        ScriptTaskKind.PARAGRAPH: "paragraph-patch/v1" if base_ref else "paragraph-artifact/v1",
        ScriptTaskKind.ELEMENT_SET: (
            "element-set-patch/v1" if base_ref else "element-set-artifact/v1"
        ),
        ScriptTaskKind.COMPARE: "comparison-artifact/v1",
        ScriptTaskKind.COMPOSE: "structured-script/v1",
        ScriptTaskKind.CANDIDATE_PORTFOLIO: "candidate-portfolio/v1",
    }[kind]
    if kind in {ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO}:
        intent = "compose" if kind is ScriptTaskKind.COMPOSE else "portfolio"
    elif supersedes:
        intent = "replace"
    else:
        intent = "explore"
    payload: dict[str, Any] = {
        "schema_version": "script-task-contract/v1",
        "task_kind": kind.value,
        "scope_ref": scope,
        "intent_class": intent,
        "objective": f"produce independently verifiable {kind.value} content",
        "input_decision_refs": list(input_refs),
        "base_artifact_ref": _ref_payload(base_ref) if base_ref else None,
        "write_scope": list(write_scope),
        "gap_ref": None,
        "output_schema": output_schema,
        "criteria": [
            {
                "criterion_id": "closed",
                "description": "the frozen increment is concrete and independently verifiable",
                "hard": True,
            }
        ],
        "budget": {
            "max_attempts": 4,
            "max_tokens": 32_000,
            "max_seconds": 900,
            "max_external_queries": 40,
            "max_no_improvement": 3,
        },
        "supersedes_decision_ids": list(supersedes),
        "candidate_closure_decision_refs": list(closure),
        "adopted_decision_ids": list(adopted),
        "held_or_rejected_decision_ids": list(held),
        "compose_order": list(order),
        "comparison_decision_refs": list(comparison_refs),
    }
    if not execution_ready and kind in {
        ScriptTaskKind.COMPOSE,
        ScriptTaskKind.CANDIDATE_PORTFOLIO,
    }:
        payload.update(
            candidate_closure_decision_refs=[],
            adopted_decision_ids=[],
            held_or_rejected_decision_ids=[],
            compose_order=[],
        )
    return payload


def _decision_ref(
    decision_id: str,
    ref: ArtifactRef,
    *,
    scope: str,
    kind: ScriptTaskKind,
) -> dict[str, Any]:
    return {
        "decision_id": decision_id,
        "artifact_ref": _ref_payload(ref),
        "scope_ref": scope,
        "expected_task_kind": kind.value,
    }


def _ref_payload(ref: ArtifactRef) -> dict[str, Any]:
    return {
        "uri": ref.uri,
        "kind": ref.kind,
        "version": ref.version,
        "digest": ref.digest,
        "summary": ref.summary,
        "metadata": ref.metadata,
    }


def _task_kind(context: Mapping[str, Any]) -> ScriptTaskKind:
    refs = cast(dict[str, Any], context["task_spec"])["context_refs"]
    values = [value.rsplit("/", 1)[-1] for value in refs if "/task-kinds/" in value]
    assert len(values) == 1
    return ScriptTaskKind(values[0])


class _ScriptedExecutor:
    """Drive real Coordinator stages through the same protected Host services."""

    def __init__(
        self,
        *,
        coordinator: TaskCoordinator,
        candidates: PhaseTwoCandidateService,
        artifacts: SqlAlchemyScriptBusinessArtifactRepository,
        script_build_id: int,
    ) -> None:
        self.coordinator = coordinator
        self.candidates = candidates
        self.artifacts = artifacts
        self.script_build_id = script_build_id

    async def run_worker(self, context: dict[str, Any]) -> WorkerRunResult:
        kind = _task_kind(context)
        candidate_context = {
            "root_trace_id": context["root_trace_id"],
            "task_id": context["task_id"],
            "attempt_id": context["attempt_id"],
            "spec_version": context["spec_version"],
        }
        if kind is ScriptTaskKind.DECODE_RETRIEVAL:
            await self.artifacts.freeze(
                script_build_id=self.script_build_id,
                task_id=context["task_id"],
                attempt_id=context["attempt_id"],
                spec_version=context["spec_version"],
                artifact=EvidenceRecordV1(
                    evidence_id=f"evidence-{context['attempt_id']}",
                    source_type="decode",
                    tool_name="retrieve_decode",
                    query={"query": "opening tension", "top_k": 3},
                    source_refs=("decode-index://fixture/1",),
                    raw_artifact_ref=None,
                    summary="A concrete opening needs a visible tension and an observable detail.",
                    supports=("direction",),
                    confidence="high",
                    limitations=(),
                    content_sha256="",
                    created_at=datetime.now(UTC),
                ),
            )
        elif kind is ScriptTaskKind.DIRECTION:
            child_results = cast(list[dict[str, Any]], context["accepted_child_results"])
            assert len(child_results) == 1
            evidence_refs = child_results[0]["submission"]["evidence_refs"]
            await self.artifacts.freeze(
                script_build_id=self.script_build_id,
                task_id=context["task_id"],
                attempt_id=context["attempt_id"],
                spec_version=context["spec_version"],
                artifact=ScriptDirectionArtifactV1(
                    goals=(DirectionGoal("goal-1", "Use a concrete reversal to reveal the topic"),),
                    evidence_refs=(str(evidence_refs[0]["uri"]),),
                    legacy_markdown=(
                        "# Direction\n\nReveal the topic through one concrete reversal."
                    ),
                ),
            )
        elif kind in {ScriptTaskKind.STRUCTURE, ScriptTaskKind.PARAGRAPH}:
            workspace = await self.candidates.read_attempt_workspace(context=candidate_context)
            if workspace["paragraphs"]:
                paragraph_id = workspace["paragraphs"][0]["paragraph_id"]
            else:
                created = await self.candidates.create_script_paragraph(
                    payload={
                        "paragraph_index": 1,
                        "name": f"{kind.value} opening",
                        "content_range": {"scope": "opening"},
                    },
                    context=candidate_context,
                )
                paragraph_id = created["paragraph_id"]
            await self.candidates.batch_update_script_paragraphs(
                updates=(
                    {
                        "paragraph_id": paragraph_id,
                        "theme": "a specific reversal",
                        "form": "contrast",
                        "function": "hook",
                        "feeling": "curiosity",
                        "description": "The observable detail changes the initial interpretation.",
                        "full_description": (
                            "The opening states a familiar assumption, then overturns it with "
                            "one visible and source-grounded detail."
                        ),
                    },
                ),
                context=candidate_context,
            )
        elif kind is ScriptTaskKind.ELEMENT_SET:
            created = await self.candidates.create_script_element(
                payload={
                    "name": "observable reversal detail",
                    "dimension_primary": "实质",
                    "dimension_secondary": "narrative evidence",
                    "topic_support": {"scope": "opening"},
                    "weight_score": {"score": 0.9},
                },
                context=candidate_context,
            )
            workspace = await self.candidates.read_attempt_workspace(context=candidate_context)
            if workspace["paragraphs"]:
                await self.candidates.batch_link_paragraph_elements(
                    links=(
                        {
                            "paragraph_id": workspace["paragraphs"][0]["paragraph_id"],
                            "element_ids": [created["element_id"]],
                        },
                    ),
                    context=candidate_context,
                )
        elif kind is ScriptTaskKind.COMPARE:
            pinned = await self.candidates.read_pinned_candidates(context=candidate_context)
            candidate_refs = [str(item["artifact_ref"]["uri"]) for item in pinned]
            await self.candidates.save_comparison_candidate(
                payload={
                    "criterion_results": [
                        {
                            "criterion_id": "closed",
                            "candidate_results": [
                                {"artifact_ref": ref, "reason": "immutable candidate reviewed"}
                                for ref in candidate_refs
                            ],
                        }
                    ],
                    "conflicts": ["the replacement has the more concrete opening"],
                    "recommendation": candidate_refs[-1],
                },
                context=candidate_context,
            )
        elif kind is ScriptTaskKind.COMPOSE:
            await self.candidates.save_structured_script_candidate(
                acceptance_notes=("all adopted increments are pinned by ACCEPT decisions",),
                context=candidate_context,
            )
        elif kind is ScriptTaskKind.CANDIDATE_PORTFOLIO:
            await self.candidates.save_candidate_portfolio(
                payload={"unresolved_defects": []}, context=candidate_context
            )
        else:  # pragma: no cover - this executor is intentionally phase-two bounded
            raise AssertionError(f"unexpected worker kind: {kind.value}")

        manifest = await self.candidates.resolve_attempt_manifest(context=candidate_context)
        safe_ref = ArtifactRef(
            manifest.artifact_ref.uri,
            manifest.artifact_ref.kind,
            manifest.artifact_ref.version,
            manifest.artifact_ref.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={safe_ref.kind};status=frozen",
                artifact_refs=[] if safe_ref.kind == ArtifactKind.EVIDENCE.value else [safe_ref],
                evidence_refs=[safe_ref] if safe_ref.kind == ArtifactKind.EVIDENCE.value else [],
            ),
        )
        return WorkerRunResult(context["worker_trace_id"], "completed")

    async def run_validator(self, context: dict[str, Any]) -> ValidatorRunResult:
        criteria = cast(dict[str, Any], context["task_spec"])["acceptance_criteria"]
        results = [
            CriterionResult(
                criterion_id=item["criterion_id"],
                verdict=ValidationVerdict.PASSED,
                reason="the immutable SQL artifact is concrete and owner/digest closed",
            )
            for item in criteria
        ]
        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"],
            },
            ValidationVerdict.PASSED,
            results,
            "all hard criteria passed against the immutable snapshot",
            (),
            (),
            (),
            "accept",
        )
        return ValidatorRunResult(context["validator_trace_id"], "completed")

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


async def _plan_one(
    planning: PhaseTwoPlanningService,
    payload: Mapping[str, Any],
    *,
    parent_task_id: str | None,
    call_id: str,
) -> str:
    result = await planning.plan_script_tasks(
        contract_payloads=(payload,),
        parent_task_id=parent_task_id,
        context={"root_trace_id": ROOT, "tool_call_id": call_id},
    )
    return cast(str, result["task_ids"][0])


async def _dispatch_accept(
    planning: PhaseTwoPlanningService, coordinator: TaskCoordinator, task_id: str
) -> tuple[str, ArtifactRef]:
    cycles = await planning.dispatch_script_tasks(
        task_ids=(task_id,),
        context={"root_trace_id": ROOT, "tool_call_id": f"dispatch:{task_id}"},
    )
    assert cycles[0]["error"] is None
    validation_id = cast(str, cycles[0]["validation_id"])
    accepted = await planning.decide_script_task(
        task_id=task_id,
        action=DecisionAction.ACCEPT.value,
        reason="independent validation passed",
        validation_id=validation_id,
        replacement_contract=None,
        child_contracts=(),
        context={"root_trace_id": ROOT, "tool_call_id": f"accept:{task_id}"},
    )
    decision_id = cast(str, accepted["decision_id"])
    ledger = await coordinator.task_store.load(ROOT)
    decision = ledger.decisions[decision_id]
    attempt = ledger.attempts[cast(str, decision.attempt_id)]
    assert attempt.submission is not None
    refs = [*attempt.submission.artifact_refs, *attempt.submission.evidence_refs]
    assert len(refs) == 1
    return decision_id, refs[0]


@pytest.mark.asyncio
async def test_sql_phase_two_dynamic_replacement_compose_portfolio_and_boundary(
    database: Any, tmp_path: Path
) -> None:
    engine, sessions = database
    mutation_statements: list[str] = []

    def observe_sql(
        _connection: Any,
        _cursor: Any,
        statement: str,
        _parameters: Any,
        _context: Any,
        _executemany: bool,
    ) -> None:
        if statement.lstrip().upper().startswith(("INSERT ", "UPDATE ", "DELETE ")):
            mutation_statements.append(statement)

    event.listen(engine.sync_engine, "before_cursor_execute", observe_sql)
    build_states = SqlAlchemyLegacyBuildStateRepository(sessions)
    script_build_id = await build_states.create(
        execution_id=101,
        topic_build_id=202,
        topic_id=303,
        agent_type="script-planner",
        agent_config={},
        data_source_url=None,
        strategies_config={},
    )
    bindings = SqlAlchemyMissionBindingRepository(sessions)
    await bindings.create(
        script_build_id=script_build_id,
        root_trace_id=ROOT,
        input_snapshot_id=SNAPSHOT_ID,
        engine_version="phase-two-test",
        schema_version="v1",
    )
    task_path = tmp_path / "task-ledger"
    artifact_path = tmp_path / "framework-artifacts"
    contract_path = tmp_path / "contract-data"
    task_store = FileSystemTaskStore(str(task_path))
    framework_artifacts = FileSystemArtifactStore(str(artifact_path))
    coordinator = TaskCoordinator(
        task_store,
        framework_artifacts,
        FileSystemTraceStore(str(tmp_path / "traces")),
    )
    await coordinator.ensure_ledger(
        ROOT,
        {
            "objective": "produce a governed phase-two candidate portfolio",
            "acceptance_criteria": [
                {"criterion_id": "portfolio", "description": "portfolio closed", "hard": True}
            ],
            "context_refs": [f"script-build://inputs/{SNAPSHOT_ID}"],
        },
    )
    contract_store = FileScriptTaskContractStore(contract_path)
    contract_reader = StoredTaskContractReader(contract_store)
    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    accepted_inputs = AcceptedInputResolver(
        task_store=task_store,
        artifact_store=framework_artifacts,
        artifacts=artifacts,
        contracts=contract_reader,
    )
    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
    candidates = PhaseTwoCandidateService(
        bindings=bindings,
        task_store=task_store,
        framework_artifact_store=framework_artifacts,
        artifacts=artifacts,
        accepted_inputs=accepted_inputs,
        active_frontier=ActiveFrontierResolver(),
        workspaces=workspaces,
    )
    boundary = ScriptPhaseTwoBoundaryVerifier(
        task_store=task_store,
        artifact_store=framework_artifacts,
        artifacts=artifacts,
        bindings=bindings,
        accepted_inputs=accepted_inputs,
    )
    planning = PhaseTwoPlanningService(
        coordinator=coordinator,
        bindings=bindings,
        contracts=contract_store,
        closure_gate=boundary,
    )
    coordinator.set_executor(
        _ScriptedExecutor(
            coordinator=coordinator,
            candidates=candidates,
            artifacts=artifacts,
            script_build_id=script_build_id,
        )
    )

    direction = await _plan_one(
        planning, _contract(ScriptTaskKind.DIRECTION), parent_task_id=None, call_id="direction"
    )
    retrieval = await _plan_one(
        planning,
        _contract(ScriptTaskKind.DECODE_RETRIEVAL, write_scope=()),
        parent_task_id=direction,
        call_id="retrieval",
    )
    await _dispatch_accept(planning, coordinator, retrieval)
    direction_decision, direction_ref = await _dispatch_accept(planning, coordinator, direction)

    portfolio = await _plan_one(
        planning,
        _contract(
            ScriptTaskKind.CANDIDATE_PORTFOLIO,
            write_scope=("script-build://writes",),
            execution_ready=False,
        ),
        parent_task_id=None,
        call_id="portfolio",
    )
    compose = await _plan_one(
        planning,
        _contract(
            ScriptTaskKind.COMPOSE,
            write_scope=("script-build://writes",),
            execution_ready=False,
        ),
        parent_task_id=portfolio,
        call_id="compose",
    )

    # Exploration order is intentionally Element -> Paragraph -> Structure.
    old_element = await _plan_one(
        planning,
        _contract(
            ScriptTaskKind.ELEMENT_SET,
            scope="script-build://scopes/full/elements",
            write_scope=("script-build://writes/elements",),
        ),
        parent_task_id=compose,
        call_id="element-first",
    )
    old_element_decision, old_element_ref = await _dispatch_accept(
        planning, coordinator, old_element
    )
    old_paragraph = await _plan_one(
        planning,
        _contract(
            ScriptTaskKind.PARAGRAPH,
            scope="script-build://scopes/full/opening",
            write_scope=("script-build://writes/paragraphs/opening",),
        ),
        parent_task_id=compose,
        call_id="paragraph-second",
    )
    old_paragraph_decision, old_paragraph_ref = await _dispatch_accept(
        planning, coordinator, old_paragraph
    )
    structure = await _plan_one(
        planning,
        _contract(
            ScriptTaskKind.STRUCTURE,
            scope="script-build://scopes/full/opening",
            write_scope=("script-build://writes/paragraphs/opening",),
            input_refs=(
                _decision_ref(
                    old_paragraph_decision,
                    old_paragraph_ref,
                    scope="script-build://scopes/full/opening",
                    kind=ScriptTaskKind.PARAGRAPH,
                ),
            ),
            base_ref=old_paragraph_ref,
        ),
        parent_task_id=compose,
        call_id="structure-third",
    )
    structure_decision, structure_ref = await _dispatch_accept(planning, coordinator, structure)

    paragraph_replacement = await _plan_one(
        planning,
        _contract(
            ScriptTaskKind.PARAGRAPH,
            scope="script-build://scopes/full/opening",
            write_scope=("script-build://writes/paragraphs/opening",),
            input_refs=(
                _decision_ref(
                    structure_decision,
                    structure_ref,
                    scope="script-build://scopes/full/opening",
                    kind=ScriptTaskKind.STRUCTURE,
                ),
                _decision_ref(
                    old_paragraph_decision,
                    old_paragraph_ref,
                    scope="script-build://scopes/full/opening",
                    kind=ScriptTaskKind.PARAGRAPH,
                ),
            ),
            base_ref=structure_ref,
            supersedes=(old_paragraph_decision,),
        ),
        parent_task_id=compose,
        call_id="replace-paragraph",
    )
    paragraph_decision, paragraph_ref = await _dispatch_accept(
        planning, coordinator, paragraph_replacement
    )
    element_replacement = await _plan_one(
        planning,
        _contract(
            ScriptTaskKind.ELEMENT_SET,
            scope="script-build://scopes/full/elements",
            write_scope=("script-build://writes/elements",),
            input_refs=(
                _decision_ref(
                    old_element_decision,
                    old_element_ref,
                    scope="script-build://scopes/full/elements",
                    kind=ScriptTaskKind.ELEMENT_SET,
                ),
                _decision_ref(
                    paragraph_decision,
                    paragraph_ref,
                    scope="script-build://scopes/full/opening",
                    kind=ScriptTaskKind.PARAGRAPH,
                ),
            ),
            base_ref=paragraph_ref,
            supersedes=(old_element_decision,),
        ),
        parent_task_id=compose,
        call_id="replace-element",
    )
    element_decision, element_ref = await _dispatch_accept(
        planning, coordinator, element_replacement
    )

    comparison = await _plan_one(
        planning,
        _contract(
            ScriptTaskKind.COMPARE,
            scope="script-build://scopes/full/opening",
            write_scope=(),
            comparison_refs=(
                _decision_ref(
                    old_paragraph_decision,
                    old_paragraph_ref,
                    scope="script-build://scopes/full/opening",
                    kind=ScriptTaskKind.PARAGRAPH,
                ),
                _decision_ref(
                    paragraph_decision,
                    paragraph_ref,
                    scope="script-build://scopes/full/opening",
                    kind=ScriptTaskKind.PARAGRAPH,
                ),
            ),
        ),
        parent_task_id=compose,
        call_id="compare-opening-candidates",
    )
    comparison_decision, comparison_ref = await _dispatch_accept(planning, coordinator, comparison)

    with pytest.raises(PhaseTwoBoundaryNotReady, match="CandidatePortfolio"):
        await planning.decide_script_task(
            task_id=(await task_store.load(ROOT)).root_task_id,
            action=DecisionAction.BLOCK.value,
            reason=PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
            validation_id=None,
            replacement_contract=None,
            child_contracts=(),
            context={"root_trace_id": ROOT, "tool_call_id": "early-boundary"},
        )

    local_refs = (
        _decision_ref(
            old_element_decision,
            old_element_ref,
            scope="script-build://scopes/full/elements",
            kind=ScriptTaskKind.ELEMENT_SET,
        ),
        _decision_ref(
            old_paragraph_decision,
            old_paragraph_ref,
            scope="script-build://scopes/full/opening",
            kind=ScriptTaskKind.PARAGRAPH,
        ),
        _decision_ref(
            structure_decision,
            structure_ref,
            scope="script-build://scopes/full/opening",
            kind=ScriptTaskKind.STRUCTURE,
        ),
        _decision_ref(
            paragraph_decision,
            paragraph_ref,
            scope="script-build://scopes/full/opening",
            kind=ScriptTaskKind.PARAGRAPH,
        ),
        _decision_ref(
            element_decision,
            element_ref,
            scope="script-build://scopes/full/elements",
            kind=ScriptTaskKind.ELEMENT_SET,
        ),
    )
    adopted = (structure_decision, paragraph_decision, element_decision)
    held = (old_element_decision, old_paragraph_decision)
    compose_contract = _contract(
        ScriptTaskKind.COMPOSE,
        write_scope=("script-build://writes",),
        input_refs=(
            _decision_ref(
                direction_decision,
                direction_ref,
                scope=FULL_SCOPE,
                kind=ScriptTaskKind.DIRECTION,
            ),
        ),
        closure=local_refs,
        adopted=adopted,
        held=held,
        order=adopted,
        comparison_refs=(
            _decision_ref(
                comparison_decision,
                comparison_ref,
                scope="script-build://scopes/full/opening",
                kind=ScriptTaskKind.COMPARE,
            ),
        ),
    )
    await planning.decide_script_task(
        task_id=compose,
        action=DecisionAction.REVISE.value,
        reason="freeze the selected active frontier and formal compose order",
        validation_id=None,
        replacement_contract=compose_contract,
        child_contracts=(),
        context={"root_trace_id": ROOT, "tool_call_id": "close-compose"},
    )
    compose_decision, compose_ref = await _dispatch_accept(planning, coordinator, compose)

    portfolio_contract = _contract(
        ScriptTaskKind.CANDIDATE_PORTFOLIO,
        write_scope=("script-build://writes",),
        closure=(
            _decision_ref(
                compose_decision,
                compose_ref,
                scope=FULL_SCOPE,
                kind=ScriptTaskKind.COMPOSE,
            ),
        ),
        adopted=(compose_decision,),
        order=(compose_decision,),
    )
    await planning.decide_script_task(
        task_id=portfolio,
        action=DecisionAction.REVISE.value,
        reason="freeze the uniquely adopted StructuredScript",
        validation_id=None,
        replacement_contract=portfolio_contract,
        child_contracts=(),
        context={"root_trace_id": ROOT, "tool_call_id": "close-portfolio"},
    )
    portfolio_decision, _ = await _dispatch_accept(planning, coordinator, portfolio)

    ledger = await task_store.load(ROOT)
    compose_attempt = ledger.attempts[cast(str, ledger.decisions[compose_decision].attempt_id)]
    frontier = await candidates.read_active_frontier(
        context={
            "root_trace_id": ROOT,
            "task_id": compose,
            "attempt_id": compose_attempt.attempt_id,
            "spec_version": compose_attempt.spec_version,
        }
    )
    assert [item["decision_id"] for item in frontier["inputs"]] == list(adopted)
    assert old_element_decision not in {item["decision_id"] for item in frontier["inputs"]}
    assert old_paragraph_decision not in {item["decision_id"] for item in frontier["inputs"]}

    await planning.decide_script_task(
        task_id=ledger.root_task_id,
        action=DecisionAction.BLOCK.value,
        reason=PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
        validation_id=None,
        replacement_contract=None,
        child_contracts=(),
        context={"root_trace_id": ROOT, "tool_call_id": "phase-two-boundary"},
    )
    await build_states.set_checkpoint(
        script_build_id,
        checkpoint_code=PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
        summary="one validated CandidatePortfolio is ready for phase three",
        root_trace_id=ROOT,
    )

    reloaded = await FileSystemTaskStore(str(task_path)).load(ROOT)
    assert reloaded.tasks[reloaded.root_task_id].status is TaskStatus.BLOCKED
    assert reloaded.tasks[reloaded.root_task_id].blocked_reason == (
        PHASE_TWO_CANDIDATE_PORTFOLIO_READY
    )
    compose_children = [reloaded.tasks[item] for item in reloaded.tasks[compose].child_task_ids]
    stored_reader = StoredTaskContractReader(FileScriptTaskContractStore(contract_path))
    assert [
        (
            await stored_reader.read_for_task(
                root_trace_id=ROOT,
                task=item,
                spec_version=item.current_spec_version,
            )
        ).task_kind
        for item in compose_children[:3]
    ] == [
        ScriptTaskKind.ELEMENT_SET,
        ScriptTaskKind.PARAGRAPH,
        ScriptTaskKind.STRUCTURE,
    ]

    compose_version = await artifacts.get_by_attempt(
        script_build_id=script_build_id,
        task_id=compose,
        attempt_id=compose_attempt.attempt_id,
    )
    assert isinstance(compose_version.artifact, StructuredScriptArtifactV1)
    assert len(compose_version.artifact.paragraphs) == 1
    assert compose_version.artifact.elements
    assert compose_version.artifact.paragraph_element_links
    portfolio_attempt = reloaded.attempts[
        cast(str, reloaded.decisions[portfolio_decision].attempt_id)
    ]
    portfolio_version = await artifacts.get_by_attempt(
        script_build_id=script_build_id,
        task_id=portfolio,
        attempt_id=portfolio_attempt.attempt_id,
    )
    assert isinstance(portfolio_version.artifact, CandidatePortfolioArtifactV1)
    assert portfolio_version.artifact.accepted_decision_ids == (compose_decision,)
    assert portfolio_version.artifact.adopted_structured_script_ref == compose_ref.uri

    async with sessions() as session:
        build = (
            (
                await session.execute(
                    select(script_build_record).where(script_build_record.c.id == script_build_id)
                )
            )
            .mappings()
            .one()
        )
        branch_values = [
            *(
                await session.scalars(
                    select(script_build_paragraph.c.branch_id).where(
                        script_build_paragraph.c.script_build_id == script_build_id
                    )
                )
            ),
            *(
                await session.scalars(
                    select(script_build_element.c.branch_id).where(
                        script_build_element.c.script_build_id == script_build_id
                    )
                )
            ),
            *(
                await session.scalars(
                    select(script_build_paragraph_element.c.branch_id).where(
                        script_build_paragraph_element.c.script_build_id == script_build_id
                    )
                )
            ),
        ]
        final_publications = await session.scalar(
            select(func.count())
            .select_from(publication_table)
            .where(
                publication_table.c.script_build_id == script_build_id,
                publication_table.c.publication_type == "final",
            )
        )
        phase_two_rows = (
            (
                await session.execute(
                    select(artifact_version_table).where(
                        artifact_version_table.c.script_build_id == script_build_id,
                        artifact_version_table.c.artifact_type.in_(
                            [
                                ArtifactKind.STRUCTURE.value,
                                ArtifactKind.PARAGRAPH.value,
                                ArtifactKind.ELEMENT_SET.value,
                                ArtifactKind.COMPARISON.value,
                                ArtifactKind.STRUCTURED_SCRIPT.value,
                                ArtifactKind.CANDIDATE_PORTFOLIO.value,
                            ]
                        ),
                    )
                )
            )
            .mappings()
            .all()
        )
    mutation_targets = {
        match.group(1).strip('`"').lower()
        for statement in mutation_statements
        if (
            match := re.match(
                r"\s*(?:INSERT\s+INTO|UPDATE|DELETE\s+FROM)\s+([`\"\w]+)",
                statement,
                re.IGNORECASE,
            )
        )
    }
    assert mutation_targets <= {
        "script_build_record",
        "script_build_mission_binding",
        "script_build_artifact_version",
        "script_build_publication",
        "script_build_paragraph",
        "script_build_element",
        "script_build_paragraph_element",
        "script_build_task_plan_step",
    }
    assert not mutation_targets & {
        "script_build_branch",
        "script_build_round",
        "script_build_multipath",
    }
    assert BuildStatus(str(build["status"])) is BuildStatus.PARTIAL
    assert build["error_message"] == PHASE_TWO_CANDIDATE_PORTFOLIO_READY
    assert build["reson_trace_id"] == ROOT and build["end_time"] is not None
    assert branch_values and min(branch_values) > 0 and 0 not in branch_values
    assert final_publications == 0
    assert (
        sum(
            row["artifact_type"] == ArtifactKind.CANDIDATE_PORTFOLIO.value for row in phase_two_rows
        )
        == 1
    )
    for row in phase_two_rows:
        if row["artifact_type"] in {
            ArtifactKind.STRUCTURE.value,
            ArtifactKind.PARAGRAPH.value,
            ArtifactKind.ELEMENT_SET.value,
            ArtifactKind.STRUCTURED_SCRIPT.value,
        }:
            assert row["legacy_branch_id"] == row["id"] > 0
        else:
            assert row["legacy_branch_id"] is None


class _SingleBinding:
    def __init__(self, root: str) -> None:
        now = datetime.now(UTC)
        self.value = MissionBinding(1, 1, root, 11, None, None, "test", "v1", now, now)

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


class _BlockingExecutor:
    def __init__(self) -> None:
        self.started = asyncio.Event()
        self.release = asyncio.Event()

    async def run_worker(self, context: dict[str, Any]) -> WorkerRunResult:
        self.started.set()
        await self.release.wait()
        return WorkerRunResult(context["worker_trace_id"], "failed", error="test release")

    async def run_validator(self, context: dict[str, Any]) -> ValidatorRunResult:
        raise AssertionError(f"unexpected validation: {context}")

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


@pytest.mark.asyncio
async def test_same_task_cannot_reserve_parallel_attempts(tmp_path: Path) -> None:
    root = "parallel-attempt-root"
    task_store = FileSystemTaskStore(str(tmp_path / "ledger"))
    executor = _BlockingExecutor()
    coordinator = TaskCoordinator(
        task_store,
        FileSystemArtifactStore(str(tmp_path / "artifacts")),
        FileSystemTraceStore(str(tmp_path / "traces")),
        executor=executor,
    )
    await coordinator.ensure_ledger(
        root,
        {
            "objective": "parallel attempt guard",
            "acceptance_criteria": [
                {"criterion_id": "guard", "description": "guard", "hard": True}
            ],
            "context_refs": ["script-build://inputs/11"],
        },
    )
    planning = PhaseTwoPlanningService(
        coordinator=coordinator,
        bindings=cast(Any, _SingleBinding(root)),
        contracts=FileScriptTaskContractStore(tmp_path / "contracts"),
    )
    task = await planning.plan_script_tasks(
        contract_payloads=(_contract(ScriptTaskKind.DIRECTION),),
        parent_task_id=None,
        context={"root_trace_id": root},
    )
    task_id = task["task_ids"][0]
    first = asyncio.create_task(
        planning.dispatch_script_tasks(task_ids=(task_id,), context={"root_trace_id": root})
    )
    await asyncio.wait_for(executor.started.wait(), timeout=2)
    second = await planning.dispatch_script_tasks(
        task_ids=(task_id,), context={"root_trace_id": root}
    )
    assert "Dispatch conflict" in cast(str, second[0]["error"])
    executor.release.set()
    await first
    ledger = await task_store.load(root)
    assert len(ledger.tasks[task_id].attempt_ids) == 1