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

from __future__ import annotations

from dataclasses import replace
from datetime import UTC, datetime
from hashlib import sha256
from types import SimpleNamespace
from typing import Any, cast

import pytest
from agent.orchestration import ArtifactRef

from script_build_host.application.phase_two_candidates import (
    CandidateScope,
    PhaseTwoCandidateError,
    PhaseTwoCandidateService,
    _candidate_state_revision,
)
from script_build_host.application.phase_two_inputs import (
    ActiveFrontierResolver,
    PhaseTwoInputError,
)
from script_build_host.domain.artifacts import (
    ArtifactKind,
    ArtifactState,
    ArtifactVersion,
    DirectionArtifact,
    DirectionGoal,
    EvidenceRecordV1,
)
from script_build_host.domain.errors import ProtocolViolation
from script_build_host.domain.phase_two_artifacts import (
    CandidateLineageV1,
    CandidatePortfolioArtifactV1,
    ComparisonArtifactV1,
    ElementSetArtifactV1,
    GoalCoverage,
    ParagraphArtifactV1,
    ScriptElementV1,
    ScriptParagraphElementLinkV1,
    ScriptParagraphV1,
    StructureArtifactV1,
    StructuredScriptArtifactV1,
)
from script_build_host.domain.task_contracts import (
    AcceptedDecisionRef,
    AcceptedInput,
    AcceptedInputBundleV1,
    ScriptCriterion,
    ScriptIntentClass,
    ScriptTaskBudget,
    ScriptTaskContractV1,
    ScriptTaskKind,
)
from script_build_host.domain.workbench import paragraph_target_key, semantic_handle
from script_build_host.domain.workspaces import CandidateWriteContext, WorkspaceError
from script_build_host.infrastructure.canonical_json import canonical_sha256
from script_build_host.repositories.sqlalchemy import (
    SqlAlchemyScriptBusinessArtifactRepository,
)
from script_build_host.repositories.workspace import (
    SqlAlchemyCandidateWorkspaceRepository,
)

DIGEST = "sha256:" + "a" * 64
SCOPE = "script-build://scopes/opening"
WRITE = "script-build://writes/paragraphs/opening"


def _complete_paragraph() -> ScriptParagraphV1:
    atom = ({"原子点": "内容", "维度": "开场", "维度类型": "主维度"},)
    return ScriptParagraphV1(
        1,
        1,
        1,
        None,
        "opening",
        {"beats": ["抛出问题", "给出答案"]},
        theme="为什么要主动改变",
        form="问题与回答",
        function="建立紧迫感",
        feeling="坚定且真诚",
        theme_elements=atom,
        form_elements=atom,
        function_elements=atom,
        feeling_elements=atom,
        description="先指出停滞的代价。再给出可执行的转折。",
        full_description="你可能一直在等待更好的时机。但真正的改变来自今天的第一步。",
    )


def _complete_element() -> ScriptElementV1:
    return ScriptElementV1(1, "停滞代价", "实质", "核心论点")


def _complete_link() -> ScriptParagraphElementLinkV1:
    return ScriptParagraphElementLinkV1(1, 1)


def _lineage() -> CandidateLineageV1:
    return CandidateLineageV1(
        scope_ref=SCOPE,
        input_snapshot_ref="script-build://input-snapshots/11",
        input_closure_digest=DIGEST,
        write_scope=(WRITE,),
        goal_ids=("goal-1",),
    )


def _source_artifacts() -> tuple[
    StructureArtifactV1,
    ParagraphArtifactV1,
    ElementSetArtifactV1,
]:
    paragraph = _complete_paragraph()
    element = _complete_element()
    link = _complete_link()
    lineage = _lineage()
    return (
        StructureArtifactV1(lineage, (paragraph,)),
        ParagraphArtifactV1(
            lineage,
            (paragraph,),
            (element,),
            (link,),
            {"created": {"paragraph_ids": [1]}},
        ),
        ElementSetArtifactV1(
            lineage,
            (element,),
            (link,),
            (paragraph,),
            {"created": {"element_ids": [1]}},
        ),
    )


def _contract(
    kind: ScriptTaskKind,
    *,
    refs: tuple[AcceptedDecisionRef, ...] = (),
    comparison_refs: tuple[AcceptedDecisionRef, ...] = (),
    adopted: tuple[str, ...] = (),
    held: tuple[str, ...] = (),
    order: tuple[str, ...] = (),
) -> ScriptTaskContractV1:
    schemas = {
        ScriptTaskKind.PARAGRAPH: "paragraph-artifact/v1",
        ScriptTaskKind.DECODE_RETRIEVAL: "evidence-record/v1",
        ScriptTaskKind.COMPARE: "comparison-artifact/v1",
        ScriptTaskKind.COMPOSE: "structured-script/v1",
        ScriptTaskKind.CANDIDATE_PORTFOLIO: "candidate-portfolio/v1",
    }
    return ScriptTaskContractV1(
        task_kind=kind,
        scope_ref=SCOPE,
        intent_class=(
            ScriptIntentClass.COMPOSE
            if kind is ScriptTaskKind.COMPOSE
            else ScriptIntentClass.COMPARE
            if kind is ScriptTaskKind.COMPARE
            else ScriptIntentClass.EXPLORE
        ),
        objective="produce one bounded candidate increment",
        input_decision_refs=refs if kind is ScriptTaskKind.PARAGRAPH else (),
        base_artifact_ref=None,
        write_scope=(WRITE,),
        gap_ref=None,
        output_schema=schemas[kind],
        criteria=(ScriptCriterion("closed", "output is concretely complete"),),
        budget=ScriptTaskBudget(),
        goal_ids=("goal-1",),
        candidate_closure_decision_refs=(
            refs if kind in {ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO} else ()
        ),
        adopted_decision_ids=adopted,
        held_or_rejected_decision_ids=held,
        compose_order=order,
        comparison_decision_refs=comparison_refs,
    )


