创作：实现候选写入、动态前沿、组合与作品集闭包

script_build_host/src/script_build_host/application/phase_two_candidates.py

@@ -0,0 +1,1552 @@
+"""Attempt-scoped candidate tools for phase-two creative work.
+
+The service deliberately accepts only framework protected context for owner
+identities.  Every read is closed over immutable ACCEPT decisions and every
+write is confined to the current Attempt workspace or its single immutable
+business Artifact.
+"""
+
+from __future__ import annotations
+
+import base64
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict, dataclass
+from hashlib import sha256
+from typing import Any, Protocol, cast
+
+from agent.orchestration import ArtifactRef
+
+from script_build_host.application.phase_two_inputs import (
+    AcceptedInputResolver,
+    ActiveFrontierResolver,
+)
+from script_build_host.domain.artifacts import (
+    ArtifactKind,
+    ArtifactState,
+    ArtifactVersion,
+    EvidenceRecordV1,
+)
+from script_build_host.domain.candidate_validation import (
+    comparison_fairness_errors,
+    placeholder_paths,
+)
+from script_build_host.domain.defects import ScriptDefectV1
+from script_build_host.domain.errors import ProtocolViolation, ScriptBuildError
+from script_build_host.domain.phase_two_artifacts import (
+    CandidateLineageV1,
+    CandidatePortfolioArtifactV1,
+    ComparisonArtifactV1,
+    ElementSetArtifactV1,
+    ParagraphArtifactV1,
+    ScriptElementV1,
+    ScriptParagraphV1,
+    StructureArtifactV1,
+    StructuredScriptArtifactV1,
+)
+from script_build_host.domain.phase_two_ports import AttemptManifest
+from script_build_host.domain.ports import (
+    MissionBindingRepository,
+    ScriptBusinessArtifactRepository,
+)
+from script_build_host.domain.task_contracts import (
+    AcceptedInput,
+    AcceptedInputBundleV1,
+    ScriptTaskContractV1,
+    ScriptTaskKind,
+)
+from script_build_host.domain.workspaces import (
+    CandidateWorkspaceSnapshot,
+    CandidateWriteContext,
+)
+
+_IDENTITY_FIELDS = frozenset(
+    {
+        "script_build_id",
+        "root_trace_id",
+        "task_id",
+        "attempt_id",
+        "spec_version",
+        "branch_id",
+        "artifact_version_id",
+        "input_snapshot_id",
+    }
+)
+_WORKSPACE_KIND: dict[ScriptTaskKind, ArtifactKind] = {
+    ScriptTaskKind.STRUCTURE: ArtifactKind.STRUCTURE,
+    ScriptTaskKind.PARAGRAPH: ArtifactKind.PARAGRAPH,
+    ScriptTaskKind.ELEMENT_SET: ArtifactKind.ELEMENT_SET,
+    ScriptTaskKind.COMPOSE: ArtifactKind.STRUCTURED_SCRIPT,
+}
+_MATERIALIZABLE = (
+    StructureArtifactV1,
+    ParagraphArtifactV1,
+    ElementSetArtifactV1,
+)
+_RAW_PREFIX = "script-build://raw-artifacts/sha256/"
+
+
+class PhaseTwoCandidateError(ScriptBuildError):
+    pass
+
+
+class CandidateWorkspaceOperations(Protocol):
+    async def get_or_create(
+        self,
+        context: CandidateWriteContext,
+        *,
+        artifact_kind: ArtifactKind,
+        base_artifact_ref: ArtifactRef | None = None,
+    ) -> Any: ...
+
+    async def require(self, context: CandidateWriteContext) -> Any: ...
+
+    async def snapshot(self, context: CandidateWriteContext) -> CandidateWorkspaceSnapshot: ...
+
+    async def create_paragraph(self, context: CandidateWriteContext, **values: Any) -> int: ...
+
+    async def batch_update_paragraphs(
+        self, context: CandidateWriteContext, updates: Sequence[Mapping[str, Any]]
+    ) -> tuple[int, ...]: ...
+
+    async def append_paragraph_atoms(
+        self, context: CandidateWriteContext, **values: Any
+    ) -> int: ...
+
+    async def delete_paragraph_atom(self, context: CandidateWriteContext, **values: Any) -> int: ...
+
+    async def create_element(self, context: CandidateWriteContext, **values: Any) -> int: ...
+
+    async def update_element(
+        self, context: CandidateWriteContext, *, element_id: int, values: Mapping[str, Any]
+    ) -> None: ...
+
+    async def batch_link(
+        self, context: CandidateWriteContext, links: Sequence[tuple[int, Sequence[int]]]
+    ) -> int: ...
+
+    async def delete_links(self, context: CandidateWriteContext, **values: Any) -> int: ...
+
+    async def freeze(
+        self,
+        context: CandidateWriteContext,
+        *,
+        lineage: CandidateLineageV1,
+        structured_script: dict[str, Any] | None = None,
+    ) -> tuple[ArtifactVersion, ArtifactRef]: ...
+
+    async def discard(self, context: CandidateWriteContext, *, reason: str) -> Any: ...
+
+
+class RawArtifactReader(Protocol):
+    async def read_bytes(self, ref: str) -> bytes: ...
+
+
+@dataclass(frozen=True, slots=True)
+class CandidateScope:
+    root_trace_id: str
+    script_build_id: int
+    input_snapshot_id: str
+    task: Any
+    attempt: Any
+    contract: ScriptTaskContractV1
+    write_context: CandidateWriteContext
+
+
+class PhaseTwoCandidateService:
+    """Implement the narrow Agent-facing candidate port."""
+
+    def __init__(
+        self,
+        *,
+        bindings: MissionBindingRepository,
+        task_store: Any,
+        framework_artifact_store: Any,
+        artifacts: ScriptBusinessArtifactRepository,
+        accepted_inputs: AcceptedInputResolver,
+        active_frontier: ActiveFrontierResolver,
+        workspaces: Any | None,
+        raw_artifacts: RawArtifactReader | None = None,
+        max_images: int = 8,
+        max_image_bytes: int = 10 * 1024 * 1024,
+        max_total_image_bytes: int = 25 * 1024 * 1024,
+    ) -> None:
+        self._bindings = bindings
+        self._task_store = task_store
+        self._framework_artifact_store = framework_artifact_store
+        self._artifacts = artifacts
+        self._accepted_inputs = accepted_inputs
+        self._active_frontier = active_frontier
+        self._workspaces = workspaces
+        self._raw_artifacts = raw_artifacts
+        self._max_images = max_images
+        self._max_image_bytes = max_image_bytes
+        self._max_total_image_bytes = max_total_image_bytes
+
+    async def read_accepted_input_bundle(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        scope, bundle = await self._resolve_bundle(context)
+        return {
+            "schema_version": "accepted-input-bundle/v1",
+            "task_kind": scope.contract.task_kind.value,
+            "scope_ref": scope.contract.scope_ref,
+            "input_snapshot_ref": f"script-build://inputs/{scope.input_snapshot_id}",
+            "input_closure_digest": bundle.input_closure_digest,
+            "inputs": [_accepted_manifest(item) for item in bundle.inputs],
+            "superseded_decision_ids": list(bundle.superseded_decision_ids),
+        }
+
+    async def read_active_frontier(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        scope, bundle = await self._resolve_bundle(context)
+        if scope.contract.task_kind is not ScriptTaskKind.COMPOSE:
+            raise PhaseTwoCandidateError(
+                "TASK_KIND_PRESET_MISMATCH", "active frontier is available to Compose only"
+            )
+        frontier, _ = await self._frontier(scope, bundle)
+        return {
+            "schema_version": "active-frontier/v1",
+            "scope_ref": scope.contract.scope_ref,
+            "input_closure_digest": bundle.input_closure_digest,
+            "compose_order": list(scope.contract.compose_order),
+            "inputs": [_accepted_manifest(item) for item in frontier],
+        }
+
+    async def read_pinned_candidates(
+        self, *, context: Mapping[str, Any]
+    ) -> Sequence[Mapping[str, Any]]:
+        scope, bundle = await self._resolve_bundle(context)
+        pinned_ids = {
+            item.decision_id
+            for item in (
+                *scope.contract.input_decision_refs,
+                *scope.contract.candidate_closure_decision_refs,
+                *scope.contract.comparison_decision_refs,
+            )
+        }
+        values: list[Mapping[str, Any]] = []
+        for item in bundle.inputs:
+            if item.decision_id not in pinned_ids:
+                continue
+            version = await self._read_accepted_version(scope, item)
+            values.append(
+                {
+                    **_accepted_manifest(item),
+                    "artifact": _artifact_payload(version),
+                }
+            )
+        if pinned_ids != {str(item["decision_id"]) for item in values}:
+            raise PhaseTwoCandidateError(
+                "CHILD_DECISION_INVALID", "one or more pinned candidates are outside the closure"
+            )
+        return values
+
+    async def view_frozen_images(
+        self,
+        *,
+        raw_artifact_refs: Sequence[str],
+        context: Mapping[str, Any],
+    ) -> Sequence[Mapping[str, Any]]:
+        if self._raw_artifacts is None:
+            raise PhaseTwoCandidateError(
+                "LEGACY_REFERENCE_INVALID", "raw Artifact storage is not configured"
+            )
+        if (
+            not raw_artifact_refs
+            or len(raw_artifact_refs) > self._max_images
+            or len(set(raw_artifact_refs)) != len(raw_artifact_refs)
+        ):
+            raise PhaseTwoCandidateError(
+                "LEGACY_REFERENCE_INVALID", "image references exceed the frozen view limit"
+            )
+        scope, bundle = await self._resolve_bundle(context)
+        allowed = await self._allowed_raw_refs(scope, bundle, context)
+        if any(ref not in allowed for ref in raw_artifact_refs):
+            raise PhaseTwoCandidateError(
+                "LEGACY_REFERENCE_INVALID", "image is outside the accepted frozen inputs"
+            )
+        total = 0
+        output: list[Mapping[str, Any]] = []
+        for ref in raw_artifact_refs:
+            content = await self._raw_artifacts.read_bytes(ref)
+            expected = ref.removeprefix(_RAW_PREFIX)
+            actual = sha256(content).hexdigest()
+            if expected == ref or actual != expected:
+                raise PhaseTwoCandidateError(
+                    "LEGACY_REFERENCE_INVALID", "raw image digest does not match its reference"
+                )
+            if not content or len(content) > self._max_image_bytes:
+                raise PhaseTwoCandidateError(
+                    "LEGACY_REFERENCE_INVALID", "raw image exceeds the per-image limit"
+                )
+            total += len(content)
+            if total > self._max_total_image_bytes:
+                raise PhaseTwoCandidateError(
+                    "LEGACY_REFERENCE_INVALID", "raw images exceed the total byte limit"
+                )
+            media_type = _image_media_type(content)
+            output.append(
+                {
+                    "type": "base64",
+                    "media_type": media_type,
+                    "data": base64.b64encode(content).decode("ascii"),
+                    "raw_artifact_ref": ref,
+                    "digest": f"sha256:{actual}",
+                    "size_bytes": len(content),
+                }
+            )
+        return output
+
+    async def deterministic_precheck(
+        self, *, context: Mapping[str, Any]
+    ) -> Sequence[Mapping[str, Any]]:
+        """Run owner-aware deterministic checks over the exact Validator snapshot."""
+
+        scope, bundle = await self._resolve_bundle(context)
+        snapshot = await self._framework_artifact_store.get(
+            scope.root_trace_id, _required_context(context, "snapshot_id")
+        )
+        if len(snapshot.artifact_refs) != 1 or snapshot.evidence_refs:
+            return (
+                {
+                    "rule_id": "business-artifact-owner",
+                    "verdict": "failed",
+                    "reason": "phase-two validation requires exactly one business Artifact",
+                },
+            )
+        ref = snapshot.artifact_refs[0]
+        version = await self._artifacts.read_by_ref(
+            ref,
+            script_build_id=scope.script_build_id,
+            task_id=scope.task.task_id,
+            attempt_id=scope.attempt.attempt_id,
+        )
+        expected_kind = _artifact_kind(scope.contract.task_kind)
+        owner_ok = (
+            version.state is ArtifactState.FROZEN
+            and version.artifact_type is expected_kind
+            and version.canonical_sha256 == ref.digest
+        )
+        rules: list[Mapping[str, Any]] = [
+            {
+                "rule_id": "business-artifact-owner",
+                "verdict": "passed" if owner_ok else "failed",
+                "reason": (
+                    "Artifact owner, kind, state and digest are closed"
+                    if owner_ok
+                    else "Artifact owner, kind, state or digest differs from the Attempt"
+                ),
+            }
+        ]
+        placeholders = placeholder_paths(version.artifact)
+        rules.append(
+            {
+                "rule_id": "realized-content",
+                "verdict": "failed" if placeholders else "passed",
+                "reason": (
+                    "Unrealized content at " + ", ".join(placeholders[:8])
+                    if placeholders
+                    else "Artifact contains no placeholder or deferred-production prose"
+                ),
+            }
+        )
+        lineage = getattr(version.artifact, "lineage", None)
+        if lineage is not None:
+            base = scope.contract.base_artifact_ref
+            lineage_ok = (
+                lineage.scope_ref == scope.contract.scope_ref
+                and lineage.input_snapshot_ref == f"script-build://inputs/{scope.input_snapshot_id}"
+                and lineage.input_closure_digest == bundle.input_closure_digest
+                and lineage.write_scope == scope.contract.write_scope
+                and lineage.supersedes_decision_ids == scope.contract.supersedes_decision_ids
+                and lineage.base_artifact_ref == (base.uri if base else None)
+                and lineage.base_artifact_digest == (base.digest if base else None)
+                and lineage.base_revision == (int(base.version) if base else None)
+            )
+            rules.append(
+                {
+                    "rule_id": "candidate-lineage",
+                    "verdict": "passed" if lineage_ok else "failed",
+                    "reason": (
+                        "Candidate lineage matches the frozen contract and input closure"
+                        if lineage_ok
+                        else "Candidate lineage differs from the frozen contract"
+                    ),
+                }
+            )
+        closure_digest = getattr(version.artifact, "input_closure_digest", None)
+        if closure_digest is not None:
+            rules.append(
+                {
+                    "rule_id": "input-closure-digest",
+                    "verdict": (
+                        "passed" if closure_digest == bundle.input_closure_digest else "failed"
+                    ),
+                    "reason": (
+                        "Artifact input closure matches the accepted bundle"
+                        if closure_digest == bundle.input_closure_digest
+                        else "Artifact input closure differs from the accepted bundle"
+                    ),
+                }
+            )
+        if isinstance(version.artifact, ComparisonArtifactV1):
+            fairness = comparison_fairness_errors(
+                version.artifact,
+                {item.criterion_id for item in scope.contract.criteria},
+            )
+            rules.append(
+                {
+                    "rule_id": "comparison-fairness",
+                    "verdict": "failed" if fairness else "passed",
+                    "reason": (
+                        "Invalid comparison fields: " + ", ".join(fairness[:8])
+                        if fairness
+                        else "Every criterion evaluates every frozen candidate symmetrically"
+                    ),
+                }
+            )
+        return tuple(rules)
+
+    async def read_attempt_workspace(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]:
+        scope = await self._scope(context)
+        if scope.contract.task_kind in {ScriptTaskKind.STRUCTURE, ScriptTaskKind.PARAGRAPH}:
+            scope = await self._workspace_scope(context, paragraph=True)
+        elif scope.contract.task_kind is ScriptTaskKind.ELEMENT_SET:
+            scope = await self._workspace_scope(context, element=True)
+        elif scope.contract.task_kind is ScriptTaskKind.COMPOSE:
+            scope = await self._workspace_scope(context, paragraph=True)
+        else:
+            raise PhaseTwoCandidateError(
+                "TASK_KIND_PRESET_MISMATCH", "this Task kind does not own a workspace"
+            )
+        snapshot = await self._workspace_repo().snapshot(scope.write_context)
+        return _workspace_payload(snapshot)
+
+    async def create_script_paragraph(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        _reject_identity(payload)
+        scope = await self._workspace_scope(context, paragraph=True)
+        paragraph_id = await self._workspace_repo().create_paragraph(
+            scope.write_context,
+            paragraph_index=_positive_int(payload.get("paragraph_index"), "paragraph_index"),
+            name=_required_text(payload, "name"),
+            content_range=_mapping(payload.get("content_range"), "content_range"),
+            level=_optional_int(payload.get("level"), default=1),
+            parent_paragraph_id=_nullable_positive_int(payload.get("parent_paragraph_id")),
+            theme_elements=_mapping_sequence(payload.get("theme_elements")),
+            form_elements=_mapping_sequence(payload.get("form_elements")),
+            function_elements=_mapping_sequence(payload.get("function_elements")),
+            feeling_elements=_mapping_sequence(payload.get("feeling_elements")),
+        )
+        return {"paragraph_id": paragraph_id}
+
+    async def append_paragraph_atoms(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        _reject_identity(payload)
+        scope = await self._workspace_scope(context, paragraph=True)
+        count = await self._workspace_repo().append_paragraph_atoms(
+            scope.write_context,
+            paragraph_id=_positive_int(payload.get("paragraph_id"), "paragraph_id"),
+            column=_required_text(payload, "column"),
+            atoms=_mapping_sequence(payload.get("atoms"), required=True),
+        )
+        return {"appended": count}
+
+    async def delete_paragraph_atom(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        _reject_identity(payload)
+        scope = await self._workspace_scope(context, paragraph=True)
+        count = await self._workspace_repo().delete_paragraph_atom(
+            scope.write_context,
+            paragraph_id=_positive_int(payload.get("paragraph_id"), "paragraph_id"),
+            column=_required_text(payload, "column"),
+            dimension=_required_text(payload, "dimension"),
+            atom_value=_required_text(payload, "atom_value"),
+        )
+        return {"deleted": count}
+
+    async def batch_update_script_paragraphs(
+        self, *, updates: Sequence[Mapping[str, Any]], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        for item in updates:
+            _reject_identity(item)
+        scope = await self._workspace_scope(context, paragraph=True)
+        ids = await self._workspace_repo().batch_update_paragraphs(scope.write_context, updates)
+        return {"paragraph_ids": list(ids), "updated": len(ids)}
+
+    async def create_script_element(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        _reject_identity(payload)
+        scope = await self._workspace_scope(context, element=True)
+        element_id = await self._workspace_repo().create_element(
+            scope.write_context,
+            name=_required_text(payload, "name"),
+            dimension_primary=_required_text(payload, "dimension_primary"),
+            dimension_secondary=_required_text(payload, "dimension_secondary"),
+            commonality_analysis=_optional_mapping(payload.get("commonality_analysis")),
+            topic_support=_optional_mapping(payload.get("topic_support")),
+            weight_score=_optional_mapping(payload.get("weight_score")),
+            support_elements=_mapping_sequence(payload.get("support_elements")),
+        )
+        return {"element_id": element_id}
+
+    async def update_script_element(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        _reject_identity(payload)
+        scope = await self._workspace_scope(context, element=True)
+        element_id = _positive_int(payload.get("element_id"), "element_id")
+        values = {key: value for key, value in payload.items() if key != "element_id"}
+        await self._workspace_repo().update_element(
+            scope.write_context, element_id=element_id, values=values
+        )
+        return {"element_id": element_id, "updated": True}
+
+    async def batch_link_paragraph_elements(
+        self, *, links: Sequence[Mapping[str, Any]], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        scope = await self._workspace_scope(context, element=True)
+        normalized: list[tuple[int, Sequence[int]]] = []
+        for item in links:
+            _reject_identity(item)
+            element_ids = item.get("element_ids")
+            if not isinstance(element_ids, Sequence) or isinstance(element_ids, (str, bytes)):
+                raise PhaseTwoCandidateError(
+                    "LEGACY_WRITE_INVALID", "element_ids must be an integer array"
+                )
+            normalized.append(
+                (
+                    _positive_int(item.get("paragraph_id"), "paragraph_id"),
+                    tuple(_positive_int(value, "element_id") for value in element_ids),
+                )
+            )
+        count = await self._workspace_repo().batch_link(scope.write_context, normalized)
+        return {"linked": count}
+
+    async def delete_paragraph_element_links(
+        self, *, links: Sequence[Mapping[str, Any]], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        scope = await self._workspace_scope(context, element=True)
+        pairs: list[tuple[int, int]] = []
+        for item in links:
+            _reject_identity(item)
+            pairs.append(
+                (
+                    _positive_int(item.get("paragraph_id"), "paragraph_id"),
+                    _positive_int(item.get("element_id"), "element_id"),
+                )
+            )
+        count = await self._workspace_repo().delete_links(scope.write_context, pairs=tuple(pairs))
+        return {"deleted": count}
+
+    async def save_comparison_candidate(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        _reject_identity(payload)
+        scope, bundle = await self._resolve_bundle(context)
+        if scope.contract.task_kind is not ScriptTaskKind.COMPARE:
+            raise PhaseTwoCandidateError(
+                "TASK_KIND_PRESET_MISMATCH", "comparison output requires a Compare contract"
+            )
+        expected = tuple(item.artifact_ref.uri for item in scope.contract.comparison_decision_refs)
+        if "candidate_artifact_refs" in payload:
+            raise PhaseTwoCandidateError(
+                "ACCEPTED_INPUT_CONFLICT",
+                "comparison candidate identities are derived from the frozen contract",
+            )
+        if len(expected) < 2:
+            raise PhaseTwoCandidateError(
+                "ACCEPTED_INPUT_CONFLICT", "comparison requires two frozen DecisionRefs"
+            )
+        await self._require_bundle_refs(bundle, scope.contract.comparison_decision_refs)
+        artifact = ComparisonArtifactV1(
+            lineage=await self._lineage(scope, bundle),
+            candidate_artifact_refs=expected,
+            criterion_results=tuple(
+                dict(item)
+                for item in _mapping_sequence(payload.get("criterion_results"), required=True)
+            ),
+            conflicts=_text_sequence(payload.get("conflicts")),
+            recommendation=_required_text(payload, "recommendation"),
+        )
+        return await self._freeze_direct(scope, artifact)
+
+    async def save_structured_script_candidate(
+        self, *, acceptance_notes: Sequence[str], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        scope, bundle = await self._resolve_bundle(context)
+        if scope.contract.task_kind is not ScriptTaskKind.COMPOSE:
+            raise PhaseTwoCandidateError(
+                "TASK_KIND_PRESET_MISMATCH", "StructuredScript output requires Compose"
+            )
+        scope.contract.require_execution_ready()
+        frontier, versions = await self._frontier(scope, bundle)
+        if tuple(item.decision_id for item in frontier) != scope.contract.compose_order:
+            raise PhaseTwoCandidateError(
+                "ACCEPTED_INPUT_CONFLICT",
+                "Compose order must enumerate every adopted active Decision exactly once",
+            )
+        workspace = await self._workspace_repo().get_or_create(
+            scope.write_context, artifact_kind=ArtifactKind.STRUCTURED_SCRIPT
+        )
+        if workspace.branch_id <= 0:
+            raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "Compose cannot write branch 0")
+        snapshot = await self._workspace_repo().snapshot(scope.write_context)
+        if snapshot.paragraphs or snapshot.elements or snapshot.links:
+            raise PhaseTwoCandidateError(
+                "MISSION_RECOVERY_REQUIRED",
+                "partially materialized Compose workspace requires explicit recovery",
+            )
+        support_versions = tuple(
+            [await self._read_accepted_version(scope, item) for item in bundle.inputs]
+        )
+        direction_ref, evidence_refs = _compose_support_refs(support_versions)
+        await self._materialize(scope, versions, support_versions)
+        version, ref = await self._workspace_repo().freeze(
+            scope.write_context,
+            lineage=await self._lineage(scope, bundle),
+            structured_script={
+                "direction_ref": direction_ref,
+                "source_artifact_refs": [item.artifact_ref.uri for item in frontier],
+                "evidence_refs": evidence_refs,
+                "acceptance_notes": tuple(acceptance_notes),
+            },
+        )
+        return _frozen_result(version, ref)
+
+    async def save_candidate_portfolio(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        _reject_identity(payload)
+        derived_fields = {
+            "adopted_structured_script_ref",
+            "candidate_structured_script_refs",
+            "accepted_decision_ids",
+            "rejected_or_held_decision_ids",
+            "input_closure_digest",
+            "compose_order",
+        }
+        if derived_fields & set(payload):
+            raise PhaseTwoCandidateError(
+                "ACCEPTED_INPUT_CONFLICT",
+                "Portfolio identities and closure are derived from the frozen contract",
+            )
+        scope, bundle = await self._resolve_bundle(context)
+        if scope.contract.task_kind is not ScriptTaskKind.CANDIDATE_PORTFOLIO:
+            raise PhaseTwoCandidateError(
+                "TASK_KIND_PRESET_MISMATCH", "Portfolio output requires a Portfolio contract"
+            )
+        scope.contract.require_execution_ready()
+        if len(scope.contract.adopted_decision_ids) != 1:
+            raise PhaseTwoCandidateError(
+                "PHASE_TWO_BOUNDARY_NOT_READY", "Portfolio must adopt exactly one StructuredScript"
+            )
+        by_decision = {item.decision_id: item for item in bundle.inputs}
+        closure_ids = tuple(
+            item.decision_id for item in scope.contract.candidate_closure_decision_refs
+        )
+        if set(closure_ids) != set(scope.contract.adopted_decision_ids) | set(
+            scope.contract.held_or_rejected_decision_ids
+        ):
+            raise PhaseTwoCandidateError(
+                "PHASE_TWO_BOUNDARY_NOT_READY", "Portfolio dispositions do not close candidates"
+            )
+        candidate_refs: list[str] = []
+        candidate_versions: list[ArtifactVersion] = []
+        for decision_id in closure_ids:
+            item = by_decision.get(decision_id)
+            if item is None or item.task_kind is not ScriptTaskKind.COMPOSE:
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID", "Portfolio candidate is not an accepted Compose"
+                )
+            version = await self._read_accepted_version(scope, item)
+            if not isinstance(version.artifact, StructuredScriptArtifactV1):
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID", "Portfolio candidate is not a StructuredScript"
+                )
+            candidate_refs.append(item.artifact_ref.uri)
+            candidate_versions.append(version)
+        adopted = by_decision[scope.contract.adopted_decision_ids[0]].artifact_ref.uri
+        superseded = _text_sequence(payload.get("superseded_decision_ids"))
+        if not set(superseded) <= set(scope.contract.held_or_rejected_decision_ids):
+            raise PhaseTwoCandidateError(
+                "ACCEPTED_INPUT_CONFLICT", "Portfolio supersession is outside held candidates"
+            )
+        unresolved = tuple(
+            ScriptDefectV1.from_payload(item)
+            for item in _mapping_sequence(payload.get("unresolved_defects"))
+        )
+        allowed_evidence_refs = {
+            item.artifact_ref.uri
+            for item in bundle.inputs
+            if item.task_kind
+            in {
+                ScriptTaskKind.PATTERN_RETRIEVAL,
+                ScriptTaskKind.DECODE_RETRIEVAL,
+                ScriptTaskKind.EXTERNAL_RETRIEVAL,
+                ScriptTaskKind.KNOWLEDGE_RETRIEVAL,
+            }
+        }
+        for version in candidate_versions:
+            artifact = version.artifact
+            if not isinstance(artifact, StructuredScriptArtifactV1):
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID", "Portfolio candidate type changed during validation"
+                )
+            allowed_evidence_refs.update(artifact.evidence_refs)
+        allowed_invalidated_inputs = {item.decision_id for item in bundle.inputs}
+        allowed_criteria = {item.criterion_id for item in scope.contract.criteria}
+        allowed_scopes = {scope.contract.scope_ref}
+        for defect in unresolved:
+            if not set(defect.evidence_refs) <= allowed_evidence_refs:
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID",
+                    "Portfolio defect evidence is outside the accepted closure",
+                )
+            if not set(defect.invalidated_inputs) <= allowed_invalidated_inputs:
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID",
+                    "Portfolio defect invalidates an input outside the accepted closure",
+                )
+            if defect.criterion_id not in allowed_criteria or not any(
+                _scope_compatible(defect.scope_ref, value) for value in allowed_scopes
+            ):
+                raise PhaseTwoCandidateError(
+                    "INPUT_SCOPE_MISMATCH",
+                    "Portfolio defect criterion or scope is outside the accepted closure",
+                )
+        artifact = CandidatePortfolioArtifactV1(
+            adopted_structured_script_ref=adopted,
+            candidate_structured_script_refs=tuple(candidate_refs),
+            accepted_decision_ids=scope.contract.adopted_decision_ids,
+            superseded_decision_ids=superseded,
+            rejected_or_held_decision_ids=tuple(
+                item
+                for item in scope.contract.held_or_rejected_decision_ids
+                if item not in superseded
+            ),
+            input_closure_digest=bundle.input_closure_digest,
+            unresolved_defects=unresolved,
+            compose_order=scope.contract.compose_order,
+        )
+        return await self._freeze_direct(scope, artifact)
+
+    async def resolve_attempt_manifest(self, *, context: Mapping[str, Any]) -> AttemptManifest:
+        scope = await self._scope(context)
+        expected = _artifact_kind(scope.contract.task_kind)
+        if scope.contract.task_kind in {
+            ScriptTaskKind.STRUCTURE,
+            ScriptTaskKind.PARAGRAPH,
+            ScriptTaskKind.ELEMENT_SET,
+        }:
+            _, bundle = await self._resolve_bundle(context)
+            version, _ = await self._workspace_repo().freeze(
+                scope.write_context, lineage=await self._lineage(scope, bundle)
+            )
+        else:
+            version = await self._artifacts.get_by_attempt(
+                script_build_id=scope.script_build_id,
+                task_id=scope.task.task_id,
+                attempt_id=scope.attempt.attempt_id,
+            )
+        if version.state is not ArtifactState.FROZEN or version.artifact_type is not expected:
+            raise PhaseTwoCandidateError(
+                "CHILD_DECISION_INVALID", "Attempt does not own its declared frozen Artifact"
+            )
+        ref = _artifact_ref(version)
+        return AttemptManifest(artifact_ref=ref, scope_ref=scope.contract.scope_ref)
+
+    async def discard_attempt(
+        self,
+        *,
+        root_trace_id: str,
+        task_id: str,
+        attempt_id: str,
+        reason: str,
+    ) -> None:
+        """Discard only an existing mutable creative workspace after a durable decision."""
+
+        ledger = await self._task_store.load(root_trace_id)
+        task = ledger.tasks.get(task_id)
+        attempt = ledger.attempts.get(attempt_id)
+        if task is None or attempt is None or attempt.task_id != task_id:
+            raise PhaseTwoCandidateError(
+                "CHILD_DECISION_INVALID", "discarded Attempt is outside this Root and Task"
+            )
+        contract = await self._accepted_inputs.contract_for_task(
+            root_trace_id=root_trace_id,
+            task=task,
+            spec_version=attempt.spec_version,
+        )
+        if contract.task_kind not in {
+            ScriptTaskKind.STRUCTURE,
+            ScriptTaskKind.PARAGRAPH,
+            ScriptTaskKind.ELEMENT_SET,
+        }:
+            return
+        binding = await self._bindings.get_by_root(root_trace_id)
+        context = CandidateWriteContext(
+            script_build_id=binding.script_build_id,
+            task_id=task_id,
+            attempt_id=attempt_id,
+            spec_version=attempt.spec_version,
+            objective=contract.objective,
+            input_refs=tuple(
+                item.artifact_ref.uri
+                for item in (
+                    *contract.input_decision_refs,
+                    *contract.candidate_closure_decision_refs,
+                    *contract.comparison_decision_refs,
+                )
+            ),
+            write_scope=contract.write_scope,
+        )
+        try:
+            workspace = await self._workspace_repo().require(context)
+        except Exception as exc:
+            if getattr(exc, "code", None) == "ATTEMPT_WORKSPACE_NOT_FOUND":
+                return
+            raise
+        if workspace.state in {ArtifactState.FROZEN, ArtifactState.DISCARDED}:
+            return
+        await self._workspace_repo().discard(context, reason=reason)
+
+    async def _scope(self, context: Mapping[str, Any]) -> CandidateScope:
+        root_trace_id = _required_context(context, "root_trace_id")
+        binding = await self._bindings.get_by_root(root_trace_id)
+        if binding.root_trace_id != root_trace_id:
+            raise PhaseTwoCandidateError("INPUT_SCOPE_MISMATCH", "binding root identity changed")
+        ledger = await self._task_store.load(root_trace_id)
+        task_id = _required_context(context, "task_id")
+        attempt_id = _required_context(context, "attempt_id")
+        task = ledger.tasks.get(task_id)
+        attempt = ledger.attempts.get(attempt_id)
+        if (
+            task is None
+            or attempt is None
+            or attempt.task_id != task_id
+            or attempt_id not in task.attempt_ids
+        ):
+            raise PhaseTwoCandidateError(
+                "CHILD_DECISION_INVALID", "protected Attempt is outside this Root and Task"
+            )
+        protected_spec = context.get("spec_version")
+        if protected_spec is not None and (
+            isinstance(protected_spec, bool)
+            or not isinstance(protected_spec, int)
+            or protected_spec != attempt.spec_version
+        ):
+            raise PhaseTwoCandidateError(
+                "CHILD_DECISION_INVALID", "protected spec version differs from the Attempt"
+            )
+        contract = await self._accepted_inputs.contract_for_task(
+            root_trace_id=root_trace_id,
+            task=task,
+            spec_version=attempt.spec_version,
+        )
+        write_context = CandidateWriteContext(
+            script_build_id=binding.script_build_id,
+            task_id=task_id,
+            attempt_id=attempt_id,
+            spec_version=attempt.spec_version,
+            objective=contract.objective,
+            input_refs=tuple(
+                item.artifact_ref.uri
+                for item in (
+                    *contract.input_decision_refs,
+                    *contract.candidate_closure_decision_refs,
+                    *contract.comparison_decision_refs,
+                )
+            ),
+            write_scope=contract.write_scope,
+        )
+        return CandidateScope(
+            root_trace_id=root_trace_id,
+            script_build_id=binding.script_build_id,
+            input_snapshot_id=str(binding.input_snapshot_id),
+            task=task,
+            attempt=attempt,
+            contract=contract,
+            write_context=write_context,
+        )
+
+    async def _resolve_bundle(
+        self, context: Mapping[str, Any]
+    ) -> tuple[CandidateScope, AcceptedInputBundleV1]:
+        scope = await self._scope(context)
+        bundle = await self._accepted_inputs.resolve(
+            root_trace_id=scope.root_trace_id,
+            script_build_id=scope.script_build_id,
+            task_id=scope.task.task_id,
+            attempt_id=scope.attempt.attempt_id,
+            contract=scope.contract,
+            input_snapshot_id=scope.input_snapshot_id,
+        )
+        return scope, bundle
+
+    async def _workspace_scope(
+        self,
+        context: Mapping[str, Any],
+        *,
+        paragraph: bool = False,
+        element: bool = False,
+    ) -> CandidateScope:
+        scope = await self._scope(context)
+        allowed = (
+            {ScriptTaskKind.STRUCTURE, ScriptTaskKind.PARAGRAPH, ScriptTaskKind.COMPOSE}
+            if paragraph
+            else {ScriptTaskKind.ELEMENT_SET, ScriptTaskKind.COMPOSE}
+        )
+        if element:
+            allowed = {ScriptTaskKind.ELEMENT_SET, ScriptTaskKind.COMPOSE}
+        if scope.contract.task_kind not in allowed:
+            raise PhaseTwoCandidateError(
+                "TASK_KIND_PRESET_MISMATCH", "workspace operation is outside this Task kind"
+            )
+        if not scope.contract.write_scope:
+            raise PhaseTwoCandidateError(
+                "WRITE_SCOPE_VIOLATION", "creative Task has no writable scope"
+            )
+        base = scope.contract.base_artifact_ref
+        if base is not None:
+            _, bundle = await self._resolve_bundle(context)
+            if base.uri not in {item.artifact_ref.uri for item in bundle.inputs}:
+                raise PhaseTwoCandidateError(
+                    "STALE_BASE_REVISION", "base Artifact is not an accepted immutable input"
+                )
+        workspace = await self._workspace_repo().get_or_create(
+            scope.write_context,
+            artifact_kind=_WORKSPACE_KIND[scope.contract.task_kind],
+            base_artifact_ref=base,
+        )
+        if workspace.branch_id <= 0:
+            raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "workspace cannot use branch 0")
+        return scope
+
+    def _workspace_repo(self) -> CandidateWorkspaceOperations:
+        if self._workspaces is None:
+            raise PhaseTwoCandidateError(
+                "ATTEMPT_WORKSPACE_NOT_FOUND", "candidate workspace storage is not configured"
+            )
+        return cast(CandidateWorkspaceOperations, self._workspaces)
+
+    async def _frontier(
+        self, scope: CandidateScope, bundle: AcceptedInputBundleV1
+    ) -> tuple[tuple[AcceptedInput, ...], tuple[ArtifactVersion, ...]]:
+        ledger = await self._task_store.load(scope.root_trace_id)
+        contracts: dict[str, ScriptTaskContractV1] = {}
+        for item in bundle.inputs:
+            decision = ledger.decisions.get(item.decision_id)
+            task = ledger.tasks.get(decision.task_id) if decision is not None else None
+            if decision is None or task is None or decision.attempt_id is None:
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID", "frontier Decision is outside the Ledger"
+                )
+            attempt = ledger.attempts.get(decision.attempt_id)
+            if attempt is None:
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID", "frontier Decision has no immutable Attempt"
+                )
+            contracts[item.decision_id] = await self._accepted_inputs.contract_for_task(
+                root_trace_id=scope.root_trace_id,
+                task=task,
+                spec_version=attempt.spec_version,
+            )
+        await self._verify_comparison_disposition(scope, bundle, contracts)
+        frontier = self._active_frontier.resolve(
+            bundle,
+            contracts_by_decision=contracts,
+            adopted_decision_ids=scope.contract.adopted_decision_ids,
+        )
+        ordered = {item.decision_id: item for item in frontier}
+        if set(ordered) != set(scope.contract.compose_order):
+            raise PhaseTwoCandidateError(
+                "ACCEPTED_INPUT_CONFLICT", "Compose order differs from its active frontier"
+            )
+        frontier = tuple(ordered[item] for item in scope.contract.compose_order)
+        versions = tuple([await self._read_accepted_version(scope, item) for item in frontier])
+        return frontier, versions
+
+    async def _verify_comparison_disposition(
+        self,
+        scope: CandidateScope,
+        bundle: AcceptedInputBundleV1,
+        contracts: Mapping[str, ScriptTaskContractV1],
+    ) -> None:
+        if not scope.contract.comparison_decision_refs:
+            return
+        by_decision = {item.decision_id: item for item in bundle.inputs}
+        candidate_by_uri = {
+            item.artifact_ref.uri: item.decision_id
+            for item in scope.contract.candidate_closure_decision_refs
+        }
+        adopted = set(scope.contract.adopted_decision_ids)
+        disposed = set(scope.contract.held_or_rejected_decision_ids) | set(
+            scope.contract.supersedes_decision_ids
+        )
+        for pinned in scope.contract.comparison_decision_refs:
+            accepted = by_decision.get(pinned.decision_id)
+            contract = contracts.get(pinned.decision_id)
+            if (
+                accepted is None
+                or contract is None
+                or contract.task_kind is not ScriptTaskKind.COMPARE
+            ):
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID", "Compose comparison is outside the accepted closure"
+                )
+            version = await self._read_accepted_version(scope, accepted)
+            artifact = version.artifact
+            if not isinstance(artifact, ComparisonArtifactV1):
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID", "comparison Decision does not own a Comparison"
+                )
+            compared = {candidate_by_uri.get(uri) for uri in artifact.candidate_artifact_refs}
+            if None in compared or len(compared) != len(artifact.candidate_artifact_refs):
+                raise PhaseTwoCandidateError(
+                    "ACCEPTED_INPUT_CONFLICT",
+                    "Comparison candidates are not all pinned in the Compose closure",
+                )
+            recommendation = candidate_by_uri.get(artifact.recommendation)
+            if recommendation is None or recommendation not in adopted:
+                raise PhaseTwoCandidateError(
+                    "ACCEPTED_INPUT_CONFLICT",
+                    "Comparison recommendation is not formally adopted by Compose",
+                )
+            rejected = {item for item in compared if item != recommendation}
+            if not rejected <= disposed:
+                raise PhaseTwoCandidateError(
+                    "ACCEPTED_INPUT_CONFLICT",
+                    "non-recommended Comparison candidates require an explicit disposition",
+                )
+            fairness = comparison_fairness_errors(
+                artifact, {item.criterion_id for item in contract.criteria}
+            )
+            if fairness:
+                raise PhaseTwoCandidateError(
+                    "ACCEPTED_INPUT_CONFLICT",
+                    "accepted Comparison is not fair: " + ", ".join(fairness[:8]),
+                )
+
+    async def _read_accepted_version(
+        self, scope: CandidateScope, item: AcceptedInput
+    ) -> ArtifactVersion:
+        return await self._artifacts.read_by_ref(
+            item.artifact_ref, script_build_id=scope.script_build_id
+        )
+
+    async def _lineage(
+        self, scope: CandidateScope, bundle: AcceptedInputBundleV1
+    ) -> CandidateLineageV1:
+        base = scope.contract.base_artifact_ref
+        if base is not None and base.uri not in {item.artifact_ref.uri for item in bundle.inputs}:
+            raise PhaseTwoCandidateError(
+                "STALE_BASE_REVISION", "base Artifact is not in the accepted input closure"
+            )
+        base_version = (
+            await self._artifacts.read_by_ref(base, script_build_id=scope.script_build_id)
+            if base is not None
+            else None
+        )
+        lineage = CandidateLineageV1(
+            scope_ref=scope.contract.scope_ref,
+            input_snapshot_ref=f"script-build://inputs/{scope.input_snapshot_id}",
+            input_closure_digest=bundle.input_closure_digest,
+            base_artifact_ref=base.uri if base is not None else None,
+            base_artifact_digest=base.digest if base is not None else None,
+            base_revision=base_version.artifact_version_id if base_version is not None else None,
+            write_scope=scope.contract.write_scope,
+            supersedes_decision_ids=scope.contract.supersedes_decision_ids,
+            source_lineage=tuple(
+                {
+                    "decision_id": item.decision_id,
+                    "artifact_ref": item.artifact_ref.uri,
+                    "digest": item.artifact_ref.digest,
+                    "source": item.source,
+                }
+                for item in bundle.inputs
+            ),
+        )
+        self._active_frontier.verify_base(lineage=lineage, base=base_version)
+        return lineage
+
+    async def _freeze_direct(self, scope: CandidateScope, artifact: Any) -> Mapping[str, Any]:
+        version, ref = await self._artifacts.freeze(
+            script_build_id=scope.script_build_id,
+            task_id=scope.task.task_id,
+            attempt_id=scope.attempt.attempt_id,
+            spec_version=scope.attempt.spec_version,
+            artifact=artifact,
+        )
+        return _frozen_result(version, ref)
+
+    async def _require_bundle_refs(
+        self, bundle: AcceptedInputBundleV1, refs: Sequence[Any]
+    ) -> None:
+        actual = {item.decision_id: item.artifact_ref.uri for item in bundle.inputs}
+        if any(actual.get(item.decision_id) != item.artifact_ref.uri for item in refs):
+            raise PhaseTwoCandidateError(
+                "CHILD_DECISION_INVALID", "pinned DecisionRef is outside the accepted closure"
+            )
+
+    async def _materialize(
+        self,
+        scope: CandidateScope,
+        versions: Sequence[ArtifactVersion],
+        support_versions: Sequence[ArtifactVersion],
+    ) -> None:
+        """Materialize the explicitly ordered logical projection once.
+
+        Local legacy IDs are scoped to an Artifact version.  A patch copied from
+        a base carries a source identity map, so resolving that chain is the only
+        safe way to decide whether two rows describe the same logical entity.
+        """
+
+        version_by_ref = {_artifact_ref(item).uri: item for item in (*support_versions, *versions)}
+        pending = list(version_by_ref.values())
+        while pending:
+            version = pending.pop()
+            lineage = getattr(version.artifact, "lineage", None)
+            base_ref = getattr(lineage, "base_artifact_ref", None)
+            if not base_ref or base_ref in version_by_ref:
+                continue
+            base_id = int(str(base_ref).rsplit("/", 1)[-1])
+            base = await self._artifacts.get_by_id(base_id, script_build_id=scope.script_build_id)
+            version_by_ref[base_ref] = base
+            pending.append(base)
+
+        entity_key_cache: dict[tuple[str, str, int], tuple[str, str, int]] = {}
+
+        def entity_key(
+            version: ArtifactVersion,
+            entity: str,
+            local_id: int,
+            trail: frozenset[tuple[str, str, int]] = frozenset(),
+        ) -> tuple[str, str, int]:
+            version_ref = _artifact_ref(version).uri
+            cache_key = (version_ref, entity, local_id)
+            cached = entity_key_cache.get(cache_key)
+            if cached is not None:
+                return cached
+            if cache_key in trail:
+                raise PhaseTwoCandidateError(
+                    "STALE_BASE_REVISION", "candidate source identity contains a cycle"
+                )
+            lineage = getattr(version.artifact, "lineage", None)
+            matches = [
+                item
+                for item in getattr(lineage, "source_lineage", ())
+                if item.get("entity") == entity and item.get("local_id") == local_id
+            ]
+            if len(matches) > 1:
+                raise PhaseTwoCandidateError(
+                    "LEGACY_REFERENCE_INVALID", "candidate source identity is ambiguous"
+                )
+            base_ref = getattr(lineage, "base_artifact_ref", None)
+            if matches and base_ref:
+                source_id = int(matches[0]["source_local_id"])
+                base = version_by_ref.get(base_ref)
+                resolved = (
+                    entity_key(base, entity, source_id, trail | {cache_key})
+                    if base is not None
+                    else (entity, str(base_ref), source_id)
+                )
+            else:
+                resolved = (entity, version_ref, local_id)
+            entity_key_cache[cache_key] = resolved
+            return resolved
+
+        paragraph_projection: dict[
+            tuple[str, str, int], tuple[ScriptParagraphV1, tuple[str, str, int] | None]
+        ] = {}
+        element_projection: dict[tuple[str, str, int], ScriptElementV1] = {}
+        structure_paragraph_keys: set[tuple[str, str, int]] = set()
+        paragraph_candidate_keys: set[tuple[str, str, int]] = set()
+        link_projection: set[tuple[tuple[str, str, int], tuple[str, str, int]]] = set()
+
+        for version in versions:
+            artifact = version.artifact
+            if not isinstance(artifact, _MATERIALIZABLE):
+                continue
+            paragraph_keys = {
+                item.paragraph_id: entity_key(version, "paragraph", item.paragraph_id)
+                for item in artifact.paragraphs
+            }
+            if isinstance(artifact, (StructureArtifactV1, ParagraphArtifactV1)):
+                for paragraph in artifact.paragraphs:
+                    key = paragraph_keys[paragraph.paragraph_id]
+                    if isinstance(artifact, StructureArtifactV1) and paragraph.is_active:
+                        structure_paragraph_keys.add(key)
+                    if isinstance(artifact, ParagraphArtifactV1) and paragraph.is_active:
+                        paragraph_candidate_keys.add(key)
+                    if not paragraph.is_active:
+                        paragraph_projection.pop(key, None)
+                        continue
+                    parent_key = (
+                        paragraph_keys.get(paragraph.parent_id)
+                        if paragraph.parent_id is not None
+                        else None
+                    )
+                    paragraph_projection[key] = (paragraph, parent_key)
+
+            element_keys = {
+                item.element_id: entity_key(version, "element", item.element_id)
+                for item in getattr(artifact, "elements", ())
+            }
+            for element in getattr(artifact, "elements", ()):
+                key = element_keys[element.element_id]
+                if element.is_active:
+                    element_projection[key] = element
+                else:
+                    element_projection.pop(key, None)
+
+            lineage = getattr(artifact, "lineage", None)
+            base_ref = getattr(lineage, "base_artifact_ref", None)
+            deleted_links = getattr(artifact, "change_manifest", {}).get("deleted_links", ())
+            base_version = version_by_ref.get(base_ref) if base_ref else None
+            if base_version is not None:
+                for deleted in deleted_links:
+                    deleted_paragraph_key = entity_key(
+                        base_version, "paragraph", int(deleted["paragraph_id"])
+                    )
+                    deleted_element_key = entity_key(
+                        base_version, "element", int(deleted["element_id"])
+                    )
+                    link_projection.discard((deleted_paragraph_key, deleted_element_key))
+            for link in getattr(artifact, "paragraph_element_links", ()):
+                linked_paragraph_key = paragraph_keys.get(link.paragraph_id)
+                linked_element_key = element_keys.get(link.element_id)
+                if linked_paragraph_key is None or linked_element_key is None:
+                    raise PhaseTwoCandidateError(
+                        "LEGACY_REFERENCE_INVALID",
+                        "composed Link is outside its frozen candidate projection",
+                    )
+                link_projection.add((linked_paragraph_key, linked_element_key))
+
+        uncovered = paragraph_candidate_keys - structure_paragraph_keys
+        if uncovered:
+            raise PhaseTwoCandidateError(
+                "INPUT_SCOPE_MISMATCH",
+                "adopted Paragraphs must be covered or absorbed by an adopted Structure",
+            )
+        if not paragraph_projection:
+            raise PhaseTwoCandidateError(
+                "LEGACY_REFERENCE_INVALID", "Compose has no adopted active Paragraph"
+            )
+        active_links = {
+            pair
+            for pair in link_projection
+            if pair[0] in paragraph_projection and pair[1] in element_projection
+        }
+        linked_elements = {item[1] for item in active_links}
+        if set(element_projection) - linked_elements:
+            raise PhaseTwoCandidateError(
+                "LEGACY_REFERENCE_INVALID",
+                "every adopted Element must be linked to an adopted Paragraph",
+            )
+
+        paragraph_ids: dict[tuple[str, str, int], int] = {}
+        next_index = 1
+        ordered_paragraphs = sorted(
+            paragraph_projection.items(),
+            key=lambda item: (item[1][0].level, item[1][0].paragraph_index, item[0]),
+        )
+        for key, (paragraph, _parent_key) in ordered_paragraphs:
+            if paragraph.level != 1:
+                continue
+            paragraph_ids[key] = await self._create_materialized_paragraph(
+                scope.write_context, paragraph, next_index, None
+            )
+            next_index += 1
+        for key, (paragraph, parent_key) in ordered_paragraphs:
+            if paragraph.level != 2:
+                continue
+            parent_id = paragraph_ids.get(cast(tuple[str, str, int], parent_key))
+            if parent_id is None:
+                raise PhaseTwoCandidateError(
+                    "LEGACY_REFERENCE_INVALID", "composed paragraph parent is not adopted"
+                )
+            paragraph_ids[key] = await self._create_materialized_paragraph(
+                scope.write_context, paragraph, next_index, parent_id
+            )
+            next_index += 1
+
+        element_ids: dict[tuple[str, str, int], int] = {}
+        for key, element in sorted(element_projection.items(), key=lambda item: item[0]):
+            element_ids[key] = await self._workspace_repo().create_element(
+                scope.write_context,
+                name=element.name,
+                dimension_primary=element.dimension_primary,
+                dimension_secondary=element.dimension_secondary,
+                commonality_analysis=element.commonality_analysis,
+                topic_support=element.topic_support,
+                weight_score=element.weight_score,
+                support_elements=element.support_elements,
+            )
+        links_by_paragraph: dict[int, list[int]] = {}
+        for paragraph_key, element_key in sorted(active_links):
+            links_by_paragraph.setdefault(paragraph_ids[paragraph_key], []).append(
+                element_ids[element_key]
+            )
+        if links_by_paragraph:
+            await self._workspace_repo().batch_link(
+                scope.write_context,
+                tuple((key, tuple(value)) for key, value in links_by_paragraph.items()),
+            )
+
+    async def _create_materialized_paragraph(
+        self,
+        context: CandidateWriteContext,
+        paragraph: ScriptParagraphV1,
+        paragraph_index: int,
+        parent_id: int | None,
+    ) -> int:
+        identifier = await self._workspace_repo().create_paragraph(
+            context,
+            paragraph_index=paragraph_index,
+            name=paragraph.name,
+            content_range=paragraph.content_range,
+            level=paragraph.level,
+            parent_paragraph_id=parent_id,
+            theme_elements=paragraph.theme_elements,
+            form_elements=paragraph.form_elements,
+            function_elements=paragraph.function_elements,
+            feeling_elements=paragraph.feeling_elements,
+        )
+        values = {
+            key: value
+            for key, value in {
+                "theme": paragraph.theme,
+                "form": paragraph.form,
+                "function": paragraph.function,
+                "feeling": paragraph.feeling,
+                "description": paragraph.description,
+                "full_description": paragraph.full_description,
+            }.items()
+            if value is not None
+        }
+        if values:
+            await self._workspace_repo().batch_update_paragraphs(
+                context, ({"paragraph_id": identifier, **values},)
+            )
+        return identifier
+
+    async def _allowed_raw_refs(
+        self,
+        scope: CandidateScope,
+        bundle: AcceptedInputBundleV1,
+        context: Mapping[str, Any],
+    ) -> set[str]:
+        allowed: set[str] = set()
+        for item in bundle.inputs:
+            version = await self._read_accepted_version(scope, item)
+            if isinstance(version.artifact, EvidenceRecordV1) and version.artifact.raw_artifact_ref:
+                allowed.add(version.artifact.raw_artifact_ref)
+        snapshot_id = context.get("snapshot_id")
+        if isinstance(snapshot_id, str) and snapshot_id:
+            snapshot = await self._framework_artifact_store.get(scope.root_trace_id, snapshot_id)
+            if snapshot.attempt_id != scope.attempt.attempt_id:
+                raise PhaseTwoCandidateError(
+                    "CHILD_DECISION_INVALID", "Validator snapshot is outside its Attempt"
+                )
+            for ref in (*snapshot.artifact_refs, *snapshot.evidence_refs):
+                version = await self._artifacts.read_by_ref(
+                    ref,
+                    script_build_id=scope.script_build_id,
+                    task_id=scope.task.task_id,
+                    attempt_id=scope.attempt.attempt_id,
+                )
+                if (
+                    isinstance(version.artifact, EvidenceRecordV1)
+                    and version.artifact.raw_artifact_ref
+                ):
+                    allowed.add(version.artifact.raw_artifact_ref)
+        return allowed
+
+
+def _artifact_kind(kind: ScriptTaskKind) -> ArtifactKind:
+    mapping = {
+        ScriptTaskKind.PATTERN_RETRIEVAL: ArtifactKind.EVIDENCE,
+        ScriptTaskKind.DECODE_RETRIEVAL: ArtifactKind.EVIDENCE,
+        ScriptTaskKind.EXTERNAL_RETRIEVAL: ArtifactKind.EVIDENCE,
+        ScriptTaskKind.KNOWLEDGE_RETRIEVAL: ArtifactKind.EVIDENCE,
+        ScriptTaskKind.DIRECTION: ArtifactKind.DIRECTION,
+        ScriptTaskKind.STRUCTURE: ArtifactKind.STRUCTURE,
+        ScriptTaskKind.PARAGRAPH: ArtifactKind.PARAGRAPH,
+        ScriptTaskKind.ELEMENT_SET: ArtifactKind.ELEMENT_SET,
+        ScriptTaskKind.COMPARE: ArtifactKind.COMPARISON,
+        ScriptTaskKind.COMPOSE: ArtifactKind.STRUCTURED_SCRIPT,
+        ScriptTaskKind.CANDIDATE_PORTFOLIO: ArtifactKind.CANDIDATE_PORTFOLIO,
+    }
+    return mapping[kind]
+
+
+def _artifact_ref(version: ArtifactVersion) -> ArtifactRef:
+    return ArtifactRef(
+        uri=f"script-build://artifact-versions/{version.artifact_version_id}",
+        kind=version.artifact_type.value,
+        version=str(version.artifact_version_id),
+        digest=version.canonical_sha256,
+    )
+
+
+def _accepted_manifest(value: AcceptedInput) -> dict[str, Any]:
+    return {
+        "decision_id": value.decision_id,
+        "artifact_ref": _ref_payload(value.artifact_ref),
+        "task_kind": value.task_kind.value,
+        "scope_ref": value.scope_ref,
+        "source": value.source,
+    }
+
+
+def _workspace_payload(value: CandidateWorkspaceSnapshot) -> dict[str, Any]:
+    return {
+        "artifact_version_id": value.workspace.artifact_version_id,
+        "branch_id": value.workspace.branch_id,
+        "state": value.workspace.state.value,
+        "artifact_kind": value.workspace.artifact_kind.value,
+        "paragraphs": [item.to_payload() for item in value.paragraphs],
+        "elements": [item.to_payload() for item in value.elements],
+        "paragraph_element_links": [item.to_payload() for item in value.links],
+    }
+
+
+def _artifact_payload(version: ArtifactVersion) -> Mapping[str, Any]:
+    content_payload = getattr(version.artifact, "content_payload", None)
+    if callable(content_payload):
+        return cast(Mapping[str, Any], content_payload())
+    return cast(Mapping[str, Any], asdict(version.artifact))
+
+
+def _frozen_result(version: ArtifactVersion, ref: ArtifactRef) -> Mapping[str, Any]:
+    return {
+        "artifact_version_id": version.artifact_version_id,
+        "artifact_ref": _ref_payload(ref),
+        "artifact": _artifact_payload(version),
+    }
+
+
+def _compose_support_refs(versions: Sequence[ArtifactVersion]) -> tuple[str, tuple[str, ...]]:
+    directions = [
+        f"script-build://artifact-versions/{item.artifact_version_id}"
+        for item in versions
+        if item.artifact_type is ArtifactKind.DIRECTION
+    ]
+    evidence = tuple(
+        f"script-build://artifact-versions/{item.artifact_version_id}"
+        for item in versions
+        if item.artifact_type is ArtifactKind.EVIDENCE
+    )
+    if len(directions) != 1:
+        raise PhaseTwoCandidateError(
+            "CHILD_DECISION_INVALID", "Compose must pin exactly one accepted Direction"
+        )
+    return directions[0], evidence
+
+
+def _image_media_type(content: bytes) -> str:
+    if content.startswith(b"\x89PNG\r\n\x1a\n"):
+        return "image/png"
+    if content.startswith(b"\xff\xd8\xff"):
+        return "image/jpeg"
+    if content.startswith((b"GIF87a", b"GIF89a")):
+        return "image/gif"
+    if len(content) >= 12 and content[:4] == b"RIFF" and content[8:12] == b"WEBP":
+        return "image/webp"
+    raise PhaseTwoCandidateError("LEGACY_REFERENCE_INVALID", "raw Artifact is not a safe image")
+
+
+def _reject_identity(payload: Mapping[str, Any]) -> None:
+    if _IDENTITY_FIELDS & set(payload):
+        raise PhaseTwoCandidateError(
+            "WRITE_SCOPE_VIOLATION", "candidate payload cannot supply protected identities"
+        )
+
+
+def _required_context(context: Mapping[str, Any], key: str) -> str:
+    value = context.get(key)
+    if not isinstance(value, str) or not value.strip():
+        raise ProtocolViolation(f"protected context is missing {key}")
+    return value
+
+
+def _required_text(payload: Mapping[str, Any], key: str) -> str:
+    value = payload.get(key)
+    if not isinstance(value, str) or not value.strip():
+        raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", f"{key} must not be blank")
+    return value.strip()
+
+
+def _positive_int(value: Any, label: str) -> int:
+    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
+        raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", f"{label} must be positive")
+    return value
+
+
+def _optional_int(value: Any, *, default: int) -> int:
+    return default if value is None else _positive_int(value, "integer field")
+
+
+def _nullable_positive_int(value: Any) -> int | None:
+    return None if value is None else _positive_int(value, "parent_paragraph_id")
+
+
+def _mapping(value: Any, label: str) -> Mapping[str, Any]:
+    if not isinstance(value, Mapping):
+        raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", f"{label} must be an object")
+    return value
+
+
+def _optional_mapping(value: Any) -> Mapping[str, Any] | None:
+    return None if value is None else _mapping(value, "object field")
+
+
+def _mapping_sequence(value: Any, *, required: bool = False) -> tuple[Mapping[str, Any], ...]:
+    if value is None and not required:
+        return ()
+    if (
+        not isinstance(value, Sequence)
+        or isinstance(value, (str, bytes))
+        or any(not isinstance(item, Mapping) for item in value)
+        or (required and not value)
+    ):
+        raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "field must be an object array")
+    return tuple(cast(Mapping[str, Any], item) for item in value)
+
+
+def _text_sequence(value: Any, *, required: bool = False) -> tuple[str, ...]:
+    if value is None and not required:
+        return ()
+    if (
+        not isinstance(value, Sequence)
+        or isinstance(value, (str, bytes))
+        or any(not isinstance(item, str) or not item.strip() for item in value)
+        or (required and not value)
+    ):
+        raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "field must be a string array")
+    result = tuple(str(item) for item in value)
+    if len(set(result)) != len(result):
+        raise PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "string array must be unique")
+    return result
+
+
+def _ref_payload(ref: ArtifactRef) -> dict[str, Any]:
+    return {
+        "uri": ref.uri,
+        "kind": ref.kind,
+        "version": ref.version,
+        "digest": ref.digest,
+    }
+
+
+def _scope_compatible(left: str, right: str) -> bool:
+    return (
+        left == right
+        or left.startswith(f"{right.rstrip('/')}/")
+        or right.startswith(f"{left.rstrip('/')}/")
+    )
+
+
+__all__ = ["PhaseTwoCandidateError", "PhaseTwoCandidateService"]

