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

from __future__ import annotations

import asyncio
import re
from collections.abc import Mapping, Sequence
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace
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,
    DirectionArtifact,
    DirectionGoal,
    EvidenceRecordV1,
)
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]:
    del write_scope, closure, adopted, held, order, execution_ready
    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] = {
        "task_kind": kind.value,
        "scope_ref": scope,
        "intent_class": intent,
        "objective": f"produce independently verifiable {kind.value} content",
        "input_decision_ids": [str(item["decision_id"]) for item in input_refs],
        "base_decision_id": next(
            (
                str(item["decision_id"])
                for item in input_refs
                if base_ref is not None
                and cast(dict[str, Any], item["artifact_ref"])["uri"] == base_ref.uri
            ),
            None,
        ),
        "gap_ref": None,
        "criteria": [
            {
                "criterion_id": "closed",
                "description": "the frozen increment is concrete and independently verifiable",
                "hard": True,
            }
        ],
        "goal_ids": (
            []
            if kind in {ScriptTaskKind.DIRECTION, ScriptTaskKind.DECODE_RETRIEVAL}
            else ["goal-1"]
        ),
        "supersedes_decision_ids": list(supersedes),
        "comparison_decision_ids": [str(item["decision_id"]) for item in comparison_refs],
    }
    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=DirectionArtifact(
                    goals=(
                        DirectionGoal(
                            "goal-1",
                            "Use a concrete reversal to reveal the topic",
                            "A visible reversal makes the direction observable",
                            success_criteria=("the opening reveals one concrete reversal",),
                        ),
                    ),
                    evidence_refs=(str(evidence_refs[0]["uri"]),),
                ),
            )
        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:
                atom = (
                    {"原子点": "可观察的反转", "维度": "开场", "维度类型": "主维度"},
                )
                created = await self.candidates.create_script_paragraphs(
                    paragraphs=(
                        {
                            "client_key": "opening",
                            "paragraph_index": 1,
                            "name": f"{kind.value} opening",
                            "content_range": {"scope": "opening"},
                            "level": 1,
                            "theme_elements": atom,
                            "form_elements": atom,
                            "function_elements": atom,
                            "feeling_elements": atom,
                        },
                    ),
                    context=candidate_context,
                )
                paragraph_id = created["paragraph_ids_by_client_key"]["opening"]
            if kind is ScriptTaskKind.PARAGRAPH:
                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
                if item["artifact_ref"]["kind"] == ArtifactKind.PARAGRAPH.value
            ]
            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,
        artifacts=artifacts,
        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_ref = await _dispatch_accept(planning, coordinator, direction)
    await bindings.set_active_direction(
        script_build_id=script_build_id,
        artifact_version_id=int(direction_ref.version),
    )

    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",),
        ),
        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",
    )
    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"},
        )

    adopted = (structure_decision, paragraph_decision, element_decision)
    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=None,
        child_contracts=(),
        selected_decision_ids=adopted,
        context={"root_trace_id": ROOT, "tool_call_id": "close-compose"},
    )
    compose_decision, compose_ref = await _dispatch_accept(planning, coordinator, compose)

    await planning.decide_script_task(
        task_id=portfolio,
        action=DecisionAction.REVISE.value,
        reason="freeze the uniquely adopted StructuredScript",
        validation_id=None,
        replacement_contract=None,
        child_contracts=(),
        selected_decision_ids=(compose_decision,),
        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"),
        artifacts=cast(Any, SimpleNamespace()),
    )
    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