class _Bindings:
    def __init__(self, active_direction_artifact_version_id: int = 10) -> None:
        self.active_direction_artifact_version_id = active_direction_artifact_version_id

    async def get_by_root(self, root_trace_id: str) -> Any:
        return SimpleNamespace(
            root_trace_id=root_trace_id,
            script_build_id=7,
            input_snapshot_id=11,
            active_direction_artifact_version_id=self.active_direction_artifact_version_id,
        )


class _TaskStore:
    def __init__(self, ledger: Any) -> None:
        self.ledger = ledger

    async def load(self, root_trace_id: str) -> Any:
        assert root_trace_id == "root"
        return self.ledger


class _AcceptedInputs:
    def __init__(
        self, contracts: dict[str, ScriptTaskContractV1], bundle: AcceptedInputBundleV1
    ) -> None:
        self.contracts = contracts
        self.bundle = bundle

    async def contract_for_task(self, *, task: Any, **_: Any) -> ScriptTaskContractV1:
        return self.contracts[task.task_id]

    async def resolve(self, **_: Any) -> AcceptedInputBundleV1:
        return self.bundle


class _FrameworkArtifacts:
    async def get(self, *_: Any) -> Any:
        raise AssertionError("framework snapshot should not be read")


class _SnapshotArtifacts:
    def __init__(self, ref: ArtifactRef) -> None:
        self.ref = ref

    async def get(self, root_trace_id: str, snapshot_id: str) -> Any:
        assert (root_trace_id, snapshot_id) == ("root", "validation-snapshot")
        return SimpleNamespace(artifact_refs=[self.ref], evidence_refs=[])


class _RawStore:
    def __init__(self, content: bytes) -> None:
        self.content = content

    async def read_bytes(self, ref: str) -> bytes:
        assert ref.endswith(sha256(self.content).hexdigest())
        return self.content


class _Artifacts:
    def __init__(self, versions: dict[str, ArtifactVersion]) -> None:
        self.versions = versions
        self.frozen: ArtifactVersion | None = None

    async def read_by_ref(self, ref: ArtifactRef, **_: Any) -> ArtifactVersion:
        return self.versions[ref.uri]

    async def get_by_attempt(self, **owners: Any) -> ArtifactVersion:
        for value in self.versions.values():
            if value.attempt_id == owners["attempt_id"]:
                return value
        raise AssertionError("Attempt artifact not found")

    async def get_by_id(self, identifier: int, **_: Any) -> ArtifactVersion:
        for value in self.versions.values():
            if value.artifact_version_id == identifier:
                return value
        if identifier != 10:
            raise AssertionError("Artifact version not found")
        direction = DirectionArtifact(
            goals=(
                DirectionGoal(
                    "goal-1",
                    "complete the script",
                    "the delivery needs one complete script",
                    success_criteria=("the script is complete",),
                ),
            ),
            evidence_refs=("script-build://artifact-versions/90",),
        )
        return ArtifactVersion(
            10,
            7,
            "direction",
            "direction-attempt",
            1,
            ArtifactKind.DIRECTION,
            DIGEST,
            ArtifactState.FROZEN,
            direction,
            datetime.now(UTC),
            datetime.now(UTC),
        )

    async def freeze(self, *, artifact: Any, **owners: Any) -> tuple[ArtifactVersion, ArtifactRef]:
        digest = canonical_sha256(artifact.content_payload()).wire
        artifact = replace(artifact, canonical_sha256=digest)
        artifact_kind = (
            ArtifactKind.COMPARISON
            if isinstance(artifact, ComparisonArtifactV1)
            else ArtifactKind.CANDIDATE_PORTFOLIO
        )
        version = ArtifactVersion(
            artifact_version_id=99,
            script_build_id=owners["script_build_id"],
            task_id=owners["task_id"],
            attempt_id=owners["attempt_id"],
            spec_version=owners["spec_version"],
            artifact_type=artifact_kind,
            canonical_sha256=digest,
            state=ArtifactState.FROZEN,
            artifact=artifact,
            created_at=datetime.now(UTC),
            frozen_at=datetime.now(UTC),
        )
        self.frozen = version
        ref = ArtifactRef(
            "script-build://artifact-versions/99",
            artifact_kind.value,
            "99",
            digest,
        )
        self.versions[ref.uri] = version
        return version, ref


def _scope_fixture(
    contract: ScriptTaskContractV1,
    *,
    bundle: AcceptedInputBundleV1 | None = None,
    task_id: str = "task",
    attempt_id: str = "attempt",
) -> tuple[Any, dict[str, Any]]:
    task = SimpleNamespace(task_id=task_id, attempt_ids=[attempt_id])
    attempt = SimpleNamespace(
        attempt_id=attempt_id,
        task_id=task_id,
        spec_version=1,
        accepted_child_decision_ids=(),
    )
    ledger = SimpleNamespace(
        tasks={task_id: task},
        attempts={attempt_id: attempt},
        decisions={},
        revision=1,
    )
    accepted = _AcceptedInputs(
        {task_id: contract},
        bundle or AcceptedInputBundleV1((), DIGEST),
    )
    return (ledger, accepted), {
        "root_trace_id": "root",
        "task_id": task_id,
        "attempt_id": attempt_id,
        "spec_version": 1,
    }


