feat(script-build): add phase three contracts and migration

script_build_host/migrations/CHECKSUMS.sha256

@@ -1,2 +1,3 @@
 0d524c7647d567ec3fba3c6d747ec09c2c96c9805506857adf14996a2539bff7  migrations/versions/0001_phase_one_business_tables.py
 c570caabd328ca4ff0c7aaa6212f44af3bae916bcdb5f2b028e40a1f16da2ff7  migrations/versions/0002_phase_two_artifact_types.py
+1e4f94fd826ca8dfe8b0b0c1dd33470f052d371221d9d47c7d4f5638cc741f80  migrations/versions/0003_phase_three_publication_fencing.py

script_build_host/migrations/versions/0003_phase_three_publication_fencing.py

@@ -0,0 +1,228 @@
+"""Add phase-three delivery, ownership fencing and HTTP command journal.
+
+Revision ID: 0003_phase_three_publication
+Revises: 0002_phase_two_artifacts
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import context, op
+from sqlalchemy.dialects import mysql
+
+revision: str = "0003_phase_three_publication"
+down_revision: str | None = "0002_phase_two_artifacts"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_PHASE_TWO_CHECK = (
+    "artifact_type IN ("
+    "'evidence', 'direction', 'structure', 'paragraph', 'element_set', "
+    "'comparison', 'structured_script', 'candidate_portfolio'"
+    ")"
+)
+_PHASE_THREE_CHECK = (
+    "artifact_type IN ("
+    "'evidence', 'direction', 'structure', 'paragraph', 'element_set', "
+    "'comparison', 'structured_script', 'candidate_portfolio', "
+    "'root_delivery_manifest'"
+    ")"
+)
+
+
+def _online() -> bool:
+    return not context.is_offline_mode()
+
+
+def _has_table(name: str) -> bool:
+    return _online() and sa.inspect(op.get_bind()).has_table(name)
+
+
+def _preflight() -> None:
+    if not _online():
+        return
+    bind = op.get_bind()
+    duplicate = bind.execute(
+        sa.text(
+            "SELECT script_build_id, publication_type, COUNT(*) AS n "
+            "FROM script_build_publication GROUP BY script_build_id, publication_type "
+            "HAVING COUNT(*) > 1 LIMIT 1"
+        )
+    ).first()
+    if duplicate is not None:
+        raise RuntimeError("phase-three preflight found duplicate build publication identities")
+    invalid_final = bind.execute(
+        sa.text(
+            "SELECT COUNT(*) FROM script_build_publication p "
+            "LEFT JOIN script_build_artifact_version a ON a.id=p.artifact_version_id "
+            "WHERE p.publication_type='final' AND "
+            "(a.id IS NULL OR a.artifact_type<>'root_delivery_manifest')"
+        )
+    ).scalar_one()
+    if int(invalid_final) > 0:
+        raise RuntimeError("phase-three preflight found an invalid final publication")
+    if _has_table("script_build_record"):
+        missing_trace = bind.execute(
+            sa.text(
+                "SELECT COUNT(*) FROM script_build_record "
+                "WHERE reson_trace_id IS NULL OR reson_trace_id=''"
+            )
+        ).scalar_one()
+        duplicate_trace = bind.execute(
+            sa.text(
+                "SELECT reson_trace_id FROM script_build_record GROUP BY reson_trace_id "
+                "HAVING COUNT(*) > 1 LIMIT 1"
+            )
+        ).first()
+        if int(missing_trace) > 0 or duplicate_trace is not None:
+            raise RuntimeError("script_build_record.reson_trace_id is not non-null and unique")
+
+
+def _create_mysql_views() -> None:
+    if context.get_context().dialect.name != "mysql":
+        return
+    for source, view in (
+        ("script_build_paragraph", "script_build_candidate_paragraph_v"),
+        ("script_build_element", "script_build_candidate_element_v"),
+        ("script_build_paragraph_element", "script_build_candidate_paragraph_element_v"),
+    ):
+        op.execute(
+            sa.text(
+                f"CREATE OR REPLACE ALGORITHM=MERGE VIEW {view} AS "
+                f"SELECT * FROM {source} WHERE branch_id > 0 WITH CASCADED CHECK OPTION"
+            )
+        )
+    op.execute(
+        sa.text(
+            "CREATE OR REPLACE ALGORITHM=MERGE VIEW script_build_runtime_record_v AS "
+            "SELECT * FROM script_build_record WHERE status <> 'success' "
+            "WITH CASCADED CHECK OPTION"
+        )
+    )
+
+
+def upgrade() -> None:
+    _preflight()
+    timestamp = sa.DateTime().with_variant(mysql.DATETIME(fsp=6), "mysql")
+    with op.batch_alter_table("script_build_mission_binding") as batch:
+        batch.add_column(
+            sa.Column("owner_epoch", sa.BigInteger(), server_default="0", nullable=False)
+        )
+        batch.add_column(
+            sa.Column("stop_epoch", sa.BigInteger(), server_default="0", nullable=False)
+        )
+        batch.add_column(sa.Column("owner_instance_id", sa.String(128)))
+        batch.add_column(sa.Column("owner_acquired_at", timestamp))
+    with op.batch_alter_table("script_build_artifact_version") as batch:
+        batch.drop_constraint("ck_artifact_business_type", type_="check")
+        batch.create_check_constraint("ck_artifact_business_type", _PHASE_THREE_CHECK)
+    with op.batch_alter_table("script_build_publication") as batch:
+        batch.create_unique_constraint(
+            "uq_publication_build_type", ["script_build_id", "publication_type"]
+        )
+    op.create_table(
+        "script_build_http_command",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("principal_scope_sha256", sa.String(64), nullable=False),
+        sa.Column("route_family", sa.String(64), nullable=False),
+        sa.Column("idempotency_key", sa.String(128), nullable=False),
+        sa.Column("request_fingerprint", sa.String(64), nullable=False),
+        sa.Column("preallocated_root_trace_id", sa.String(128)),
+        sa.Column("resource_id", sa.BigInteger()),
+        sa.Column("state", sa.String(16), nullable=False),
+        sa.Column("response_status", sa.Integer()),
+        sa.Column("response_json", sa.JSON()),
+        sa.Column("created_at", timestamp, nullable=False),
+        sa.Column("updated_at", timestamp, nullable=False),
+        sa.CheckConstraint("length(principal_scope_sha256) = 64", name="ck_http_principal_sha256"),
+        sa.CheckConstraint("length(request_fingerprint) = 64", name="ck_http_request_sha256"),
+        sa.CheckConstraint(
+            "state IN ('reserved', 'completed', 'failed')", name="ck_http_command_state"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "principal_scope_sha256",
+            "route_family",
+            "idempotency_key",
+            name="uq_http_command_scope_route_key",
+        ),
+        mysql_engine="InnoDB",
+        mysql_charset="utf8mb4",
+    )
+    op.create_index(
+        "ix_http_command_resource", "script_build_http_command", ["resource_id", "route_family"]
+    )
+    if context.get_context().dialect.name == "mysql":
+        op.alter_column(
+            "script_build_record",
+            "reson_trace_id",
+            existing_type=sa.String(200),
+            nullable=False,
+        )
+        op.create_unique_constraint(
+            "uq_script_build_record_reson_trace", "script_build_record", ["reson_trace_id"]
+        )
+    _create_mysql_views()
+
+
+def _assert_downgrade_safe() -> None:
+    if not _online():
+        raise RuntimeError("phase-three downgrade requires an online safety preflight")
+    bind = op.get_bind()
+    manifest_count = bind.execute(
+        sa.text(
+            "SELECT COUNT(*) FROM script_build_artifact_version "
+            "WHERE artifact_type='root_delivery_manifest'"
+        )
+    ).scalar_one()
+    final_count = bind.execute(
+        sa.text("SELECT COUNT(*) FROM script_build_publication WHERE publication_type='final'")
+    ).scalar_one()
+    pointer_or_epoch = bind.execute(
+        sa.text(
+            "SELECT COUNT(*) FROM script_build_mission_binding WHERE "
+            "accepted_root_artifact_version_id IS NOT NULL OR owner_epoch<>0 OR stop_epoch<>0"
+        )
+    ).scalar_one()
+    command_count = bind.execute(
+        sa.text("SELECT COUNT(*) FROM script_build_http_command")
+    ).scalar_one()
+    if any(
+        int(value) > 0 for value in (manifest_count, final_count, pointer_or_epoch, command_count)
+    ):
+        raise RuntimeError("cannot downgrade while phase-three publication or fencing facts exist")
+
+
+def downgrade() -> None:
+    _assert_downgrade_safe()
+    if context.get_context().dialect.name == "mysql":
+        for view in (
+            "script_build_runtime_record_v",
+            "script_build_candidate_paragraph_element_v",
+            "script_build_candidate_element_v",
+            "script_build_candidate_paragraph_v",
+        ):
+            op.execute(sa.text(f"DROP VIEW IF EXISTS {view}"))
+        op.drop_constraint(
+            "uq_script_build_record_reson_trace", "script_build_record", type_="unique"
+        )
+        op.alter_column(
+            "script_build_record",
+            "reson_trace_id",
+            existing_type=sa.String(200),
+            nullable=True,
+        )
+    op.drop_index("ix_http_command_resource", table_name="script_build_http_command")
+    op.drop_table("script_build_http_command")
+    with op.batch_alter_table("script_build_publication") as batch:
+        batch.drop_constraint("uq_publication_build_type", type_="unique")
+    with op.batch_alter_table("script_build_artifact_version") as batch:
+        batch.drop_constraint("ck_artifact_business_type", type_="check")
+        batch.create_check_constraint("ck_artifact_business_type", _PHASE_TWO_CHECK)
+    with op.batch_alter_table("script_build_mission_binding") as batch:
+        batch.drop_column("owner_acquired_at")
+        batch.drop_column("owner_instance_id")
+        batch.drop_column("stop_epoch")
+        batch.drop_column("owner_epoch")

script_build_host/src/script_build_host/domain/artifacts.py

@@ -5,6 +5,7 @@ from datetime import datetime
 from enum import StrEnum
 from typing import Any
 
+from .phase_three_artifacts import RootDeliveryManifestV1
 from .phase_two_artifacts import (
     CandidatePortfolioArtifactV1,
     ComparisonArtifactV1,
@@ -24,6 +25,7 @@ class ArtifactKind(StrEnum):
     COMPARISON = "comparison"
     STRUCTURED_SCRIPT = "structured_script"
     CANDIDATE_PORTFOLIO = "candidate_portfolio"
+    ROOT_DELIVERY_MANIFEST = "root_delivery_manifest"
 
 
 class ArtifactState(StrEnum):
@@ -179,6 +181,7 @@ BusinessArtifact = (
     | ComparisonArtifactV1
     | StructuredScriptArtifactV1
     | CandidatePortfolioArtifactV1
+    | RootDeliveryManifestV1
 )
 
 

script_build_host/src/script_build_host/domain/errors.py

@@ -116,3 +116,64 @@ class InvalidConfiguration(ScriptBuildError):
 class ProtocolViolation(ScriptBuildError):
     def __init__(self, summary: str) -> None:
         super().__init__("PROTOCOL_VIOLATION", summary)
+
+
+class LegacyCanonicalIdentityConflict(ScriptBuildError):
+    def __init__(self, summary: str = "legacy projection contains conflicting logical identities"):
+        super().__init__("LEGACY_CANONICAL_IDENTITY_CONFLICT", summary)
+
+
+class MissionAlreadyOwned(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__("MISSION_ALREADY_OWNED", "the mission is active in another Host owner")
+
+
+class MissionFencingTokenStale(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__("MISSION_FENCING_TOKEN_STALE", "the mission owner token is stale")
+
+
+class MissionStopRequested(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__("MISSION_STOP_REQUESTED", "the mission has a durable stop request")
+
+
+class PhaseThreeBoundaryNotReady(ScriptBuildError):
+    def __init__(self, summary: str = "the phase-three continuation preconditions are not met"):
+        super().__init__("PHASE_THREE_BOUNDARY_NOT_READY", summary)
+
+
+class PublicationReadbackMismatch(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "PUBLICATION_READBACK_MISMATCH",
+            "the published legacy projection does not match the accepted manifest",
+        )
+
+
+class PublicationStateInconsistent(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "PUBLICATION_STATE_INCONSISTENT",
+            "the final publication is only partially closed and requires reconciliation",
+        )
+
+
+class PublicationLockTimeout(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__("PUBLICATION_LOCK_TIMEOUT", "the publication lock timed out")
+
+
+class PublicationDeadlockRetryable(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "PUBLICATION_DEADLOCK_RETRYABLE", "the publication transaction was a deadlock victim"
+        )
+
+
+class HttpIdempotencyConflict(ScriptBuildError):
+    def __init__(self) -> None:
+        super().__init__(
+            "IDEMPOTENCY_KEY_CONFLICT",
+            "the Idempotency-Key was already used with a different request",
+        )

script_build_host/src/script_build_host/domain/phase_three_artifacts.py

@@ -0,0 +1,347 @@
+"""Phase-three delivery and legacy projection contracts.
+
+The legacy canonical graph deliberately contains no database or candidate IDs.  It
+is the single content identity used by the Root dry-run and publication readback.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Mapping
+from dataclasses import dataclass, replace
+from typing import Any, Self
+
+from .canonical_json import canonical_json_bytes, canonical_sha256, normalize_json
+from .digests import Sha256Digest
+from .errors import LegacyCanonicalIdentityConflict, ProtocolViolation
+from .phase_two_artifacts import StructuredScriptArtifactV1
+
+MAX_ROOT_DELIVERY_MANIFEST_BYTES = 64 * 1024
+MAX_BUILD_SUMMARY_CHARS = 2_000
+MAX_BLOCKING_DEFECTS = 50
+_ARTIFACT_PREFIX = "script-build://artifact-versions/"
+
+
+def _artifact_uri(value: str, label: str) -> None:
+    suffix = value.removeprefix(_ARTIFACT_PREFIX)
+    if not value.startswith(_ARTIFACT_PREFIX) or not suffix.isdigit():
+        raise ProtocolViolation(f"{label} must pin one immutable artifact version")
+
+
+def root_delivery_input_closure_digest(
+    *,
+    direction_ref: str,
+    direction_digest: str,
+    candidate_portfolio_ref: str,
+    candidate_portfolio_digest: str,
+    structured_script_ref: str,
+    structured_script_digest: str,
+) -> str:
+    """Bind the three immutable Root inputs without conflating their own lineages."""
+
+    for label, ref in (
+        ("direction_ref", direction_ref),
+        ("candidate_portfolio_ref", candidate_portfolio_ref),
+        ("structured_script_ref", structured_script_ref),
+    ):
+        _artifact_uri(ref, label)
+    for digest in (
+        direction_digest,
+        candidate_portfolio_digest,
+        structured_script_digest,
+    ):
+        Sha256Digest.from_wire(digest)
+    return canonical_sha256(
+        {
+            "schema_version": "root-delivery-input-closure/v1",
+            "direction": {"ref": direction_ref, "digest": direction_digest},
+            "candidate_portfolio": {
+                "ref": candidate_portfolio_ref,
+                "digest": candidate_portfolio_digest,
+            },
+            "structured_script": {
+                "ref": structured_script_ref,
+                "digest": structured_script_digest,
+            },
+        }
+    ).wire
+
+
+@dataclass(frozen=True, slots=True)
+class LegacyProjectionCanonicalV1:
+    """Physical-ID-independent business graph exposed by legacy detail reads."""
+
+    direction: str | None
+    summary: str | None
+    paragraphs: tuple[dict[str, Any], ...]
+    elements: tuple[dict[str, Any], ...]
+    paragraph_element_links: tuple[dict[str, Any], ...]
+    paragraph_count: int
+    element_count: int
+    link_count: int
+    schema_version: str = "legacy-projection-canonical/v1"
+
+    def __post_init__(self) -> None:
+        if self.schema_version != "legacy-projection-canonical/v1":
+            raise ProtocolViolation("unsupported legacy projection canonical schema")
+        if self.paragraph_count != len(self.paragraphs):
+            raise ProtocolViolation("canonical paragraph count does not match its graph")
+        if self.element_count != len(self.elements):
+            raise ProtocolViolation("canonical element count does not match its graph")
+        if self.link_count != len(self.paragraph_element_links):
+            raise ProtocolViolation("canonical link count does not match its graph")
+
+    def content_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": self.schema_version,
+            "direction": self.direction,
+            "summary": self.summary,
+            "paragraph_count": self.paragraph_count,
+            "element_count": self.element_count,
+            "link_count": self.link_count,
+            "paragraphs": list(self.paragraphs),
+            "elements": list(self.elements),
+            "paragraph_element_links": list(self.paragraph_element_links),
+        }
+
+    @property
+    def canonical_sha256(self) -> str:
+        return canonical_sha256(self.content_payload()).wire
+
+    @classmethod
+    def from_structured_script(
+        cls,
+        script: StructuredScriptArtifactV1,
+        *,
+        direction: str | None = None,
+        summary: str | None = None,
+    ) -> LegacyProjectionCanonicalV1:
+        return canonicalize_legacy_projection(script, direction=direction, summary=summary)
+
+
+@dataclass(frozen=True, slots=True)
+class RootDeliveryManifestV1:
+    direction_ref: str
+    candidate_portfolio_ref: str
+    structured_script_ref: str
+    input_closure_digest: str
+    legacy_projection_digest: str
+    build_summary: str = ""
+    blocking_defects: tuple[str, ...] = ()
+    canonical_sha256: str = ""
+
+    def __post_init__(self) -> None:
+        for label, value in (
+            ("direction_ref", self.direction_ref),
+            ("candidate_portfolio_ref", self.candidate_portfolio_ref),
+            ("structured_script_ref", self.structured_script_ref),
+        ):
+            _artifact_uri(value, label)
+        Sha256Digest.from_wire(self.input_closure_digest)
+        Sha256Digest.from_wire(self.legacy_projection_digest)
+        if len(self.build_summary) > MAX_BUILD_SUMMARY_CHARS:
+            raise ProtocolViolation("root delivery build summary exceeds 2,000 characters")
+        if len(self.blocking_defects) > MAX_BLOCKING_DEFECTS:
+            raise ProtocolViolation("root delivery may contain at most 50 blocking defects")
+        if any(not isinstance(item, str) or not item.strip() for item in self.blocking_defects):
+            raise ProtocolViolation("root delivery blocking defects must be non-blank strings")
+        if len(set(self.blocking_defects)) != len(self.blocking_defects):
+            raise ProtocolViolation("root delivery blocking defects must be unique")
+        if len(canonical_json_bytes(self.content_payload())) > MAX_ROOT_DELIVERY_MANIFEST_BYTES:
+            raise ProtocolViolation("root delivery manifest exceeds 64 KiB")
+        if self.canonical_sha256:
+            expected = canonical_sha256(self.content_payload()).wire
+            if self.canonical_sha256 != expected:
+                raise ProtocolViolation("root delivery manifest digest does not match its content")
+
+    def content_payload(self) -> dict[str, Any]:
+        return {
+            "schema_version": "root-delivery-manifest/v1",
+            "direction_ref": self.direction_ref,
+            "candidate_portfolio_ref": self.candidate_portfolio_ref,
+            "structured_script_ref": self.structured_script_ref,
+            "input_closure_digest": self.input_closure_digest,
+            "legacy_projection_digest": self.legacy_projection_digest,
+            "build_summary": self.build_summary,
+            "blocking_defects": list(self.blocking_defects),
+        }
+
+    def with_digest(self) -> Self:
+        return replace(self, canonical_sha256=canonical_sha256(self.content_payload()).wire)
+
+    def require_publishable(self) -> None:
+        if self.blocking_defects:
+            raise ProtocolViolation("a validated root delivery cannot contain blocking defects")
+
+    @classmethod
+    def from_payload(cls, value: Mapping[str, Any]) -> Self:
+        if set(value) - {
+            "schema_version",
+            "direction_ref",
+            "candidate_portfolio_ref",
+            "structured_script_ref",
+            "input_closure_digest",
+            "legacy_projection_digest",
+            "build_summary",
+            "blocking_defects",
+        }:
+            raise ProtocolViolation("root delivery manifest contains unknown fields")
+        if value.get("schema_version") != "root-delivery-manifest/v1":
+            raise ProtocolViolation("unsupported root delivery manifest schema")
+        defects = value.get("blocking_defects", [])
+        if not isinstance(defects, list) or any(not isinstance(item, str) for item in defects):
+            raise ProtocolViolation("root delivery blocking_defects must be a string array")
+        return cls(
+            direction_ref=str(value.get("direction_ref", "")),
+            candidate_portfolio_ref=str(value.get("candidate_portfolio_ref", "")),
+            structured_script_ref=str(value.get("structured_script_ref", "")),
+            input_closure_digest=str(value.get("input_closure_digest", "")),
+            legacy_projection_digest=str(value.get("legacy_projection_digest", "")),
+            build_summary=str(value.get("build_summary", "")),
+            blocking_defects=tuple(defects),
+        )
+
+
+def _element_key(value: Mapping[str, Any]) -> str:
+    natural = [
+        str(value.get("dimension_primary", "")),
+        str(value.get("dimension_secondary", "")),
+        str(value.get("name", "")),
+    ]
+    return hashlib.sha256(canonical_json_bytes(natural)).hexdigest()
+
+
+def _assert_parent_graph(parents: Mapping[int, int | None]) -> None:
+    for start in parents:
+        seen: set[int] = set()
+        current: int | None = start
+        while current is not None:
+            if current in seen:
+                raise LegacyCanonicalIdentityConflict("paragraph parent graph contains a cycle")
+            seen.add(current)
+            if current not in parents:
+                raise LegacyCanonicalIdentityConflict("paragraph parent reference is dangling")
+            current = parents[current]
+
+
+def canonicalize_legacy_projection(
+    script: StructuredScriptArtifactV1,
+    *,
+    direction: str | None = None,
+    summary: str | None = None,
+) -> LegacyProjectionCanonicalV1:
+    """Convert a frozen StructuredScript to the one canonical legacy graph."""
+
+    active_paragraphs = [item for item in script.paragraphs if item.is_active]
+    candidate_to_index = {item.paragraph_id: item.paragraph_index for item in active_paragraphs}
+    if len(candidate_to_index) != len(active_paragraphs) or len(
+        set(candidate_to_index.values())
+    ) != len(active_paragraphs):
+        raise LegacyCanonicalIdentityConflict("active paragraph logical identities are duplicated")
+    parents: dict[int, int | None] = {}
+    paragraph_payloads: list[dict[str, Any]] = []
+    for paragraph in active_paragraphs:
+        parent_key = None
+        if paragraph.parent_id is not None:
+            parent_key = candidate_to_index.get(paragraph.parent_id)
+            if parent_key is None:
+                raise LegacyCanonicalIdentityConflict("paragraph parent reference is dangling")
+        parents[paragraph.paragraph_index] = parent_key
+        paragraph_payloads.append(
+            {
+                "logical_key": paragraph.paragraph_index,
+                "paragraph_index": paragraph.paragraph_index,
+                "level": paragraph.level,
+                "parent_logical_key": parent_key,
+                "name": paragraph.name,
+                "content_range": paragraph.content_range,
+                "theme": paragraph.theme,
+                "form": paragraph.form,
+                "function": paragraph.function,
+                "feeling": paragraph.feeling,
+                "theme_elements": list(paragraph.theme_elements),
+                "form_elements": list(paragraph.form_elements),
+                "function_elements": list(paragraph.function_elements),
+                "feeling_elements": list(paragraph.feeling_elements),
+                "description": paragraph.description,
+                "full_description": paragraph.full_description,
+            }
+        )
+    _assert_parent_graph(parents)
+
+    active_elements = [item for item in script.elements if item.is_active]
+    element_keys: dict[int, str] = {}
+    seen_natural: set[tuple[str, str, str]] = set()
+    element_payloads: list[dict[str, Any]] = []
+    for element in active_elements:
+        natural = (element.dimension_primary, element.dimension_secondary, element.name)
+        if natural in seen_natural or element.element_id in element_keys:
+            raise LegacyCanonicalIdentityConflict(
+                "active element natural identities are duplicated"
+            )
+        seen_natural.add(natural)
+        payload = element.to_payload()
+        key = _element_key(payload)
+        element_keys[element.element_id] = key
+        element_payloads.append(
+            {
+                "logical_key": key,
+                "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": list(element.support_elements),
+            }
+        )
+
+    link_payloads: list[dict[str, Any]] = []
+    seen_links: set[tuple[int, str]] = set()
+    for link in script.paragraph_element_links:
+        paragraph_key = candidate_to_index.get(link.paragraph_id)
+        element_key = element_keys.get(link.element_id)
+        if paragraph_key is None or element_key is None:
+            raise LegacyCanonicalIdentityConflict("paragraph-element link is dangling")
+        pair = (paragraph_key, element_key)
+        if pair in seen_links:
+            raise LegacyCanonicalIdentityConflict("paragraph-element logical link is duplicated")
+        seen_links.add(pair)
+        link_payloads.append(
+            {"paragraph_logical_key": paragraph_key, "element_logical_key": element_key}
+        )
+
+    paragraph_payloads.sort(key=lambda item: int(item["logical_key"]))
+    element_payloads.sort(key=lambda item: str(item["logical_key"]))
+    link_payloads.sort(
+        key=lambda item: (int(item["paragraph_logical_key"]), str(item["element_logical_key"]))
+    )
+    return LegacyProjectionCanonicalV1(
+        direction=direction,
+        summary=summary,
+        paragraphs=tuple(normalize_json(item) for item in paragraph_payloads),
+        elements=tuple(normalize_json(item) for item in element_payloads),
+        paragraph_element_links=tuple(normalize_json(item) for item in link_payloads),
+        paragraph_count=len(paragraph_payloads),
+        element_count=len(element_payloads),
+        link_count=len(link_payloads),
+    )
+
+
+def hydrate_root_delivery_manifest(
+    payload: Mapping[str, Any], digest: str = ""
+) -> RootDeliveryManifestV1:
+    value = RootDeliveryManifestV1.from_payload(payload)
+    return replace(value, canonical_sha256=digest) if digest else value
+
+
+__all__ = [
+    "MAX_BLOCKING_DEFECTS",
+    "MAX_BUILD_SUMMARY_CHARS",
+    "MAX_ROOT_DELIVERY_MANIFEST_BYTES",
+    "LegacyProjectionCanonicalV1",
+    "RootDeliveryManifestV1",
+    "canonicalize_legacy_projection",
+    "hydrate_root_delivery_manifest",
+    "root_delivery_input_closure_digest",
+]