script_build_host/src/script_build_host/tools/contracts.py

@@ -0,0 +1,127 @@
+"""Application ports and bounded values used by Agent-visible Host tools.
+
+The tool registry is an adapter: it must not know how a Task contract is
+stored or how a candidate workspace is persisted.  These protocols keep those
+decisions behind application ports while still giving the Agent a precise,
+testable surface.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from typing import Any, Protocol
+
+from script_build_host.domain.phase_two_ports import AttemptManifest
+
+
+class ScriptPlannerToolPort(Protocol):
+    """Phase-aware planner operations over the generic Coordinator."""
+
+    async def plan_script_tasks(
+        self,
+        *,
+        contract_payloads: Sequence[Mapping[str, Any]],
+        parent_task_id: str | None,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]: ...
+
+    async def decide_script_task(
+        self,
+        *,
+        task_id: str,
+        action: str,
+        reason: str,
+        validation_id: str | None,
+        replacement_contract: Mapping[str, Any] | None,
+        child_contracts: Sequence[Mapping[str, Any]],
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]: ...
+
+    async def inspect_script_plan(
+        self,
+        *,
+        context: Mapping[str, Any],
+    ) -> Mapping[str, Any]: ...
+
+    async def dispatch_script_tasks(
+        self,
+        *,
+        task_ids: Sequence[str],
+        context: Mapping[str, Any],
+    ) -> Sequence[Mapping[str, Any]]: ...
+
+
+class ScriptCandidateToolPort(Protocol):
+    """Ownership-aware candidate operations resolved from protected context."""
+
+    async def read_accepted_input_bundle(
+        self, *, context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def read_active_frontier(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]: ...
+
+    async def read_pinned_candidates(
+        self, *, context: Mapping[str, Any]
+    ) -> Sequence[Mapping[str, Any]]: ...
+
+    async def view_frozen_images(
+        self,
+        *,
+        raw_artifact_refs: Sequence[str],
+        context: Mapping[str, Any],
+    ) -> Sequence[Mapping[str, Any]]: ...
+
+    async def deterministic_precheck(
+        self, *, context: Mapping[str, Any]
+    ) -> Sequence[Mapping[str, Any]]: ...
+
+    async def read_attempt_workspace(self, *, context: Mapping[str, Any]) -> Mapping[str, Any]: ...
+
+    async def create_script_paragraph(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def append_paragraph_atoms(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def delete_paragraph_atom(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def batch_update_script_paragraphs(
+        self, *, updates: Sequence[Mapping[str, Any]], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def create_script_element(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def update_script_element(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def batch_link_paragraph_elements(
+        self, *, links: Sequence[Mapping[str, Any]], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def delete_paragraph_element_links(
+        self, *, links: Sequence[Mapping[str, Any]], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def save_comparison_candidate(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def save_structured_script_candidate(
+        self, *, acceptance_notes: Sequence[str], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def save_candidate_portfolio(
+        self, *, payload: Mapping[str, Any], context: Mapping[str, Any]
+    ) -> Mapping[str, Any]: ...
+
+    async def resolve_attempt_manifest(self, *, context: Mapping[str, Any]) -> AttemptManifest: ...
+
+
+__all__ = ["AttemptManifest", "ScriptCandidateToolPort", "ScriptPlannerToolPort"]