@pytest.mark.asyncio
async def test_phase_one_manifest_does_not_require_candidate_workspace() -> None:
    contract = _contract(ScriptTaskKind.DECODE_RETRIEVAL)
    (ledger, accepted), context = _scope_fixture(contract)
    evidence = EvidenceRecordV1(
        evidence_id="evidence",
        source_type="decode",
        tool_name="retrieve_decode",
        query={},
        source_refs=("source",),
        raw_artifact_ref=None,
        summary="bounded evidence",
        supports=("goal",),
        confidence="high",
        limitations=(),
        content_sha256=DIGEST,
        created_at=datetime.now(UTC),
    )
    version = ArtifactVersion(
        1,
        7,
        "task",
        "attempt",
        1,
        ArtifactKind.EVIDENCE,
        DIGEST,
        ArtifactState.FROZEN,
        evidence,
        datetime.now(UTC),
        datetime.now(UTC),
    )
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=cast(Any, _Artifacts({"script-build://artifact-versions/1": version})),
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=None,
    )
    manifest = await service.resolve_attempt_manifest(context=context)
    assert manifest.artifact_ref.kind == "evidence"
    assert manifest.scope_ref == SCOPE


@pytest.mark.asyncio
async def test_paragraph_workspace_uses_positive_branch_and_freezes_on_manifest(
    database: Any,
) -> None:
    _, sessions = database
    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository)
    contract = _contract(ScriptTaskKind.PARAGRAPH)
    (ledger, accepted), context = _scope_fixture(contract)
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=repository,
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=workspaces,
    )
    created = await service.create_script_paragraph(
        payload={"paragraph_index": 1, "name": "opening", "content_range": {}},
        context=context,
    )
    manifest = await service.resolve_attempt_manifest(context=context)
    version = await repository.get_by_attempt(
        script_build_id=7, task_id="task", attempt_id="attempt"
    )
    assert created["paragraph_id"] > 0
    assert version.artifact_version_id > 0
    assert manifest.artifact_ref.digest == version.canonical_sha256
    assert manifest.artifact_ref.kind == "paragraph"

    with pytest.raises(PhaseTwoCandidateError, match="WRITE_SCOPE_VIOLATION"):
        await service.create_script_paragraph(
            payload={
                "paragraph_index": 2,
                "name": "forged",
                "content_range": {},
                "branch_id": 0,
            },
            context=context,
        )


@pytest.mark.asyncio
async def test_semantic_paragraph_save_is_atomic_and_exact_replay_is_idempotent(
    database: Any,
) -> None:
    _, sessions = database
    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository)
    contract = _contract(ScriptTaskKind.PARAGRAPH)
    (ledger, accepted), context = _scope_fixture(contract)
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=repository,
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=workspaces,
    )
    scope = await service._workspace_scope(context, paragraph=True)
    before = await workspaces.snapshot(scope.write_context)
    revision = _candidate_state_revision(scope, before, ledger.revision)
    atom = [{"原子点": "observable reversal", "维度": "opening", "维度类型": "主维度"}]
    payload = (
        {
            "client_key": "opening",
            "paragraph_index": 1,
            "name": "opening",
            "content_range": {"scope": "opening"},
            "level": 1,
            "theme_elements": atom,
            "form_elements": atom,
            "function_elements": atom,
            "feeling_elements": [{"原子点": "curiosity", "维度": "tone"}],
            "theme": "A visible detail changes the original assumption.",
            "form": "Concrete setup followed by a concise reversal.",
            "function": "Give the audience a reason to continue.",
            "feeling": "Measured surprise grounded in observation.",
            "description": "The opening overturns one assumption with evidence.",
            "full_description": (
                "Start from a familiar expectation and show exactly how the observed "
                "detail changes it."
            ),
        },
    )
    first = await service.save_script_paragraphs(
        paragraphs=payload,
        expected_state_revision=revision,
        context=context,
    )
    replay = await service.save_script_paragraphs(
        paragraphs=payload,
        expected_state_revision=revision,
        context=context,
    )

    assert first["status"] == "saved"
    assert first["created"] == 1
    assert replay["status"] == "not_modified"
    assert replay["created"] == replay["updated"] == 0
    assert replay["changed_target_keys"] == first["changed_target_keys"]
    assert len((await workspaces.snapshot(scope.write_context)).paragraphs) == 1

    changed_payload = (dict(payload[0], name="changed after stale projection"),)
    with pytest.raises(PhaseTwoCandidateError, match="STALE_WORKBENCH_STATE"):
        await service.save_script_paragraphs(
            paragraphs=changed_payload,
            expected_state_revision=revision,
            context=context,
        )


@pytest.mark.asyncio
async def test_paragraph_batch_maps_parent_keys_and_rolls_back_atomically(database: Any) -> None:
    _, sessions = database
    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository)
    contract = _contract(ScriptTaskKind.PARAGRAPH)
    (ledger, accepted), context = _scope_fixture(contract)
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=repository,
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=workspaces,
    )
    created = await service.create_script_paragraphs(
        paragraphs=(
            {
                "client_key": "root",
                "paragraph_index": 1,
                "name": "opening",
                "content_range": {"beats": ["setup", "turn"]},
                "theme": "具体主题正文",
                "form": "对话漫画",
                "function": "建立冲突",
                "feeling": "好奇",
                "description": "角色在器械旁发生可见冲突。",
                "full_description": "角色先指责别人占用器械,随后被发现自己也在刷手机。",
            },
            {
                "client_key": "child",
                "parent_client_key": "root",
                "paragraph_index": 2,
                "level": 2,
                "name": "turn",
                "content_range": {"beats": ["turn"]},
            },
        ),
        context=context,
    )
    assert created["created"] == 2
    assert set(created["paragraph_ids_by_client_key"]) == {"root", "child"}
    workspace = await service.read_attempt_workspace(context=context)
    root_row = next(item for item in workspace["paragraphs"] if item["name"] == "opening")
    assert root_row["theme"] == "具体主题正文"
    assert root_row["full_description"].endswith("自己也在刷手机。")

    contract2 = replace(contract, objective="second isolated workspace")
    (ledger2, accepted2), context2 = _scope_fixture(
        contract2, task_id="task-2", attempt_id="attempt-2"
    )
    service2 = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger2),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=repository,
        accepted_inputs=accepted2,
        active_frontier=ActiveFrontierResolver(),
        workspaces=workspaces,
    )
    with pytest.raises(WorkspaceError, match="child content range"):
        await service2.create_script_paragraphs(
            paragraphs=(
                {
                    "client_key": "root",
                    "paragraph_index": 1,
                    "name": "root",
                    "content_range": {"beats": ["setup"]},
                },
                {
                    "client_key": "child",
                    "parent_client_key": "root",
                    "paragraph_index": 2,
                    "level": 2,
                    "name": "child",
                    "content_range": {"beats": ["missing"]},
                },
            ),
            context=context2,
        )
    write_context = CandidateWriteContext(
        script_build_id=7,
        task_id="task-2",
        attempt_id="attempt-2",
        spec_version=1,
        objective=contract2.objective,
        input_refs=(),
        write_scope=contract2.write_scope,
    )
    assert not (await workspaces.snapshot(write_context)).paragraphs