script_build_host/src/script_build_host/domain/ports.py

@@ -155,6 +155,7 @@ class LegacyBuildStateRepository(Protocol):
         agent_config: dict[str, Any] | None,
         data_source_url: str | None,
         strategies_config: dict[str, Any],
+        root_trace_id: str | None = None,
     ) -> int: ...
 
     async def set_status(

script_build_host/src/script_build_host/domain/records.py

@@ -26,6 +26,49 @@ class PublicationState(StrEnum):
     FAILED = "failed"
 
 
+class RecoveryClassification(StrEnum):
+    ALREADY_COMPLETE = "already_complete"
+    FINALIZE_ONLY = "finalize_only"
+    RESUME_PENDING_OPERATION = "resume_pending_operation"
+    RESUME_VALIDATION = "resume_validation"
+    REPLAN_REQUIRED = "replan_required"
+    MANUAL_RECONCILIATION = "manual_reconciliation"
+
+
+@dataclass(frozen=True, slots=True)
+class MissionOwnerToken:
+    script_build_id: int
+    root_trace_id: str
+    owner_epoch: int
+    stop_epoch: int
+    owner_instance_id: str
+
+    def __post_init__(self) -> None:
+        if self.script_build_id < 1 or self.owner_epoch < 1 or self.stop_epoch < 0:
+            raise ValueError("mission owner token contains an invalid epoch")
+        if not self.root_trace_id.strip() or not self.owner_instance_id.strip():
+            raise ValueError("mission owner token identities must not be blank")
+
+
+@dataclass(frozen=True, slots=True)
+class PublicationResult:
+    script_build_id: int
+    publication_id: int
+    publication_type: PublicationType
+    state: PublicationState
+    legacy_projection_digest: str
+    committed: bool
+    post_commit_readback_ok: bool = True
+
+
+@dataclass(frozen=True, slots=True)
+class ResumeResult:
+    script_build_id: int
+    classification: RecoveryClassification
+    status: BuildStatus
+    detail: str = ""
+
+
 @dataclass(frozen=True, slots=True)
 class MissionBinding:
     binding_id: int
@@ -38,6 +81,10 @@ class MissionBinding:
     schema_version: str
     created_at: datetime
     updated_at: datetime
+    owner_epoch: int = 0
+    stop_epoch: int = 0
+    owner_instance_id: str | None = None
+    owner_acquired_at: datetime | None = None
 
 
 @dataclass(frozen=True, slots=True)

script_build_host/src/script_build_host/domain/task_contracts.py

@@ -53,6 +53,7 @@ class ScriptTaskKind(StrEnum):
     COMPARE = "compare"
     COMPOSE = "compose"
     CANDIDATE_PORTFOLIO = "candidate-portfolio"
+    ROOT_DELIVERY = "root-delivery"
 
 
 class ScriptIntentClass(StrEnum):
@@ -78,6 +79,7 @@ _OUTPUT_SCHEMAS: dict[ScriptTaskKind, frozenset[str]] = {
     ScriptTaskKind.COMPARE: frozenset({"comparison-artifact/v1"}),
     ScriptTaskKind.COMPOSE: frozenset({"structured-script/v1"}),
     ScriptTaskKind.CANDIDATE_PORTFOLIO: frozenset({"candidate-portfolio/v1"}),
+    ScriptTaskKind.ROOT_DELIVERY: frozenset({"root-delivery-manifest/v1"}),
 }
 
 _ADOPTION_FIELDS_KINDS = frozenset({ScriptTaskKind.COMPOSE, ScriptTaskKind.CANDIDATE_PORTFOLIO})

script_build_host/src/script_build_host/infrastructure/legacy_tables.py

@@ -202,6 +202,11 @@ script_build_record = Table(
     Column("end_time", DateTime),
 )
 
+_runtime_view_metadata = MetaData()
+script_build_runtime_record = script_build_record.to_metadata(
+    _runtime_view_metadata, name="script_build_runtime_record_v"
+)
+
 script_build_paragraph = Table(
     "script_build_paragraph",
     legacy_metadata,
@@ -267,6 +272,19 @@ script_build_paragraph_element = Table(
     ),
 )
 
+# Runtime candidate writers use the CHECK OPTION views on MySQL.  These Table
+# objects are mappings only; Alembic owns view creation.
+_candidate_view_metadata = MetaData()
+script_build_candidate_paragraph = script_build_paragraph.to_metadata(
+    _candidate_view_metadata, name="script_build_candidate_paragraph_v"
+)
+script_build_candidate_element = script_build_element.to_metadata(
+    _candidate_view_metadata, name="script_build_candidate_element_v"
+)
+script_build_candidate_paragraph_element = script_build_paragraph_element.to_metadata(
+    _candidate_view_metadata, name="script_build_candidate_paragraph_element_v"
+)
+
 script_build_task_plan_step = Table(
     "script_build_task_plan_step",
     legacy_metadata,
@@ -291,3 +309,27 @@ script_build_task_plan_step = Table(
         unique=True,
     ),
 )
+
+# Historical installations used these tables for legacy detail/log fallbacks.
+# Phase three maps them explicitly instead of reflecting an unbounded schema.
+script_build_decision_trace = Table(
+    "script_build_decision_trace",
+    legacy_metadata,
+    Column("id", _primary_key_type, primary_key=True, autoincrement=True),
+    Column("script_build_id", BigInteger, nullable=False),
+    Column("trace_type", String(64)),
+    Column("content", JSON),
+    Column("summary", Text),
+    Column("created_at", DateTime),
+    Index("ix_script_build_decision_trace_build", "script_build_id", "id"),
+)
+
+script_build_log = Table(
+    "script_build_log",
+    legacy_metadata,
+    Column("id", _primary_key_type, primary_key=True, autoincrement=True),
+    Column("script_build_id", BigInteger, nullable=False),
+    Column("log_content", Text),
+    Column("created_at", DateTime),
+    Index("ix_script_build_log_build", "script_build_id", "id"),
+)

script_build_host/src/script_build_host/infrastructure/tables.py

@@ -13,9 +13,14 @@ from sqlalchemy import (
     Table,
     UniqueConstraint,
 )
+from sqlalchemy.dialects import mysql
 
 metadata = MetaData()
 _primary_key_type = BigInteger().with_variant(Integer, "sqlite")
+_timestamp_type = DateTime(timezone=True).with_variant(
+    mysql.DATETIME(fsp=6),  # type: ignore[no-untyped-call]
+    "mysql",
+)
 
 mission_binding_table = Table(
     "script_build_mission_binding",
@@ -28,8 +33,12 @@ mission_binding_table = Table(
     Column("accepted_root_artifact_version_id", BigInteger),
     Column("engine_version", String(64), nullable=False),
     Column("schema_version", String(32), nullable=False),
-    Column("created_at", DateTime(timezone=True), nullable=False),
-    Column("updated_at", DateTime(timezone=True), nullable=False),
+    Column("owner_epoch", BigInteger, nullable=False, default=0, server_default="0"),
+    Column("stop_epoch", BigInteger, nullable=False, default=0, server_default="0"),
+    Column("owner_instance_id", String(128)),
+    Column("owner_acquired_at", _timestamp_type),
+    Column("created_at", _timestamp_type, nullable=False),
+    Column("updated_at", _timestamp_type, nullable=False),
     UniqueConstraint("script_build_id", name="uq_binding_build"),
     UniqueConstraint("root_trace_id", name="uq_binding_root"),
 )
@@ -44,7 +53,7 @@ input_snapshot_table = Table(
     Column("canonical_sha256", String(64), nullable=False),
     Column("prompt_manifest_json", JSON, nullable=False),
     Column("schema_version", String(32), nullable=False),
-    Column("created_at", DateTime(timezone=True), nullable=False),
+    Column("created_at", _timestamp_type, nullable=False),
     UniqueConstraint("script_build_id", "version", name="uq_input_version"),
     UniqueConstraint("script_build_id", "canonical_sha256", name="uq_input_hash"),
     CheckConstraint("length(canonical_sha256) = 64", name="ck_input_sha256_length"),
@@ -65,15 +74,16 @@ artifact_version_table = Table(
     Column("canonical_json", JSON),
     Column("canonical_sha256", String(64)),
     Column("state", String(16), nullable=False),
-    Column("created_at", DateTime(timezone=True), nullable=False),
-    Column("frozen_at", DateTime(timezone=True)),
-    Column("published_at", DateTime(timezone=True)),
+    Column("created_at", _timestamp_type, nullable=False),
+    Column("frozen_at", _timestamp_type),
+    Column("published_at", _timestamp_type),
     UniqueConstraint("attempt_id", name="uq_artifact_attempt"),
     UniqueConstraint("script_build_id", "legacy_branch_id", name="uq_artifact_branch"),
     CheckConstraint(
         "artifact_type IN ("
         "'evidence', 'direction', 'structure', 'paragraph', 'element_set', "
-        "'comparison', 'structured_script', 'candidate_portfolio'"
+        "'comparison', 'structured_script', 'candidate_portfolio', "
+        "'root_delivery_manifest'"
         ")",
         name="ck_artifact_business_type",
     ),
@@ -104,10 +114,11 @@ publication_table = Table(
     Column("publication_revision", BigInteger, nullable=False, default=0),
     Column("last_error_code", String(64)),
     Column("last_error_summary", String(1000)),
-    Column("created_at", DateTime(timezone=True), nullable=False),
-    Column("updated_at", DateTime(timezone=True), nullable=False),
-    Column("published_at", DateTime(timezone=True)),
+    Column("created_at", _timestamp_type, nullable=False),
+    Column("updated_at", _timestamp_type, nullable=False),
+    Column("published_at", _timestamp_type),
     UniqueConstraint("accept_decision_id", name="uq_publication_decision"),
+    UniqueConstraint("script_build_id", "publication_type", name="uq_publication_build_type"),
     CheckConstraint("publication_type IN ('direction', 'final')", name="ck_publication_type"),
     CheckConstraint(
         "state IN ('pending', 'publishing', 'published', 'failed')", name="ck_publication_state"
@@ -115,3 +126,30 @@ publication_table = Table(
     CheckConstraint("length(expected_sha256) = 64", name="ck_publication_sha256_length"),
     Index("ix_publication_build", "script_build_id", "publication_type", "state"),
 )
+
+http_command_table = Table(
+    "script_build_http_command",
+    metadata,
+    Column("id", _primary_key_type, primary_key=True, autoincrement=True),
+    Column("principal_scope_sha256", String(64), nullable=False),
+    Column("route_family", String(64), nullable=False),
+    Column("idempotency_key", String(128), nullable=False),
+    Column("request_fingerprint", String(64), nullable=False),
+    Column("preallocated_root_trace_id", String(128)),
+    Column("resource_id", BigInteger),
+    Column("state", String(16), nullable=False),
+    Column("response_status", Integer),
+    Column("response_json", JSON),
+    Column("created_at", _timestamp_type, nullable=False),
+    Column("updated_at", _timestamp_type, nullable=False),
+    UniqueConstraint(
+        "principal_scope_sha256",
+        "route_family",
+        "idempotency_key",
+        name="uq_http_command_scope_route_key",
+    ),
+    CheckConstraint("length(principal_scope_sha256) = 64", name="ck_http_principal_sha256"),
+    CheckConstraint("length(request_fingerprint) = 64", name="ck_http_request_sha256"),
+    CheckConstraint("state IN ('reserved', 'completed', 'failed')", name="ck_http_command_state"),
+    Index("ix_http_command_resource", "resource_id", "route_family"),
+)

script_build_host/src/script_build_host/repositories/sqlalchemy.py

@@ -31,6 +31,10 @@ from script_build_host.domain.errors import (
     ProtocolViolation,
 )
 from script_build_host.domain.input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
+from script_build_host.domain.phase_three_artifacts import (
+    RootDeliveryManifestV1,
+    hydrate_root_delivery_manifest,
+)
 from script_build_host.domain.phase_two_artifacts import (
     CandidatePortfolioArtifactV1,
     ComparisonArtifactV1,
@@ -242,6 +246,10 @@ def _binding_from_row(row: RowMapping) -> MissionBinding:
         schema_version=str(row["schema_version"]),
         created_at=cast(datetime, row["created_at"]),
         updated_at=cast(datetime, row["updated_at"]),
+        owner_epoch=int(row.get("owner_epoch") or 0),
+        stop_epoch=int(row.get("stop_epoch") or 0),
+        owner_instance_id=row.get("owner_instance_id"),
+        owner_acquired_at=row.get("owner_acquired_at"),
     )
 
 
@@ -440,6 +448,7 @@ def _artifact_kind(value: BusinessArtifact) -> ArtifactKind:
         (ComparisonArtifactV1, ArtifactKind.COMPARISON),
         (StructuredScriptArtifactV1, ArtifactKind.STRUCTURED_SCRIPT),
         (CandidatePortfolioArtifactV1, ArtifactKind.CANDIDATE_PORTFOLIO),
+        (RootDeliveryManifestV1, ArtifactKind.ROOT_DELIVERY_MANIFEST),
     )
     for artifact_type, kind in mapping:
         if isinstance(value, artifact_type):
@@ -479,6 +488,8 @@ def _hydrate_artifact(kind: ArtifactKind, payload: dict[str, Any], digest: str)
             legacy_markdown=str(payload.get("legacy_markdown", "")),
             canonical_sha256=digest,
         )
+    if kind is ArtifactKind.ROOT_DELIVERY_MANIFEST:
+        return hydrate_root_delivery_manifest(payload, digest)
     return replace(hydrate_phase_two_artifact(payload), canonical_sha256=digest)
 
 
@@ -759,8 +770,6 @@ class SqlAlchemyPublicationRepository:
         artifact_version_id: int,
         expected_sha256: str,
     ) -> Publication:
-        if publication_type is not PublicationType.DIRECTION:
-            raise ProtocolViolation("phase one can publish only an accepted direction")
         digest = Sha256Digest.from_wire(expected_sha256)
         now = _utc_now()
         try:
@@ -801,6 +810,8 @@ class SqlAlchemyPublicationRepository:
                 publication_id = _inserted_id(result)
         except IntegrityError:
             existing = await self._find_by_decision(accept_decision_id)
+            if existing is None:
+                existing = await self._find_by_build_type(script_build_id, publication_type)
             if existing is None:
                 raise
             return self._validate_existing(
@@ -812,6 +823,23 @@ class SqlAlchemyPublicationRepository:
             )
         return await self._get(publication_id)
 
+    async def _find_by_build_type(
+        self, script_build_id: int, publication_type: PublicationType
+    ) -> RowMapping | None:
+        async with self._sessions() as session:
+            return (
+                (
+                    await session.execute(
+                        select(publication_table).where(
+                            publication_table.c.script_build_id == script_build_id,
+                            publication_table.c.publication_type == publication_type.value,
+                        )
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+
     @staticmethod
     def _validate_existing(
         row: RowMapping,

script_build_host/tests/fixtures/legacy_http_goldens_manifest.json

@@ -0,0 +1,64 @@
+{
+  "schema_version": "legacy-http-golden-manifest/v1",
+  "host_baseline_commit": "1495d063",
+  "legacy_source": {
+    "kind": "deployed-legacy-service",
+    "base_url": "https://pattern.aiddit.com",
+    "source_commit": "unavailable-from-deployment",
+    "captured_at": "2026-07-19T16:00:00Z"
+  },
+  "redaction": {
+    "rules": [
+      "No authorization header, cookie, principal identifier, or database credential is stored.",
+      "Payload hashes cover the unmodified public response bytes; payload bodies are not checked in."
+    ]
+  },
+  "captures": [
+    {
+      "name": "run-454-detail",
+      "method": "GET",
+      "route": "/api/pattern/script_builds/454",
+      "status": 200,
+      "sha256": "87a2d7bc11e70391c92cb52013365d758338e0d918dd47339c5a83c214cd42f6",
+      "comparison": "legacy-fields-exact-overlay-declared",
+      "legacy_top_level_fields": [
+        "success",
+        "topic",
+        "build",
+        "paragraphs",
+        "elements",
+        "decision_traces",
+        "post_datas",
+        "script_datas"
+      ],
+      "allowed_overlay_fields": []
+    },
+    {
+      "name": "run-454-log",
+      "method": "GET",
+      "route": "/api/pattern/script_builds/454/log",
+      "status": 200,
+      "sha256": "ac62b87aac1999ab8eb94c8a71b0be5f35457e11cdc89b10c3d10cf7ec7eaa07",
+      "comparison": "legacy-fields-exact-overlay-declared",
+      "legacy_top_level_fields": [
+        "success",
+        "log_content",
+        "live",
+        "status"
+      ],
+      "allowed_overlay_fields": []
+    }
+  ],
+  "uncaptured_routes": [
+    "start",
+    "upload",
+    "list",
+    "prefill",
+    "retry",
+    "delete",
+    "favorite",
+    "overview",
+    "stop",
+    "trace_messages"
+  ]
+}

script_build_host/tests/test_legacy_goldens.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+
+
+def test_checked_in_legacy_capture_manifest_is_complete_and_bounded() -> None:
+    path = Path(__file__).parent / "fixtures" / "legacy_http_goldens_manifest.json"
+    manifest = json.loads(path.read_text(encoding="utf-8"))
+    assert manifest["schema_version"] == "legacy-http-golden-manifest/v1"
+    assert manifest["host_baseline_commit"] == "1495d063"
+    assert manifest["legacy_source"]["source_commit"]
+    assert manifest["redaction"]["rules"]
+    captures = manifest["captures"]
+    assert {item["name"] for item in captures} == {"run-454-detail", "run-454-log"}
+    for capture in captures:
+        assert capture["method"] == "GET"
+        assert capture["route"].startswith("/api/pattern/script_builds/454")
+        assert capture["status"] == 200
+        assert re.fullmatch(r"[0-9a-f]{64}", capture["sha256"])
+        assert capture["comparison"] in {
+            "exact",
+            "legacy-fields-exact-overlay-declared",
+        }
+        assert len(capture["legacy_top_level_fields"]) == len(
+            set(capture["legacy_top_level_fields"])
+        )