@pytest.mark.asyncio
async def test_element_set_materializes_all_accepted_paragraphs_before_linking(
    database: Any,
) -> None:
    _, sessions = database
    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository)
    paragraph = replace(
        _complete_paragraph(),
        theme_elements=(({"原子点": "内容", "维度": "主题", "维度类型": "主维度"}),),
        form_elements=(({"原子点": "对话", "维度": "形式", "维度类型": "主维度"}),),
        function_elements=(({"原子点": "冲突", "维度": "作用", "维度类型": "主维度"}),),
        feeling_elements=(({"原子点": "好奇", "维度": "感受"}),),
    )
    _, paragraph_ref = await repository.freeze(
        script_build_id=7,
        task_id="paragraph-source",
        attempt_id="paragraph-source-attempt",
        spec_version=1,
        artifact=ParagraphArtifactV1(
            _lineage(),
            (paragraph,),
            (),
            (),
            {"created": {"paragraph_ids": [1]}},
        ),
    )
    paragraph_ref = ArtifactRef(
        paragraph_ref.uri,
        paragraph_ref.kind,
        paragraph_ref.version,
        paragraph_ref.digest,
    )
    accepted_ref = AcceptedDecisionRef(
        "paragraph-decision",
        paragraph_ref,
        SCOPE,
        ScriptTaskKind.PARAGRAPH,
    )
    contract = ScriptTaskContractV1(
        task_kind=ScriptTaskKind.ELEMENT_SET,
        scope_ref=SCOPE,
        intent_class=ScriptIntentClass.EXPLORE,
        objective="attach concrete elements to every accepted paragraph",
        input_decision_refs=(accepted_ref,),
        base_artifact_ref=None,
        write_scope=("script-build://writes/elements",),
        gap_ref=None,
        output_schema="element-set-artifact/v1",
        criteria=(ScriptCriterion("closed", "all elements are linked"),),
        budget=ScriptTaskBudget(),
        goal_ids=("goal-1",),
    )
    bundle = AcceptedInputBundleV1(
        (
            AcceptedInput(
                "paragraph-decision",
                paragraph_ref,
                ScriptTaskKind.PARAGRAPH,
                SCOPE,
                "explicit",
            ),
        ),
        DIGEST,
    )
    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=repository,
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=workspaces,
    )

    workspace = await service.read_attempt_workspace(context=context)
    assert len(workspace["paragraphs"]) == 1
    paragraph_id = workspace["paragraphs"][0]["paragraph_id"]
    assert paragraph_id > 0
    assert workspace["paragraphs"][0]["full_description"]

    with pytest.raises(WorkspaceError, match="link paragraph"):
        await service.create_script_elements(
            elements=(
                {
                    "client_key": "invalid",
                    "name": "不应落库的元素",
                    "dimension_primary": "实质",
                    "dimension_secondary": "回滚验证",
                },
            ),
            links=(
                {
                    "paragraph_id": paragraph_id + 999,
                    "element_client_keys": ["invalid"],
                },
            ),
            context=context,
        )
    rolled_back = await service.read_attempt_workspace(context=context)
    assert rolled_back["elements"] == []
    assert rolled_back["paragraph_element_links"] == []

    scope = await service._workspace_scope(context, element=True)
    before = await workspaces.snapshot(scope.write_context)
    revision = _candidate_state_revision(scope, before, ledger.revision)
    target = paragraph_target_key("root", "task", "attempt", paragraph_id)
    elements = (
        {
            "client_key": "machine",
            "name": "被占用的高位下拉机",
            "dimension_primary": "实质",
            "dimension_secondary": "冲突道具",
        },
        {
            "client_key": "bubble",
            "name": "占着不练的对话气泡",
            "dimension_primary": "形式",
            "dimension_secondary": "视觉文字",
        },
    )
    links = (
        {
            "paragraph_target_key": target,
            "element_client_keys": ["machine", "bubble"],
        },
    )
    created = await service.save_script_elements(
        elements=elements,
        links=links,
        expected_state_revision=revision,
        context=context,
    )
    replay = await service.save_script_elements(
        elements=elements,
        links=links,
        expected_state_revision=revision,
        context=context,
    )
    assert created["status"] == "saved"
    assert created["created"] == 2
    assert created["linked"] == 2
    assert replay["status"] == "not_modified"
    final = await workspaces.snapshot(scope.write_context)
    assert len(final.elements) == 2
    assert len(final.links) == 2

    changed_elements = (dict(elements[0], name="changed after stale projection"), elements[1])
    with pytest.raises(PhaseTwoCandidateError, match="STALE_WORKBENCH_STATE"):
        await service.save_script_elements(
            elements=changed_elements,
            links=links,
            expected_state_revision=revision,
            context=context,
        )

    manifest = await service.resolve_attempt_manifest(context=context)
    assert manifest.artifact_ref.kind == ArtifactKind.ELEMENT_SET.value


@pytest.mark.asyncio
async def test_element_seed_lineage_survives_full_compose_projection(database: Any) -> None:
    _, sessions = database
    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
    paragraph = _complete_paragraph()

    structure_version, structure_ref = await artifacts.freeze(
        script_build_id=7,
        task_id="structure-source",
        attempt_id="structure-attempt",
        spec_version=1,
        artifact=StructureArtifactV1(_lineage(), (paragraph,)),
    )
    paragraph_lineage = replace(
        _lineage(),
        base_artifact_ref=structure_ref.uri,
        base_artifact_digest=structure_ref.digest,
        base_revision=structure_version.artifact_version_id,
        source_lineage=(
            {"entity": "paragraph", "local_id": 1, "source_local_id": 1},
        ),
    )
    paragraph_version, paragraph_ref = await artifacts.freeze(
        script_build_id=7,
        task_id="paragraph-source",
        attempt_id="paragraph-attempt",
        spec_version=1,
        artifact=ParagraphArtifactV1(
            paragraph_lineage,
            (paragraph,),
            (),
            (),
            {"created": {"paragraph_ids": [1]}},
        ),
    )

    element_context = CandidateWriteContext(
        script_build_id=7,
        task_id="element-task",
        attempt_id="element-attempt",
        spec_version=1,
        objective="link one grounded element",
        input_refs=(paragraph_ref.uri,),
        write_scope=("script-build://writes/elements",),
    )
    await workspaces.get_or_create(element_context, artifact_kind=ArtifactKind.ELEMENT_SET)
    seeded = await workspaces.seed_from_accepted_artifacts(
        element_context, (paragraph_version,)
    )
    seeded_id = seeded.paragraphs[0].paragraph_id
    await workspaces.create_elements(
        element_context,
        (
            {
                "client_key": "element",
                "name": "grounded element",
                "dimension_primary": "实质",
                "dimension_secondary": "观察",
            },
        ),
        ({"paragraph_id": seeded_id, "element_client_keys": ["element"]},),
    )
    element_version, element_ref = await workspaces.freeze(
        element_context,
        lineage=replace(
            _lineage(),
            write_scope=("script-build://writes/elements",),
        ),
    )
    assert element_version.artifact.lineage.source_lineage[0]["source_artifact_ref"] == (
        paragraph_ref.uri
    )

    compose_contract = replace(
        _contract(ScriptTaskKind.COMPOSE),
        write_scope=("script-build://writes",),
    )
    compose_context = CandidateWriteContext(
        script_build_id=7,
        task_id="compose-task",
        attempt_id="compose-attempt",
        spec_version=1,
        objective=compose_contract.objective,
        input_refs=(structure_ref.uri, paragraph_ref.uri, element_ref.uri),
        write_scope=compose_contract.write_scope,
    )
    await workspaces.get_or_create(
        compose_context, artifact_kind=ArtifactKind.STRUCTURED_SCRIPT
    )
    scope = CandidateScope(
        root_trace_id="root",
        script_build_id=7,
        input_snapshot_id="11",
        task=SimpleNamespace(task_id="compose-task"),
        attempt=SimpleNamespace(attempt_id="compose-attempt", spec_version=1),
        contract=compose_contract,
        write_context=compose_context,
    )
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=cast(Any, _TaskStore(SimpleNamespace())),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=artifacts,
        accepted_inputs=cast(Any, None),
        active_frontier=ActiveFrontierResolver(),
        workspaces=workspaces,
    )
    await service._materialize(
        scope,
        (structure_version, paragraph_version, element_version),
        (),
    )
    composed = await workspaces.snapshot(compose_context)
    assert len(composed.paragraphs) == 1
    assert len(composed.elements) == 1
    assert len(composed.links) == 1


@pytest.mark.asyncio
async def test_element_seed_disambiguates_same_local_id_and_replays_idempotently(
    database: Any,
) -> None:
    _, sessions = database
    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
    versions: list[ArtifactVersion] = []
    refs: list[ArtifactRef] = []
    for index in (1, 2):
        paragraph = replace(
            _complete_paragraph(),
            name=f"paragraph-{index}",
            full_description=f"complete source paragraph {index}",
        )
        version, ref = await artifacts.freeze(
            script_build_id=7,
            task_id=f"paragraph-{index}",
            attempt_id=f"paragraph-attempt-{index}",
            spec_version=1,
            artifact=ParagraphArtifactV1(
                _lineage(),
                (paragraph,),
                (),
                (),
                {"created": {"paragraph_ids": [1]}},
            ),
        )
        versions.append(version)
        refs.append(ref)
    context = CandidateWriteContext(
        script_build_id=7,
        task_id="element-multi",
        attempt_id="element-multi-attempt",
        spec_version=1,
        objective="seed two accepted paragraphs",
        input_refs=tuple(item.uri for item in refs),
        write_scope=("script-build://writes/elements",),
    )
    await workspaces.get_or_create(context, artifact_kind=ArtifactKind.ELEMENT_SET)
    first = await workspaces.seed_from_accepted_artifacts(context, tuple(versions))
    replay = await workspaces.seed_from_accepted_artifacts(context, tuple(reversed(versions)))

    assert len(first.paragraphs) == len(replay.paragraphs) == 2
    assert {item["source_local_id"] for item in first.source_identity_map} == {1}
    assert {item["source_artifact_ref"] for item in first.source_identity_map} == {
        item.uri for item in refs
    }


@pytest.mark.asyncio
async def test_candidate_service_discards_existing_draft_workspace_on_abandoned_attempt(
    database: Any,
) -> None:
    _, sessions = database
    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    workspaces = SqlAlchemyCandidateWorkspaceRepository(sessions, repository)
    contract = _contract(ScriptTaskKind.PARAGRAPH)
    (ledger, accepted), context = _scope_fixture(contract)
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=repository,
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=workspaces,
    )
    await service.create_script_paragraph(
        payload={"paragraph_index": 1, "name": "draft", "content_range": {}},
        context=context,
    )
    await service.discard_attempt(
        root_trace_id="root",
        task_id="task",
        attempt_id="attempt",
        reason="validator blocked the bounded increment",
    )
    write_context = CandidateWriteContext(
        script_build_id=7,
        task_id="task",
        attempt_id="attempt",
        spec_version=1,
        objective=contract.objective,
        input_refs=(),
        write_scope=contract.write_scope,
    )
    assert (await workspaces.require(write_context)).state is ArtifactState.DISCARDED
    with pytest.raises(WorkspaceError, match="ATTEMPT_WORKSPACE_FROZEN"):
        await service.create_script_paragraph(
            payload={"paragraph_index": 2, "name": "late", "content_range": {}},
            context=context,
        )


@pytest.mark.asyncio
async def test_frozen_image_must_be_accepted_and_digest_and_mime_are_rechecked() -> None:
    content = b"\x89PNG\r\n\x1a\n" + b"safe"
    raw_ref = f"script-build://raw-artifacts/sha256/{sha256(content).hexdigest()}"
    ref = ArtifactRef("script-build://artifact-versions/2", "evidence", "2", DIGEST)
    item = AcceptedInput("d1", ref, ScriptTaskKind.DECODE_RETRIEVAL, SCOPE, "explicit")
    bundle = AcceptedInputBundleV1((item,), DIGEST)
    contract = _contract(ScriptTaskKind.PARAGRAPH)
    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
    evidence = EvidenceRecordV1(
        "e",
        "decode",
        "retrieve_decode",
        {},
        ("source",),
        raw_ref,
        "summary",
        ("goal",),
        "high",
        (),
        DIGEST,
        datetime.now(UTC),
    )
    version = ArtifactVersion(
        2,
        7,
        "child",
        "child-attempt",
        1,
        ArtifactKind.EVIDENCE,
        DIGEST,
        ArtifactState.FROZEN,
        evidence,
        datetime.now(UTC),
        datetime.now(UTC),
    )
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=cast(Any, _Artifacts({ref.uri: version})),
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=None,
        raw_artifacts=_RawStore(content),
    )
    images = await service.view_frozen_images(
        image_handles=[semantic_handle("image", raw_ref)], context=context
    )
    assert images[0]["media_type"] == "image/png"
    assert "url" not in images[0]
    with pytest.raises(PhaseTwoCandidateError, match="outside the accepted"):
        await service.view_frozen_images(
            image_handles=[semantic_handle("image", _RAW_OTHER)], context=context
        )


@pytest.mark.asyncio
async def test_active_frontier_write_conflict_is_not_resolved_by_completion_order() -> None:
    ref1 = ArtifactRef("script-build://artifact-versions/1", "paragraph", "1", DIGEST)
    ref2 = ArtifactRef("script-build://artifact-versions/2", "paragraph", "2", DIGEST)
    refs = (
        AcceptedDecisionRef("d1", ref1, SCOPE, ScriptTaskKind.PARAGRAPH),
        AcceptedDecisionRef("d2", ref2, SCOPE, ScriptTaskKind.PARAGRAPH),
    )
    contract = _contract(
        ScriptTaskKind.COMPOSE, refs=refs, adopted=("d1", "d2"), order=("d1", "d2")
    )
    bundle = AcceptedInputBundleV1(
        (
            AcceptedInput("d1", ref1, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit"),
            AcceptedInput("d2", ref2, ScriptTaskKind.PARAGRAPH, SCOPE, "explicit"),
        ),
        DIGEST,
    )
    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
    child_contract = _contract(ScriptTaskKind.PARAGRAPH)
    for decision_id in ("d1", "d2"):
        task_id = f"task-{decision_id}"
        ledger.tasks[task_id] = SimpleNamespace(task_id=task_id, attempt_ids=[f"a-{decision_id}"])
        ledger.attempts[f"a-{decision_id}"] = SimpleNamespace(
            attempt_id=f"a-{decision_id}", task_id=task_id, spec_version=1
        )
        ledger.decisions[decision_id] = SimpleNamespace(
            task_id=task_id, attempt_id=f"a-{decision_id}"
        )
        accepted.contracts[task_id] = child_contract
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=cast(Any, _Artifacts({})),
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=None,
    )
    with pytest.raises(PhaseTwoInputError, match="WRITE_SCOPE_CONFLICT"):
        await service.read_active_frontier(context=context)


@pytest.mark.asyncio
async def test_comparison_candidates_are_contract_derived_and_pass_fairness_precheck() -> None:
    candidate_refs = (
        ArtifactRef("script-build://artifact-versions/1", "paragraph", "1", DIGEST),
        ArtifactRef("script-build://artifact-versions/2", "paragraph", "2", DIGEST),
    )
    decision_refs = tuple(
        AcceptedDecisionRef(
            f"candidate-{index}",
            ref,
            SCOPE,
            ScriptTaskKind.PARAGRAPH,
        )
        for index, ref in enumerate(candidate_refs, start=1)
    )
    bundle = AcceptedInputBundleV1(
        tuple(
            AcceptedInput(
                decision_ref.decision_id,
                decision_ref.artifact_ref,
                ScriptTaskKind.PARAGRAPH,
                SCOPE,
                "explicit",
            )
            for decision_ref in decision_refs
        ),
        DIGEST,
    )
    contract = _contract(ScriptTaskKind.COMPARE, comparison_refs=decision_refs)
    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
    artifacts = _Artifacts({})
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=cast(Any, artifacts),
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=None,
    )

    criterion_results = [
        {
            "criterion_id": "closed",
            "candidate_results": [
                {"decision_id": "candidate-1", "reason": "uses one visible detail"},
                {"decision_id": "candidate-2", "reason": "remains more abstract"},
            ],
        }
    ]
    frozen = await service.save_comparison_candidate(
        payload={
            "criterion_results": criterion_results,
            "conflicts": ["both candidates write the opening scope"],
            "recommended_decision_id": "candidate-1",
        },
        context=context,
    )
    assert artifacts.frozen is not None
    comparison = artifacts.frozen.artifact
    assert isinstance(comparison, ComparisonArtifactV1)
    assert comparison.candidate_artifact_refs == tuple(ref.uri for ref in candidate_refs)
    assert [
        item["artifact_ref"] for item in comparison.criterion_results[0]["candidate_results"]
    ] == [ref.uri for ref in candidate_refs]

    frozen_ref = ArtifactRef.from_dict(frozen["artifact_ref"])
    service._framework_artifact_store = _SnapshotArtifacts(frozen_ref)
    rules = await service.deterministic_precheck(
        context={**context, "snapshot_id": "validation-snapshot"}
    )
    assert {str(item["rule_id"]): item["verdict"] for item in rules} == {
        "business-artifact-owner": "passed",
        "realized-content": "passed",
        "candidate-lineage": "passed",
        "comparison-fairness": "passed",
    }

    with pytest.raises(PhaseTwoCandidateError, match="derived from the frozen contract"):
        await service.save_comparison_candidate(
            payload={
                "candidate_artifact_refs": [candidate_refs[1].uri],
                "criterion_results": criterion_results,
                "recommended_decision_id": "candidate-2",
            },
            context=context,
        )


@pytest.mark.asyncio
async def test_portfolio_adoption_and_canonical_digest_come_from_frozen_contract() -> None:
    ref1 = ArtifactRef("script-build://artifact-versions/1", "structured_script", "1", DIGEST)
    ref2 = ArtifactRef("script-build://artifact-versions/2", "structured_script", "2", DIGEST)
    refs = (
        AcceptedDecisionRef("d1", ref1, SCOPE, ScriptTaskKind.COMPOSE),
        AcceptedDecisionRef("d2", ref2, SCOPE, ScriptTaskKind.COMPOSE),
    )
    contract = _contract(
        ScriptTaskKind.CANDIDATE_PORTFOLIO,
        refs=refs,
        adopted=("d1",),
        held=("d2",),
        order=("d1",),
    )
    bundle = AcceptedInputBundleV1(
        (
            AcceptedInput("d1", ref1, ScriptTaskKind.COMPOSE, SCOPE, "explicit"),
            AcceptedInput("d2", ref2, ScriptTaskKind.COMPOSE, SCOPE, "explicit"),
        ),
        DIGEST,
    )
    (ledger, accepted), context = _scope_fixture(contract, bundle=bundle)
    paragraph = _complete_paragraph()
    element = _complete_element()
    link = _complete_link()

    def source_versions(identifier: int) -> dict[str, ArtifactVersion]:
        base = 20 + identifier * 10
        versions: dict[str, ArtifactVersion] = {}
        for offset, (kind, artifact) in enumerate(
            zip(
                (
                    ArtifactKind.STRUCTURE,
                    ArtifactKind.PARAGRAPH,
                    ArtifactKind.ELEMENT_SET,
                ),
                _source_artifacts(),
                strict=True,
            ),
            start=1,
        ):
            version_id = base + offset
            uri = f"script-build://artifact-versions/{version_id}"
            versions[uri] = ArtifactVersion(
                version_id,
                7,
                f"source-{version_id}",
                f"source-attempt-{version_id}",
                1,
                kind,
                DIGEST,
                ArtifactState.FROZEN,
                artifact,
                datetime.now(UTC),
                datetime.now(UTC),
            )
        return versions

    def structured(identifier: int) -> ArtifactVersion:
        source_refs = tuple(source_versions(identifier))
        artifact = StructuredScriptArtifactV1(
            direction_ref="script-build://artifact-versions/10",
            input_closure_digest=DIGEST,
            paragraphs=(paragraph,),
            elements=(element,),
            paragraph_element_links=(link,),
            source_artifact_refs=source_refs,
            goal_coverage=(GoalCoverage("goal-1", (source_refs[1], source_refs[2])),),
            evidence_refs=(),
            acceptance_notes=("accepted",),
            canonical_sha256=DIGEST,
        )
        return ArtifactVersion(
            identifier,
            7,
            f"compose-{identifier}",
            f"compose-attempt-{identifier}",
            1,
            ArtifactKind.STRUCTURED_SCRIPT,
            DIGEST,
            ArtifactState.FROZEN,
            artifact,
            datetime.now(UTC),
            datetime.now(UTC),
        )

    versions = {
        **source_versions(1),
        **source_versions(2),
        ref1.uri: structured(1),
        ref2.uri: structured(2),
    }
    artifacts = _Artifacts(versions)
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings()),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=cast(Any, artifacts),
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=None,
    )
    result = await service.save_candidate_portfolio(
        payload={
            "unresolved_defects": [
                {
                    "severity": "warning",
                    "defect_code": "MINOR_STYLE",
                    "criterion_id": "closed",
                    "scope_ref": SCOPE,
                    "observed_excerpt": "opening cadence is slightly repetitive",
                    "evidence_refs": [],
                    "invalidated_inputs": [],
                    "recommended_action_class": "revise",
                }
            ],
        },
        context=context,
    )
    assert artifacts.frozen is not None
    portfolio = artifacts.frozen.artifact
    assert isinstance(portfolio, CandidatePortfolioArtifactV1)
    assert portfolio.accepted_decision_ids == ("d1",)
    assert portfolio.rejected_or_held_decision_ids == ("d2",)
    assert result["artifact_ref"]["digest"] == canonical_sha256(portfolio.content_payload()).wire

    with pytest.raises(PhaseTwoCandidateError, match="outside the accepted closure"):
        await service.save_candidate_portfolio(
            payload={
                "unresolved_defects": [
                    {
                        "severity": "warning",
                        "defect_code": "UNTRUSTED_EVIDENCE",
                        "criterion_id": "closed",
                        "scope_ref": SCOPE,
                        "observed_excerpt": "unsupported warning",
                        "evidence_refs": ["script-build://artifact-versions/999"],
                        "invalidated_inputs": [],
                        "recommended_action_class": "revise",
                    }
                ],
            },
            context=context,
        )

    with pytest.raises(ProtocolViolation, match=r"unsupported enum|must not be blank"):
        await service.save_candidate_portfolio(
            payload={
                "unresolved_defects": [
                    {"severity": "warning", "defect_code": "INCOMPLETE_WARNING"}
                ],
            },
            context=context,
        )

    with pytest.raises(PhaseTwoCandidateError, match="derived from the frozen contract"):
        await service.save_candidate_portfolio(
            payload={"adopted_structured_script_ref": ref2.uri}, context=context
        )


@pytest.mark.asyncio
async def test_candidate_portfolio_command_replay_reuses_one_frozen_version(database: Any) -> None:
    _, sessions = database
    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
    paragraph = _complete_paragraph()
    element = _complete_element()
    link = _complete_link()
    direction_version, direction_ref = await repository.freeze(
        script_build_id=7,
        task_id="direction",
        attempt_id="direction-attempt",
        spec_version=1,
        artifact=DirectionArtifact(
            goals=(
                DirectionGoal(
                    "goal-1",
                    "complete the script",
                    "the delivery needs one complete script",
                    success_criteria=("the script is complete",),
                ),
            ),
            evidence_refs=("script-build://artifact-versions/90",),
        ),
    )

    async def freeze_structured(identifier: int) -> tuple[ArtifactVersion, ArtifactRef]:
        structure, paragraph_source, element_source = _source_artifacts()
        source_refs: list[str] = []
        for suffix, source in (
            ("structure", structure),
            ("paragraph", paragraph_source),
            ("elements", element_source),
        ):
            _, source_ref = await repository.freeze(
                script_build_id=7,
                task_id=f"{suffix}-{identifier}",
                attempt_id=f"{suffix}-attempt-{identifier}",
                spec_version=1,
                artifact=source,
            )
            source_refs.append(source_ref.uri)
        return await repository.freeze(
            script_build_id=7,
            task_id=f"compose-{identifier}",
            attempt_id=f"compose-attempt-{identifier}",
            spec_version=1,
            artifact=StructuredScriptArtifactV1(
                direction_ref=direction_ref.uri,
                input_closure_digest=DIGEST,
                paragraphs=(paragraph,),
                elements=(element,),
                paragraph_element_links=(link,),
                source_artifact_refs=tuple(source_refs),
                goal_coverage=(GoalCoverage("goal-1", (source_refs[1], source_refs[2])),),
                evidence_refs=(),
                acceptance_notes=("accepted",),
            ),
        )

    (_, stored_ref1), (_, stored_ref2) = await freeze_structured(1), await freeze_structured(2)
    ref1 = ArtifactRef(
        stored_ref1.uri,
        stored_ref1.kind,
        stored_ref1.version,
        stored_ref1.digest,
    )
    ref2 = ArtifactRef(
        stored_ref2.uri,
        stored_ref2.kind,
        stored_ref2.version,
        stored_ref2.digest,
    )
    refs = (
        AcceptedDecisionRef("d1", ref1, SCOPE, ScriptTaskKind.COMPOSE),
        AcceptedDecisionRef("d2", ref2, SCOPE, ScriptTaskKind.COMPOSE),
    )
    contract = _contract(
        ScriptTaskKind.CANDIDATE_PORTFOLIO,
        refs=refs,
        adopted=("d1",),
        held=("d2",),
        order=("d1",),
    )
    bundle = AcceptedInputBundleV1(
        (
            AcceptedInput("d1", ref1, ScriptTaskKind.COMPOSE, SCOPE, "explicit"),
            AcceptedInput("d2", ref2, ScriptTaskKind.COMPOSE, SCOPE, "explicit"),
        ),
        DIGEST,
    )
    (ledger, accepted), context = _scope_fixture(
        contract,
        bundle=bundle,
        task_id="portfolio",
        attempt_id="portfolio-attempt",
    )
    service = PhaseTwoCandidateService(
        bindings=cast(Any, _Bindings(direction_version.artifact_version_id)),
        task_store=_TaskStore(ledger),
        framework_artifact_store=_FrameworkArtifacts(),
        artifacts=repository,
        accepted_inputs=accepted,
        active_frontier=ActiveFrontierResolver(),
        workspaces=None,
    )

    first = await service.save_candidate_portfolio(
        payload={"unresolved_defects": []}, context=context
    )
    replay = await service.save_candidate_portfolio(
        payload={"unresolved_defects": []}, context=context
    )
    frozen = await repository.get_by_attempt(
        script_build_id=7,
        task_id="portfolio",
        attempt_id="portfolio-attempt",
    )

    assert first == replay
    assert first["artifact_ref"]["version"] == str(frozen.artifact_version_id)
    assert frozen.artifact_type is ArtifactKind.CANDIDATE_PORTFOLIO


_RAW_OTHER = "script-build://raw-artifacts/sha256/" + "b" * 64