test(script-build): gate phase three end to end

script_build_host/PHASE_THREE_RELEASE.md

@@ -0,0 +1,78 @@
+# Phase 3 release gate
+
+Phase 3 keeps the generic Agent state machine unchanged. The Host owns Root delivery,
+publication, recovery, authorization, and the legacy HTTP projection. The implementation
+baseline is `main@1495d063`; migration `0003_phase_three_publication` is the first schema
+revision that permits final publication.
+
+## Database deployment
+
+Use MySQL 8.0.16 or newer. Verify `migrations/CHECKSUMS.sha256`, back up the legacy schema,
+and run migrations only with the deployment administrator:
+
+```bash
+SCRIPT_BUILD_WRITE_DATABASE_URL='mysql+asyncmy://migration_admin:...@db/script_build' \
+  python -m alembic upgrade head
+```
+
+The 0003 preflight refuses missing or duplicate `script_build_record.reson_trace_id` values and
+duplicate publication identities. The revision adds `DATETIME(6)` ownership fencing, the Root
+manifest artifact kind, the HTTP command journal, one final publication identity per build, and
+candidate/runtime views with `WITH CASCADED CHECK OPTION`. Its revision identifier deliberately
+fits Alembic's MySQL `version_num VARCHAR(32)`.
+
+Production must use three different database principals:
+
+- read: `SELECT` on authorized legacy/Host tables and views only;
+- runtime: Host-table mutations, candidate-view writes, and only the legacy build mutations
+  required before final success; it must not be able to write branch 0 or set `status=success`;
+- final publisher: the minimum branch-0, artifact, binding, publication, and build update grants
+  required by `SqlAlchemyFinalPublicationUnitOfWork`.
+
+Grant names are deployment-specific and therefore are not embedded in Alembic. Revoke the old
+unrestricted writer before enabling Phase 3. Configure `SCRIPT_BUILD_READ_DATABASE_URL`,
+`SCRIPT_BUILD_WRITE_DATABASE_URL`, and `SCRIPT_BUILD_FINAL_DATABASE_URL`; production startup
+rejects reused usernames. MySQL sessions are explicitly opened at `READ COMMITTED`.
+Use `migrations/mysql_phase_three_grants.sql.example` as the reviewed least-privilege template.
+
+## Publication and recovery
+
+Publication is allowed only after one frozen Manifest, passed Root Validation, and explicit Root
+ACCEPT. A single final AsyncSession replaces branch 0, reads the canonical legacy projection back,
+publishes the Manifest and StructuredScript, sets the accepted pointer, records the publication,
+and changes the build to `success`. Any pre-commit error rolls back all of those writes. A lost
+commit acknowledgement is classified by the unique final publication identity and is read back;
+it is never blindly replayed.
+
+Use per-build endpoints only; startup does not scan the database:
+
+- `POST .../{id}/phase-three/advance` enters the frozen Root delivery continuation;
+- `POST .../{id}/resume` classifies durable state and performs only a safe continuation;
+- `POST .../{id}/finalize` retries the same immutable final identity;
+- `POST .../{id}/reconcile` is inspection-only unless explicitly retrying finalize;
+- `GET .../{id}/publication?type=direction|final` reports publication state.
+
+If recovery reports manual reconciliation, do not edit an Artifact, digest, Decision, or accepted
+pointer. Preserve the task ledger and database facts for inspection.
+
+## Required verification
+
+Run all of the following before release:
+
+```bash
+python -m pytest
+SCRIPT_BUILD_TEST_MYSQL_DSN='mysql+asyncmy://.../disposable_db' python -m pytest -m mysql
+python -m ruff check src migrations tests
+python -m ruff format --check src migrations tests
+python -m mypy src/script_build_host
+python -m compileall -q src
+git diff --check
+```
+
+The MySQL database must first contain the legacy tables and be upgraded from 0001 through 0003.
+The gate checks MySQL version, `READ COMMITTED`, `DATETIME(6)`, enforced CHECK constraints,
+candidate/runtime view check options, the Alembic head, and a real atomic final publication.
+
+`tests/fixtures/legacy_http_goldens_manifest.json` records the two legacy Run 454 responses that
+were actually captured, including route, status, redaction policy, comparison mode, and SHA-256.
+Uncaptured legacy routes are listed explicitly rather than represented by invented fixtures.

script_build_host/migrations/mysql_phase_three_grants.sql.example

@@ -0,0 +1,47 @@
+-- Substitute database and account names in deployment automation. Do not run as runtime.
+-- The migration administrator executes this after Alembic 0003 and then revokes the old writer.
+
+GRANT SELECT ON `${SCRIPT_BUILD_DATABASE}`.* TO `${SCRIPT_BUILD_READ_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+
+GRANT SELECT ON `${SCRIPT_BUILD_DATABASE}`.* TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_candidate_paragraph_v
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_candidate_element_v
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_candidate_paragraph_element_v
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE ON `${SCRIPT_BUILD_DATABASE}`.script_build_runtime_record_v
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT UPDATE (is_favorited, is_deleted) ON `${SCRIPT_BUILD_DATABASE}`.script_build_record
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_mission_binding
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_input_snapshot
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_artifact_version
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_publication
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_http_command
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_task_plan_step
+  TO `${SCRIPT_BUILD_RUNTIME_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+
+GRANT SELECT ON `${SCRIPT_BUILD_DATABASE}`.* TO `${SCRIPT_BUILD_FINAL_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_paragraph
+  TO `${SCRIPT_BUILD_FINAL_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_element
+  TO `${SCRIPT_BUILD_FINAL_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE, DELETE ON `${SCRIPT_BUILD_DATABASE}`.script_build_paragraph_element
+  TO `${SCRIPT_BUILD_FINAL_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT UPDATE ON `${SCRIPT_BUILD_DATABASE}`.script_build_record
+  TO `${SCRIPT_BUILD_FINAL_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT UPDATE ON `${SCRIPT_BUILD_DATABASE}`.script_build_mission_binding
+  TO `${SCRIPT_BUILD_FINAL_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT UPDATE ON `${SCRIPT_BUILD_DATABASE}`.script_build_artifact_version
+  TO `${SCRIPT_BUILD_FINAL_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+GRANT INSERT, UPDATE ON `${SCRIPT_BUILD_DATABASE}`.script_build_publication
+  TO `${SCRIPT_BUILD_FINAL_USER}`@`${SCRIPT_BUILD_DB_HOST}`;
+
+-- Deployment-specific old writer account:
+-- REVOKE ALL PRIVILEGES, GRANT OPTION FROM `old_script_build_writer`@`${SCRIPT_BUILD_DB_HOST}`;

script_build_host/tests/test_agent_surface.py

@@ -14,7 +14,10 @@ from script_build_host.agents.model_resolver import (
 )
 from script_build_host.agents.presets import register_script_presets
 from script_build_host.agents.prompt_resolver import SnapshotRoleSystemPromptResolver
-from script_build_host.agents.prompts import script_build_prompt_manifest
+from script_build_host.agents.prompts import (
+    phase_three_prompt_manifest,
+    script_build_prompt_manifest,
+)
 from script_build_host.domain.canonical_json import canonical_sha256
 from script_build_host.domain.errors import ProtocolViolation
 from script_build_host.domain.input_snapshot import ScriptBuildInputSnapshotV1
@@ -77,6 +80,21 @@ def test_script_presets_are_exact_and_never_expose_legacy_control_tools() -> Non
     for validator_name in ("script_retrieval_validator", "script_candidate_validator"):
         assert "view_frozen_images" in (get_preset(validator_name).allowed_tools or [])
 
+    assert set(get_preset("script_root_worker").allowed_tools or []) == {
+        "read_input_snapshot",
+        "read_accepted_portfolio",
+        "legacy_projection_dry_run",
+        "save_root_delivery_manifest",
+        "submit_attempt",
+    }
+    assert set(get_preset("script_root_validator").allowed_tools or []) == {
+        "read_input_snapshot",
+        "read_accepted_portfolio",
+        "query_validation_evidence",
+        "deterministic_precheck",
+        "submit_validation",
+    }
+
 
 def test_new_build_prompt_manifest_freezes_every_phase_one_and_phase_two_role() -> None:
     manifest = script_build_prompt_manifest()
@@ -97,6 +115,15 @@ def test_new_build_prompt_manifest_freezes_every_phase_one_and_phase_two_role()
         assert len(str(item["content_sha256"])) == 71
 
 
+def test_phase_three_prompt_manifest_is_versioned_separately() -> None:
+    manifest = phase_three_prompt_manifest()
+    assert {str(item["preset"]) for item in manifest} == {
+        "script_root_worker",
+        "script_root_validator",
+    }
+    assert all(str(item["content_sha256"]).startswith("sha256:") for item in manifest)
+
+
 class _Gateway:
     coordinator = SimpleNamespace(task_store=None)
 

script_build_host/tests/test_api_security.py

@@ -226,6 +226,12 @@ async def test_observation_is_build_scoped_and_control_routes_are_not_mounted()
     paths = set(app.openapi()["paths"])
     assert not any(path.endswith("/operations") for path in paths)
     assert not any(path.startswith("/api/traces") for path in paths)
+    assert (
+        "patch" in app.openapi()["paths"]["/api/pattern/script_builds/{script_build_id}/favorite"]
+    )
+    assert "/api/pattern/script_builds/overview/{topic_build_id}" in paths
+    assert "/api/pattern/script_builds/{script_build_id}/trace_messages" in paths
+    assert "/script_build_detail/{script_build_id}" in paths
     async with httpx.AsyncClient(
         transport=httpx.ASGITransport(app=app), base_url="http://test"
     ) as client:

script_build_host/tests/test_http_commands.py

@@ -0,0 +1,209 @@
+from __future__ import annotations
+
+import asyncio
+from datetime import UTC, datetime
+from typing import Any
+
+import pytest
+from sqlalchemy import insert, select
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+
+from script_build_host.domain.canonical_json import canonical_sha256
+from script_build_host.domain.errors import HttpIdempotencyConflict, ProtocolViolation
+from script_build_host.domain.records import Principal
+from script_build_host.infrastructure.http_commands import HttpCommandJournal
+from script_build_host.infrastructure.tables import http_command_table
+
+
+@pytest.mark.asyncio
+async def test_http_command_replays_completed_response_and_rejects_fingerprint_conflict(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    journal = HttpCommandJournal(sessions)
+    principal = Principal("subject", "tenant")
+    calls = 0
+
+    async def command() -> tuple[int, dict[str, Any]]:
+        nonlocal calls
+        calls += 1
+        return 201, {"success": True, "resource_id": 9}
+
+    first = await journal.execute(
+        principal=principal,
+        route_family="script-build:test",
+        idempotency_key="same",
+        request_payload={"value": 1},
+        command=command,
+        resource_id=9,
+    )
+    second = await journal.execute(
+        principal=principal,
+        route_family="script-build:test",
+        idempotency_key="same",
+        request_payload={"value": 1},
+        command=command,
+        resource_id=9,
+    )
+    assert first == second == (201, {"success": True, "resource_id": 9})
+    assert calls == 1
+
+    with pytest.raises(HttpIdempotencyConflict):
+        await journal.execute(
+            principal=principal,
+            route_family="script-build:test",
+            idempotency_key="same",
+            request_payload={"value": 2},
+            command=command,
+        )
+
+
+@pytest.mark.asyncio
+async def test_concurrent_same_http_command_executes_once_and_waits_for_completion(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    journal = HttpCommandJournal(sessions)
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    calls = 0
+
+    async def command() -> tuple[int, dict[str, Any]]:
+        nonlocal calls
+        calls += 1
+        entered.set()
+        await release.wait()
+        return 200, {"success": True}
+
+    first = asyncio.create_task(
+        journal.execute(
+            principal=Principal("subject"),
+            route_family="script-build:concurrent",
+            idempotency_key="one",
+            request_payload={"build": 1},
+            command=command,
+        )
+    )
+    await entered.wait()
+    second = asyncio.create_task(
+        journal.execute(
+            principal=Principal("subject"),
+            route_family="script-build:concurrent",
+            idempotency_key="one",
+            request_payload={"build": 1},
+            command=command,
+        )
+    )
+    release.set()
+    assert await first == await second == (200, {"success": True})
+    assert calls == 1
+
+
+@pytest.mark.asyncio
+async def test_failed_http_command_is_not_blindly_replayed(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    journal = HttpCommandJournal(sessions)
+
+    async def command() -> tuple[int, dict[str, Any]]:
+        raise RuntimeError("lost acknowledgement")
+
+    values = dict(
+        principal=Principal("subject"),
+        route_family="script-build:failed",
+        idempotency_key="one",
+        request_payload={"build": 1},
+        command=command,
+    )
+    with pytest.raises(RuntimeError, match="lost acknowledgement"):
+        await journal.execute(**values)
+    with pytest.raises(ProtocolViolation, match="reconciliation"):
+        await journal.execute(**values)
+
+
+@pytest.mark.asyncio
+async def test_reserved_http_command_resumes_preallocated_identity(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    principal = Principal("subject", "tenant")
+    route = "script-build:start"
+    key = "crash-window"
+    payload = {"topic_id": 3}
+    scope = canonical_sha256(
+        {"subject": principal.subject, "tenant_id": principal.tenant_id}
+    ).hex_value
+    fingerprint = canonical_sha256({"route_family": route, "request": payload}).hex_value
+    now = datetime.now(UTC)
+    async with sessions() as session, session.begin():
+        await session.execute(
+            insert(http_command_table).values(
+                principal_scope_sha256=scope,
+                route_family=route,
+                idempotency_key=key,
+                request_fingerprint=fingerprint,
+                preallocated_root_trace_id="root-preallocated",
+                resource_id=41,
+                state="reserved",
+                created_at=now,
+                updated_at=now,
+            )
+        )
+
+    async def command() -> tuple[int, dict[str, Any]]:
+        raise AssertionError("reserved command must not be blindly replayed")
+
+    async def resume(root: str | None, resource: int | None) -> tuple[int, dict[str, Any]]:
+        assert (root, resource) == ("root-preallocated", 41)
+        return 200, {"script_build_id": 41, "root_trace_id": root}
+
+    result = await HttpCommandJournal(sessions).execute(
+        principal=principal,
+        route_family=route,
+        idempotency_key=key,
+        request_payload=payload,
+        command=command,
+        resume_reserved=resume,
+    )
+    assert result == (
+        200,
+        {"script_build_id": 41, "root_trace_id": "root-preallocated"},
+    )
+    async with sessions() as session:
+        assert (
+            await session.scalar(
+                select(http_command_table.c.state).where(
+                    http_command_table.c.idempotency_key == key
+                )
+            )
+            == "completed"
+        )
+
+
+@pytest.mark.asyncio
+async def test_oversized_idempotent_response_is_failed_not_left_reserved(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+
+    async def command() -> tuple[int, dict[str, Any]]:
+        return 200, {"value": "x" * (65 * 1024)}
+
+    with pytest.raises(ProtocolViolation, match="exceeds"):
+        await HttpCommandJournal(sessions).execute(
+            principal=Principal("subject"),
+            route_family="script-build:oversized",
+            idempotency_key="oversized",
+            request_payload={},
+            command=command,
+        )
+    async with sessions() as session:
+        assert (
+            await session.scalar(
+                select(http_command_table.c.state).where(
+                    http_command_table.c.idempotency_key == "oversized"
+                )
+            )
+            == "failed"
+        )

script_build_host/tests/test_migrations.py

@@ -8,7 +8,9 @@ from pathlib import Path
 import pytest
 from alembic import command
 from alembic.config import Config
+from alembic.script import ScriptDirectory
 from sqlalchemy import create_engine, inspect, text
+from sqlalchemy.exc import DBAPIError
 from sqlalchemy.ext.asyncio import create_async_engine
 
 
@@ -42,6 +44,12 @@ def test_migration_checksum_manifest() -> None:
         assert hashlib.sha256(path.read_bytes()).hexdigest() == expected
 
 
+def test_revision_identifiers_fit_alembic_mysql_version_column() -> None:
+    root = Path(__file__).parents[1]
+    scripts = ScriptDirectory.from_config(Config(str(root / "alembic.ini")))
+    assert all(len(revision.revision) <= 32 for revision in scripts.walk_revisions())
+
+
 def test_mysql_offline_sql_compiles_without_credentials(monkeypatch: object) -> None:
     root = Path(__file__).parents[1]
     output = StringIO()
@@ -51,11 +59,33 @@ def test_mysql_offline_sql_compiles_without_credentials(monkeypatch: object) ->
     config = Config(str(root / "alembic.ini"), output_buffer=output)
     command.upgrade(config, "head", sql=True)
     sql = output.getvalue()
-    assert sql.count("CREATE TABLE script_build_") == 4
-    assert "script_build_paragraph" not in sql
-    assert "script_build_element" not in sql
+    assert sql.count("CREATE TABLE script_build_") == 5
+    assert "CREATE TABLE script_build_paragraph" not in sql
+    assert "CREATE TABLE script_build_element" not in sql
     assert "DROP CHECK ck_artifact_phase_one_type" in sql
     assert "ADD CONSTRAINT ck_artifact_business_type CHECK" in sql
+    assert "'root_delivery_manifest'" in sql
+    assert "CREATE TABLE script_build_http_command" in sql
+    assert "CONSTRAINT uq_publication_build_type UNIQUE" in sql
+    assert "owner_epoch BIGINT NOT NULL" in sql
+    assert "owner_acquired_at DATETIME(6)" in sql
+    assert sql.count("WITH CASCADED CHECK OPTION") == 4
+    assert "uq_script_build_record_reson_trace" in sql
+
+
+def test_phase_three_offline_downgrade_fails_closed(monkeypatch: object) -> None:
+    root = Path(__file__).parents[1]
+    output = StringIO()
+    monkeypatch.setenv(  # type: ignore[attr-defined]
+        "SCRIPT_BUILD_WRITE_DATABASE_URL", "mysql+asyncmy://"
+    )
+    config = Config(str(root / "alembic.ini"), output_buffer=output)
+    with pytest.raises(RuntimeError, match="online safety preflight"):
+        command.downgrade(
+            config,
+            "0003_phase_three_publication:0002_phase_two_artifacts",
+            sql=True,
+        )
 
 
 def test_phase_two_artifacts_make_downgrade_refuse_data_loss(
@@ -97,3 +127,183 @@ async def test_explicit_mysql_dsn_is_reachable() -> None:
             assert await connection.scalar(text("SELECT 1")) == 1
     finally:
         await engine.dispose()
+
+
+@pytest.mark.mysql
+@pytest.mark.asyncio
+async def test_mysql_phase_three_schema_semantics() -> None:
+    dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_DSN")
+    if not dsn:
+        pytest.skip("SCRIPT_BUILD_TEST_MYSQL_DSN is not configured")
+    engine = create_async_engine(dsn, pool_pre_ping=True, isolation_level="READ COMMITTED")
+    try:
+        async with engine.connect() as connection:
+            version = str(await connection.scalar(text("SELECT VERSION()"))).split("-", 1)[0]
+            assert tuple(int(part) for part in version.split(".")[:3]) >= (8, 0, 16)
+            assert await connection.scalar(text("SELECT @@transaction_isolation")) == (
+                "READ-COMMITTED"
+            )
+            assert await connection.scalar(text("SELECT version_num FROM alembic_version")) == (
+                "0003_phase_three_publication"
+            )
+            precision = await connection.scalar(
+                text(
+                    "SELECT DATETIME_PRECISION FROM information_schema.COLUMNS "
+                    "WHERE TABLE_SCHEMA=DATABASE() "
+                    "AND TABLE_NAME='script_build_mission_binding' "
+                    "AND COLUMN_NAME='owner_acquired_at'"
+                )
+            )
+            assert precision == 6
+            views = set(
+                (
+                    await connection.execute(
+                        text(
+                            "SELECT TABLE_NAME FROM information_schema.VIEWS "
+                            "WHERE TABLE_SCHEMA=DATABASE()"
+                        )
+                    )
+                ).scalars()
+            )
+            assert {
+                "script_build_candidate_paragraph_v",
+                "script_build_candidate_element_v",
+                "script_build_candidate_paragraph_element_v",
+                "script_build_runtime_record_v",
+            } <= views
+
+            transaction = await connection.begin_nested()
+            try:
+                await connection.execute(
+                    text(
+                        "INSERT INTO script_build_candidate_paragraph_v "
+                        "(script_build_id, branch_id, paragraph_index, level, is_active) "
+                        "VALUES (900000, 1, 1, 1, 1)"
+                    )
+                )
+                with pytest.raises(DBAPIError):
+                    await connection.execute(
+                        text(
+                            "INSERT INTO script_build_candidate_paragraph_v "
+                            "(script_build_id, branch_id, paragraph_index, level, is_active) "
+                            "VALUES (900000, 0, 2, 1, 1)"
+                        )
+                    )
+                await connection.execute(
+                    text(
+                        "INSERT INTO script_build_runtime_record_v "
+                        "(id, execution_id, topic_build_id, topic_id, status, reson_trace_id, "
+                        "is_deleted, is_favorited) VALUES "
+                        "(900000, 1, 1, 1, 'running', 'mysql-gate-runtime', 0, 0)"
+                    )
+                )
+                with pytest.raises(DBAPIError):
+                    await connection.execute(
+                        text(
+                            "UPDATE script_build_runtime_record_v SET status='success' "
+                            "WHERE id=900000"
+                        )
+                    )
+                with pytest.raises(DBAPIError):
+                    await connection.execute(
+                        text(
+                            "INSERT INTO script_build_http_command "
+                            "(principal_scope_sha256, route_family, idempotency_key, "
+                            "request_fingerprint, state, created_at, updated_at) VALUES "
+                            "(:digest, 'gate', 'bad-state', :digest, 'invalid', NOW(6), NOW(6))"
+                        ),
+                        {"digest": "a" * 64},
+                    )
+            finally:
+                await transaction.rollback()
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.mysql
+@pytest.mark.asyncio
+async def test_mysql_phase_three_least_privilege_accounts() -> None:
+    read_dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_READ_DSN")
+    runtime_dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_RUNTIME_DSN")
+    final_dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_FINAL_DSN")
+    if not all((read_dsn, runtime_dsn, final_dsn)):
+        pytest.skip("the three least-privilege MySQL DSNs are not configured")
+    engines = [
+        create_async_engine(str(read_dsn), isolation_level="READ COMMITTED"),
+        create_async_engine(str(runtime_dsn), isolation_level="READ COMMITTED"),
+        create_async_engine(str(final_dsn), isolation_level="READ COMMITTED"),
+    ]
+    try:
+        async with engines[0].connect() as connection:
+            with pytest.raises(DBAPIError):
+                await connection.execute(
+                    text(
+                        "INSERT INTO script_build_http_command "
+                        "(principal_scope_sha256, route_family, idempotency_key, "
+                        "request_fingerprint, state, created_at, updated_at) VALUES "
+                        "(:digest, 'grant-gate', 'read', :digest, 'reserved', NOW(6), NOW(6))"
+                    ),
+                    {"digest": "b" * 64},
+                )
+
+        async with engines[1].connect() as connection:
+            transaction = await connection.begin()
+            try:
+                with pytest.raises(DBAPIError):
+                    await connection.execute(
+                        text(
+                            "INSERT INTO script_build_paragraph "
+                            "(script_build_id, branch_id, paragraph_index, level, is_active) "
+                            "VALUES (930001, 0, 1, 1, 1)"
+                        )
+                    )
+                await connection.execute(
+                    text(
+                        "INSERT INTO script_build_candidate_paragraph_v "
+                        "(script_build_id, branch_id, paragraph_index, level, is_active) "
+                        "VALUES (930001, 1, 1, 1, 1)"
+                    )
+                )
+                await connection.execute(
+                    text(
+                        "INSERT INTO script_build_runtime_record_v "
+                        "(id, execution_id, topic_build_id, topic_id, status, reson_trace_id, "
+                        "is_deleted, is_favorited) VALUES "
+                        "(930001, 1, 1, 1, 'running', 'least-privilege-runtime', 0, 0)"
+                    )
+                )
+                with pytest.raises(DBAPIError):
+                    await connection.execute(
+                        text(
+                            "UPDATE script_build_runtime_record_v SET status='success' "
+                            "WHERE id=930001"
+                        )
+                    )
+            finally:
+                await transaction.rollback()
+
+        async with engines[2].connect() as connection:
+            transaction = await connection.begin()
+            try:
+                await connection.execute(
+                    text(
+                        "INSERT INTO script_build_paragraph "
+                        "(script_build_id, branch_id, paragraph_index, level, is_active) "
+                        "VALUES (930002, 0, 1, 1, 1)"
+                    )
+                )
+                with pytest.raises(DBAPIError):
+                    await connection.execute(
+                        text(
+                            "INSERT INTO script_build_http_command "
+                            "(principal_scope_sha256, route_family, idempotency_key, "
+                            "request_fingerprint, state, created_at, updated_at) VALUES "
+                            "(:digest, 'grant-gate', 'final', :digest, 'reserved', NOW(6), NOW(6))"
+                        ),
+                        {"digest": "c" * 64},
+                    )
+            finally:
+                await transaction.rollback()
+    finally:
+        for engine in engines:
+            await engine.dispose()

script_build_host/tests/test_phase_three_contracts.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+import pytest
+
+from script_build_host.domain.errors import (
+    LegacyCanonicalIdentityConflict,
+    ProtocolViolation,
+)
+from script_build_host.domain.phase_three_artifacts import (
+    RootDeliveryManifestV1,
+    canonicalize_legacy_projection,
+)
+from script_build_host.domain.phase_two_artifacts import (
+    ScriptElementV1,
+    ScriptParagraphElementLinkV1,
+    ScriptParagraphV1,
+    StructuredScriptArtifactV1,
+)
+
+_DIGEST = "sha256:" + "a" * 64
+
+
+def _script(
+    *,
+    root_id: int = 10,
+    child_id: int = 20,
+    element_id: int = 30,
+    child_description: str = "child body",
+) -> StructuredScriptArtifactV1:
+    return StructuredScriptArtifactV1(
+        direction_ref="script-build://artifact-versions/1",
+        input_closure_digest=_DIGEST,
+        paragraphs=(
+            ScriptParagraphV1(
+                paragraph_id=root_id,
+                paragraph_index=1,
+                level=1,
+                parent_id=None,
+                name="root",
+                content_range={"start": 0, "end": 10},
+                description="root body",
+            ),
+            ScriptParagraphV1(
+                paragraph_id=child_id,
+                paragraph_index=2,
+                level=2,
+                parent_id=root_id,
+                name="child",
+                content_range={"start": 10, "end": 20},
+                description=child_description,
+            ),
+        ),
+        elements=(
+            ScriptElementV1(
+                element_id=element_id,
+                name="hook",
+                dimension_primary="形式",
+                dimension_secondary="opening",
+                topic_support={"score": 1},
+            ),
+        ),
+        paragraph_element_links=(
+            ScriptParagraphElementLinkV1(
+                paragraph_id=child_id,
+                element_id=element_id,
+            ),
+        ),
+        source_artifact_refs=(),
+        evidence_refs=(),
+        acceptance_notes=(),
+    )
+
+
+def test_legacy_canonical_digest_ignores_every_physical_identity() -> None:
+    first = canonicalize_legacy_projection(_script(), direction="# D", summary="summary")
+    second = canonicalize_legacy_projection(
+        _script(root_id=101, child_id=202, element_id=303),
+        direction="# D",
+        summary="summary",
+    )
+
+    assert first.content_payload() == second.content_payload()
+    assert first.canonical_sha256 == second.canonical_sha256
+
+
+def test_legacy_canonical_digest_changes_with_business_content() -> None:
+    first = canonicalize_legacy_projection(_script(), direction="# D", summary="summary")
+    changed_body = canonicalize_legacy_projection(
+        _script(child_description="changed"), direction="# D", summary="summary"
+    )
+    changed_direction = canonicalize_legacy_projection(
+        _script(), direction="# other", summary="summary"
+    )
+
+    assert first.canonical_sha256 != changed_body.canonical_sha256
+    assert first.canonical_sha256 != changed_direction.canonical_sha256
+
+
+def test_legacy_canonical_rejects_parent_cycles_and_duplicate_natural_elements() -> None:
+    cycle = replace(
+        _script(),
+        paragraphs=(
+            ScriptParagraphV1(10, 1, 2, 20, "a", {}),
+            ScriptParagraphV1(20, 2, 2, 10, "b", {}),
+        ),
+    )
+    with pytest.raises(LegacyCanonicalIdentityConflict, match="cycle"):
+        canonicalize_legacy_projection(cycle)
+
+    duplicate_elements = replace(
+        _script(),
+        elements=(
+            ScriptElementV1(30, "hook", "形式", "opening"),
+            ScriptElementV1(31, "hook", "形式", "opening"),
+        ),
+        paragraph_element_links=(),
+    )
+    with pytest.raises(LegacyCanonicalIdentityConflict, match="duplicated"):
+        canonicalize_legacy_projection(duplicate_elements)
+
+
+def test_root_delivery_manifest_is_strict_bounded_and_publishable() -> None:
+    manifest = RootDeliveryManifestV1.from_payload(
+        {
+            "schema_version": "root-delivery-manifest/v1",
+            "direction_ref": "script-build://artifact-versions/1",
+            "candidate_portfolio_ref": "script-build://artifact-versions/2",
+            "structured_script_ref": "script-build://artifact-versions/3",
+            "input_closure_digest": _DIGEST,
+            "legacy_projection_digest": _DIGEST,
+            "build_summary": "ok",
+            "blocking_defects": [],
+        }
+    ).with_digest()
+    manifest.require_publishable()
+    assert manifest.canonical_sha256.startswith("sha256:")
+
+    with pytest.raises(ProtocolViolation, match="unknown fields"):
+        RootDeliveryManifestV1.from_payload({**manifest.content_payload(), "unknown": True})
+    with pytest.raises(ProtocolViolation, match="blocking defects"):
+        replace(
+            manifest, blocking_defects=("not ready",), canonical_sha256=""
+        ).require_publishable()
+    with pytest.raises(ProtocolViolation, match="2,000"):
+        replace(manifest, build_summary="x" * 2_001, canonical_sha256="")

script_build_host/tests/test_phase_three_ownership.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+
+from script_build_host.domain.errors import (
+    MissionAlreadyOwned,
+    MissionFencingTokenStale,
+    MissionStopRequested,
+)
+from script_build_host.domain.records import BuildStatus
+from script_build_host.infrastructure.ownership import FencedCommandGate, OwnerLease
+from script_build_host.repositories.legacy_state import SqlAlchemyLegacyBuildStateRepository
+from script_build_host.repositories.sqlalchemy import SqlAlchemyMissionBindingRepository
+
+
+async def _bound_build(sessions: async_sessionmaker[AsyncSession]) -> int:
+    state = SqlAlchemyLegacyBuildStateRepository(sessions)
+    build_id = await state.create(
+        execution_id=1,
+        topic_build_id=2,
+        topic_id=3,
+        agent_type="AigcAgent",
+        agent_config={},
+        data_source_url=None,
+        strategies_config={},
+        root_trace_id="root-owner-test",
+    )
+    await SqlAlchemyMissionBindingRepository(sessions).create(
+        script_build_id=build_id,
+        root_trace_id="root-owner-test",
+        input_snapshot_id=1,
+        engine_version="test",
+        schema_version="test/v1",
+    )
+    return build_id
+
+
+@pytest.mark.asyncio
+async def test_owner_lease_excludes_second_owner_and_grows_epoch(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], tmp_path: Path
+) -> None:
+    _, sessions = database
+    build_id = await _bound_build(sessions)
+    first = OwnerLease(sessions, tmp_path / "leases", owner_instance_id="owner-a")
+    second = OwnerLease(sessions, tmp_path / "leases", owner_instance_id="owner-b")
+
+    first_token = await first.acquire(build_id)
+    with pytest.raises(MissionAlreadyOwned):
+        await second.acquire(build_id)
+    await first.release()
+
+    second_token = await second.acquire(build_id)
+    assert second_token.owner_epoch == first_token.owner_epoch + 1
+    with pytest.raises(MissionFencingTokenStale):
+        await FencedCommandGate(sessions).verify(first_token)
+    await second.release()
+
+
+@pytest.mark.asyncio
+async def test_stop_epoch_fences_current_owner_without_taking_process_lease(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], tmp_path: Path
+) -> None:
+    _, sessions = database
+    build_id = await _bound_build(sessions)
+    lease = OwnerLease(sessions, tmp_path / "leases", owner_instance_id="owner")
+    token = await lease.acquire(build_id)
+    gate = FencedCommandGate(sessions)
+    await gate.verify(token)
+
+    assert await gate.request_stop(build_id) == token.stop_epoch + 1
+    with pytest.raises(MissionStopRequested):
+        await gate.verify(token)
+    assert (
+        await SqlAlchemyLegacyBuildStateRepository(sessions).get_status(build_id)
+        is BuildStatus.STOPPING
+    )
+    await lease.release()

script_build_host/tests/test_phase_three_publication.py

@@ -0,0 +1,393 @@
+from __future__ import annotations
+
+import os
+from dataclasses import replace
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+import pytest
+from sqlalchemy import insert, select, text
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+
+from script_build_host.application.legacy_projection import LegacyDetailProjectionService
+from script_build_host.domain.artifacts import (
+    ArtifactState,
+    DirectionGoal,
+    ScriptDirectionArtifactV1,
+)
+from script_build_host.domain.errors import PublicationReadbackMismatch
+from script_build_host.domain.phase_three_artifacts import (
+    LegacyProjectionCanonicalV1,
+    RootDeliveryManifestV1,
+    canonicalize_legacy_projection,
+)
+from script_build_host.domain.phase_two_artifacts import (
+    ScriptElementV1,
+    ScriptParagraphElementLinkV1,
+    ScriptParagraphV1,
+    StructuredScriptArtifactV1,
+)
+from script_build_host.domain.records import BuildStatus, PublicationState, PublicationType
+from script_build_host.infrastructure.final_publication import (
+    SqlAlchemyFinalPublicationUnitOfWork,
+)
+from script_build_host.infrastructure.legacy_tables import (
+    script_build_element,
+    script_build_paragraph,
+    script_build_record,
+)
+from script_build_host.infrastructure.ownership import FencedCommandGate, OwnerLease
+from script_build_host.infrastructure.tables import (
+    artifact_version_table,
+    mission_binding_table,
+    publication_table,
+)
+from script_build_host.repositories.legacy_state import SqlAlchemyLegacyBuildStateRepository
+from script_build_host.repositories.sqlalchemy import (
+    SqlAlchemyMissionBindingRepository,
+    SqlAlchemyPublicationRepository,
+    SqlAlchemyScriptBusinessArtifactRepository,
+)
+
+_CLOSURE = "sha256:" + "b" * 64
+
+
+class _Authorizer:
+    async def require_access(self, principal: Any, script_build_id: int) -> None:
+        return None
+
+
+class _MismatchingProjection(LegacyDetailProjectionService):
+    def to_canonical(self, detail: Any) -> LegacyProjectionCanonicalV1:
+        return replace(super().to_canonical(detail), summary="tampered")
+
+
+def _structured(direction_ref: str) -> StructuredScriptArtifactV1:
+    return StructuredScriptArtifactV1(
+        direction_ref=direction_ref,
+        input_closure_digest=_CLOSURE,
+        paragraphs=(
+            ScriptParagraphV1(10, 1, 1, None, "opening", {"start": 0, "end": 4}),
+            ScriptParagraphV1(20, 2, 2, 10, "detail", {"start": 4, "end": 8}),
+        ),
+        elements=(ScriptElementV1(30, "hook", "形式", "opening"),),
+        paragraph_element_links=(ScriptParagraphElementLinkV1(20, 30),),
+        source_artifact_refs=(),
+        evidence_refs=(),
+        acceptance_notes=(),
+    )
+
+
+async def _publication_fixture(
+    sessions: async_sessionmaker[AsyncSession],
+    tmp_path: Path,
+    *,
+    publication_sessions: async_sessionmaker[AsyncSession] | None = None,
+    branch_sessions: async_sessionmaker[AsyncSession] | None = None,
+    prepare_publication: bool = True,
+) -> dict[str, Any]:
+    state = SqlAlchemyLegacyBuildStateRepository(sessions)
+    root_trace_id = f"root-publication-{uuid4()}"
+    build_id = await state.create(
+        execution_id=1,
+        topic_build_id=2,
+        topic_id=3,
+        agent_type="AigcAgent",
+        agent_config={},
+        data_source_url=None,
+        strategies_config={},
+        root_trace_id=root_trace_id,
+    )
+    await SqlAlchemyMissionBindingRepository(sessions).create(
+        script_build_id=build_id,
+        root_trace_id=root_trace_id,
+        input_snapshot_id=1,
+        engine_version="test",
+        schema_version="test/v1",
+    )
+    now = datetime.now(UTC)
+    async with (branch_sessions or sessions)() as session, session.begin():
+        await session.execute(
+            insert(script_build_paragraph).values(
+                script_build_id=build_id,
+                branch_id=0,
+                paragraph_index=99,
+                level=1,
+                name="old",
+                content_range={},
+                is_active=True,
+                created_at=now,
+                updated_at=now,
+            )
+        )
+        await session.execute(
+            insert(script_build_element).values(
+                script_build_id=build_id,
+                branch_id=0,
+                name="old",
+                dimension_primary="形式",
+                dimension_secondary="old",
+                is_active=True,
+                created_at=now,
+                updated_at=now,
+            )
+        )
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    direction, _ = await artifacts.freeze(
+        script_build_id=build_id,
+        task_id="direction",
+        attempt_id=f"direction-{build_id}",
+        spec_version=1,
+        artifact=ScriptDirectionArtifactV1(
+            goals=(DirectionGoal("goal", "deliver", "source"),),
+            evidence_refs=("script-build://artifact-versions/777",),
+            legacy_markdown="# Direction",
+        ),
+    )
+    script = _structured(f"script-build://artifact-versions/{direction.artifact_version_id}")
+    structured, _ = await artifacts.freeze(
+        script_build_id=build_id,
+        task_id="structured",
+        attempt_id=f"structured-{build_id}",
+        spec_version=1,
+        artifact=script,
+    )
+    canonical = canonicalize_legacy_projection(
+        script, direction="# Direction", summary="final summary"
+    )
+    manifest_value = RootDeliveryManifestV1(
+        direction_ref=f"script-build://artifact-versions/{direction.artifact_version_id}",
+        candidate_portfolio_ref="script-build://artifact-versions/999",
+        structured_script_ref=(
+            f"script-build://artifact-versions/{structured.artifact_version_id}"
+        ),
+        input_closure_digest=_CLOSURE,
+        legacy_projection_digest=canonical.canonical_sha256,
+        build_summary="final summary",
+    )
+    manifest, _ = await artifacts.freeze(
+        script_build_id=build_id,
+        task_id="root",
+        attempt_id=f"root-{build_id}",
+        spec_version=1,
+        artifact=manifest_value,
+    )
+    publication = None
+    if prepare_publication:
+        publication = await SqlAlchemyPublicationRepository(
+            publication_sessions or sessions
+        ).prepare(
+            script_build_id=build_id,
+            publication_type=PublicationType.FINAL,
+            accept_decision_id=f"accept-{build_id}",
+            artifact_version_id=manifest.artifact_version_id,
+            expected_sha256=manifest.canonical_sha256,
+        )
+    lease = OwnerLease(sessions, tmp_path / "leases", owner_instance_id=f"owner-{build_id}")
+    token = await lease.acquire(build_id)
+    return {
+        "build_id": build_id,
+        "script": script,
+        "structured": structured,
+        "manifest": manifest,
+        "manifest_value": manifest.artifact,
+        "publication": publication,
+        "lease": lease,
+        "token": token,
+    }
+
+
+@pytest.mark.asyncio
+async def test_final_publication_commits_projection_pointer_artifacts_and_success(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], tmp_path: Path
+) -> None:
+    _, sessions = database
+    values = await _publication_fixture(sessions, tmp_path)
+    projection = LegacyDetailProjectionService(sessions, _Authorizer())  # type: ignore[arg-type]
+    uow = SqlAlchemyFinalPublicationUnitOfWork(sessions, projection, FencedCommandGate(sessions))
+    try:
+        result = await uow.publish(
+            script_build_id=values["build_id"],
+            publication_id=values["publication"].publication_id,
+            manifest_version=values["manifest"],
+            structured_version=values["structured"],
+            manifest=values["manifest_value"],
+            structured_script=values["script"],
+            direction_markdown="# Direction",
+            owner_token=values["token"],
+        )
+        assert result.committed is True
+        replay = await uow.publish(
+            script_build_id=values["build_id"],
+            publication_id=values["publication"].publication_id,
+            manifest_version=values["manifest"],
+            structured_version=values["structured"],
+            manifest=values["manifest_value"],
+            structured_script=values["script"],
+            direction_markdown="# Direction",
+            owner_token=values["token"],
+        )
+        assert replay.committed is True
+        async with sessions() as session:
+            assert (
+                await session.scalar(
+                    select(script_build_record.c.status).where(
+                        script_build_record.c.id == values["build_id"]
+                    )
+                )
+                == BuildStatus.SUCCESS.value
+            )
+            assert (
+                await session.scalar(
+                    select(mission_binding_table.c.accepted_root_artifact_version_id).where(
+                        mission_binding_table.c.script_build_id == values["build_id"]
+                    )
+                )
+                == values["manifest"].artifact_version_id
+            )
+            assert set(
+                (
+                    await session.execute(
+                        select(artifact_version_table.c.state).where(
+                            artifact_version_table.c.id.in_(
+                                [
+                                    values["manifest"].artifact_version_id,
+                                    values["structured"].artifact_version_id,
+                                ]
+                            )
+                        )
+                    )
+                ).scalars()
+            ) == {ArtifactState.PUBLISHED.value}
+    finally:
+        await values["lease"].release()
+
+
+@pytest.mark.asyncio
+async def test_final_publication_readback_failure_rolls_back_every_final_write(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], tmp_path: Path
+) -> None:
+    _, sessions = database
+    values = await _publication_fixture(sessions, tmp_path)
+    projection = _MismatchingProjection(sessions, _Authorizer())  # type: ignore[arg-type]
+    uow = SqlAlchemyFinalPublicationUnitOfWork(sessions, projection, FencedCommandGate(sessions))
+    try:
+        with pytest.raises(PublicationReadbackMismatch):
+            await uow.publish(
+                script_build_id=values["build_id"],
+                publication_id=values["publication"].publication_id,
+                manifest_version=values["manifest"],
+                structured_version=values["structured"],
+                manifest=values["manifest_value"],
+                structured_script=values["script"],
+                direction_markdown="# Direction",
+                owner_token=values["token"],
+            )
+        async with sessions() as session:
+            assert (
+                await session.scalar(
+                    select(script_build_record.c.status).where(
+                        script_build_record.c.id == values["build_id"]
+                    )
+                )
+                == BuildStatus.RUNNING.value
+            )
+            assert (
+                await session.scalar(
+                    select(script_build_paragraph.c.name).where(
+                        script_build_paragraph.c.script_build_id == values["build_id"],
+                        script_build_paragraph.c.branch_id == 0,
+                    )
+                )
+                == "old"
+            )
+            assert (
+                await session.scalar(
+                    select(mission_binding_table.c.accepted_root_artifact_version_id).where(
+                        mission_binding_table.c.script_build_id == values["build_id"]
+                    )
+                )
+                is None
+            )
+            assert (
+                await session.scalar(
+                    select(publication_table.c.state).where(
+                        publication_table.c.id == values["publication"].publication_id
+                    )
+                )
+                == PublicationState.PENDING.value
+            )
+    finally:
+        await values["lease"].release()
+
+
+@pytest.mark.mysql
+@pytest.mark.asyncio
+async def test_mysql_final_publication_commits_one_atomic_projection(tmp_path: Path) -> None:
+    runtime_dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_RUNTIME_DSN") or os.environ.get(
+        "SCRIPT_BUILD_TEST_MYSQL_DSN"
+    )
+    final_dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_FINAL_DSN") or runtime_dsn
+    if not runtime_dsn or not final_dsn:
+        pytest.skip("SCRIPT_BUILD_TEST_MYSQL_DSN is not configured")
+    runtime_engine = create_async_engine(
+        runtime_dsn, pool_pre_ping=True, isolation_level="READ COMMITTED"
+    )
+    final_engine = create_async_engine(
+        final_dsn, pool_pre_ping=True, isolation_level="READ COMMITTED"
+    )
+    sessions = async_sessionmaker(runtime_engine, expire_on_commit=False)
+    final_sessions = async_sessionmaker(final_engine, expire_on_commit=False)
+    values = await _publication_fixture(
+        sessions,
+        tmp_path,
+        publication_sessions=final_sessions,
+        branch_sessions=final_sessions,
+        prepare_publication=False,
+    )
+    projection = LegacyDetailProjectionService(final_sessions, _Authorizer())  # type: ignore[arg-type]
+    uow = SqlAlchemyFinalPublicationUnitOfWork(
+        final_sessions, projection, FencedCommandGate(final_sessions)
+    )
+    try:
+        publication = await uow.reserve(
+            script_build_id=values["build_id"],
+            accept_decision_id=f"accept-{values['build_id']}",
+            artifact_version_id=values["manifest"].artifact_version_id,
+            expected_sha256=values["manifest"].canonical_sha256,
+            owner_token=values["token"],
+        )
+        result = await uow.publish(
+            script_build_id=values["build_id"],
+            publication_id=publication.publication_id,
+            manifest_version=values["manifest"],
+            structured_version=values["structured"],
+            manifest=values["manifest_value"],
+            structured_script=values["script"],
+            direction_markdown="# Direction",
+            owner_token=values["token"],
+        )
+        assert result.committed is True
+        async with final_sessions() as session:
+            assert await session.scalar(text("SELECT @@transaction_isolation")) == (
+                "READ-COMMITTED"
+            )
+            assert (
+                await session.scalar(
+                    select(script_build_record.c.status).where(
+                        script_build_record.c.id == values["build_id"]
+                    )
+                )
+                == BuildStatus.SUCCESS.value
+            )
+    finally:
+        await values["lease"].release()
+        await runtime_engine.dispose()
+        await final_engine.dispose()

script_build_host/tests/test_phase_two_real_runner_e2e.py

@@ -14,9 +14,11 @@ from sqlalchemy import select
 
 from script_build_host.agents.prompts import (
     phase_one_prompt_manifest,
+    phase_three_prompt_manifest,
     script_build_prompt_manifest,
 )
 from script_build_host.application.input_snapshot_service import ScriptInputSnapshotService
+from script_build_host.application.legacy_projection import LegacyDetailProjectionService
 from script_build_host.application.mission_service import PHASE_ONE_CAPABILITY_BOUNDARY
 from script_build_host.application.phase_two_planning import (
     PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
@@ -26,14 +28,19 @@ from script_build_host.domain.artifacts import ArtifactKind
 from script_build_host.domain.input_snapshot import ScriptBuildInput
 from script_build_host.domain.records import BuildStatus, Principal
 from script_build_host.infrastructure.canonical_json import canonical_sha256
+from script_build_host.infrastructure.final_publication import (
+    SqlAlchemyFinalPublicationUnitOfWork,
+)
 from script_build_host.infrastructure.legacy_tables import (
     script_build_paragraph,
     script_build_record,
 )
+from script_build_host.infrastructure.ownership import FencedCommandGate, OwnerLease
 from script_build_host.infrastructure.tables import (
     artifact_version_table,
     input_snapshot_table,
     mission_binding_table,
+    publication_table,
 )
 from script_build_host.infrastructure.task_contract_store import FileScriptTaskContractStore
 from script_build_host.repositories.legacy_state import SqlAlchemyLegacyBuildStateRepository
@@ -871,6 +878,55 @@ class _PhaseTwoScriptedLLM:
         )
 
 
+class _PhaseThreeScriptedLLM(_PhaseTwoScriptedLLM):
+    """Continue the real Phase2 runner through the protected Root delivery."""
+
+    def __init__(self, store: FileSystemTaskStore) -> None:
+        super().__init__(store)
+        self.phase_three_active = False
+
+    async def __call__(self, messages, tools, **kwargs):
+        self.phase_three_active = self.phase_three_active or any(
+            "script-build-phase-three/v1" in str(item.get("content", "")) for item in messages
+        )
+        names = {item["function"]["name"] for item in tools or []}
+        if self.phase_three_active and "plan_script_tasks" in names:
+            return await self._phase_three_planner()
+        return await super().__call__(messages, tools, **kwargs)
+
+    async def _phase_three_planner(self) -> dict[str, Any]:
+        ledger = await self.store.load(ROOT)
+        root = ledger.tasks[ledger.root_task_id]
+        assert _task_kind(root) == "root-delivery"
+        if root.status in {TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN}:
+            return self._call("dispatch_script_tasks", {"task_ids": [root.task_id]})
+        if root.status is TaskStatus.AWAITING_DECISION:
+            return self._accept(root, ledger)
+        if root.status is TaskStatus.COMPLETED:
+            return {"content": "Root delivery accepted", "finish_reason": "stop"}
+        raise AssertionError(f"unexpected Root delivery status {root.status}")
+
+    def _worker(self, messages, names: set[str]) -> dict[str, Any]:
+        prompt = _role_prompt(messages)
+        refs = cast(Sequence[str], cast(Mapping[str, Any], prompt["task_spec"])["context_refs"])
+        kind = next(item.rsplit("/", 1)[-1] for item in refs if "/task-kinds/" in item)
+        if kind != "root-delivery":
+            return super()._worker(messages, names)
+        if kind not in self.worker_kinds:
+            self.worker_kinds.append(kind)
+        last = _last_tool_value(messages)
+        if isinstance(last, dict) and "artifact_ref" in last:
+            return self._call("submit_attempt", {})
+        if isinstance(last, dict) and "legacy_projection_digest" in last:
+            return self._call(
+                "save_root_delivery_manifest",
+                {"build_summary": "final summary", "blocking_defects": []},
+            )
+        if isinstance(last, dict) and "direction_ref" in last:
+            return self._call("legacy_projection_dry_run", {"build_summary": "final summary"})
+        return self._call("read_accepted_portfolio", {})
+
+
 class _OrderedExplorationScriptedLLM(_PhaseTwoScriptedLLM):
     """Close Phase2 after exercising each creative preset in a chosen order."""
 
@@ -1285,7 +1341,7 @@ class _PromptCatalog:
     def __init__(self) -> None:
         self._entries = {
             (str(item["preset"]), str(item["role"])): dict(item)
-            for item in script_build_prompt_manifest()
+            for item in (*script_build_prompt_manifest(), *phase_three_prompt_manifest())
         }
 
     async def load(self, requests: Sequence[Any]) -> tuple[dict[str, Any], ...]:
@@ -2008,6 +2064,172 @@ async def test_historical_partial_advances_via_http_and_reenters_after_filestore
     assert _policy_migration_count(final_events) == 1
 
 
+@pytest.mark.asyncio
+async def test_real_runner_phase_one_to_three_final_publication_and_legacy_readback(
+    database: Any, tmp_path: Path
+) -> None:
+    _, sessions = database
+    states = SqlAlchemyLegacyBuildStateRepository(sessions)
+    script_build_id = await states.create(
+        execution_id=801,
+        topic_build_id=802,
+        topic_id=803,
+        agent_type="script_planner",
+        agent_config={},
+        data_source_url=None,
+        strategies_config={},
+    )
+    phase_two_prompts = script_build_prompt_manifest()
+    all_prompts = (*phase_two_prompts, *phase_three_prompt_manifest())
+    model_manifest = {
+        "presets": {
+            str(item["preset"]): {
+                "model": "scripted-fake-model",
+                "temperature": 0,
+                "max_iterations": 100,
+            }
+            for item in all_prompts
+        }
+    }
+    snapshots = SqlAlchemyInputSnapshotRepository(sessions)
+    snapshot = await snapshots.freeze(
+        ScriptBuildInput(
+            script_build_id=script_build_id,
+            execution_id=801,
+            topic_build_id=802,
+            topic_id=803,
+            topic={"topic": {"id": 803, "result": "A grounded reversal"}},
+            account={"account_name": "fixture-account"},
+            prompt_manifest=tuple(dict(item) for item in phase_two_prompts),
+            model_manifest=model_manifest,
+        )
+    )
+    bindings = SqlAlchemyMissionBindingRepository(sessions)
+    await bindings.create(
+        script_build_id=script_build_id,
+        root_trace_id=ROOT,
+        input_snapshot_id=int(snapshot.snapshot_id),
+        engine_version="phase-three-real-runner-test",
+        schema_version="v3",
+    )
+    task_store = FileSystemTaskStore(str(tmp_path / "ledger"))
+    trace_store = FileSystemTraceStore(str(tmp_path / "trace"))
+    llm = _PhaseThreeScriptedLLM(task_store)
+    runner = AgentRunner(trace_store=trace_store, llm_call=llm)
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    input_service = ScriptInputSnapshotService(
+        legacy_input=cast(Any, None),
+        persona_source=cast(Any, None),
+        strategy_source=cast(Any, None),
+        prompt_source=_PromptCatalog(),
+        snapshots=snapshots,
+    )
+    projection = LegacyDetailProjectionService(sessions, _Authorizer())
+    fencing = FencedCommandGate(sessions)
+    final_uow = SqlAlchemyFinalPublicationUnitOfWork(sessions, projection, fencing)
+    lease_root = tmp_path / "leases"
+
+    def lease_factory(_script_build_id: int) -> OwnerLease:
+        return OwnerLease(sessions, lease_root)
+
+    composition = compose_host(
+        HostDependencies(
+            runner=runner,
+            task_store=task_store,
+            framework_artifact_store=FileSystemArtifactStore(str(tmp_path / "framework-artifacts")),
+            input_snapshot_service=input_service,
+            input_snapshots=snapshots,
+            bindings=bindings,
+            business_artifacts=artifacts,
+            publications=SqlAlchemyPublicationRepository(sessions),
+            legacy_state=states,
+            principal_provider=_PrincipalProvider(),
+            build_authorizer=_Authorizer(),
+            retrieval_adapters={"decode": _DecodeAdapter()},
+            enabled_phase=2,
+            agent_data_root=tmp_path / "agent-data",
+            candidate_workspaces=SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts),
+            default_model_manifest=model_manifest,
+            owner_lease_factory=lease_factory,
+            fenced_command_gate=fencing,
+            final_publication_uow=final_uow,
+            legacy_projection=projection,
+        )
+    )
+    await states.set_status(script_build_id, BuildStatus.RUNNING)
+    await composition.mission_service.run(script_build_id)
+    async with sessions() as session:
+        candidate_ids_before = tuple(
+            (
+                await session.execute(
+                    select(script_build_paragraph.c.id).where(
+                        script_build_paragraph.c.script_build_id == script_build_id,
+                        script_build_paragraph.c.branch_id > 0,
+                    )
+                )
+            ).scalars()
+        )
+
+    assert composition.phase_three is not None
+    await composition.phase_three.advance(script_build_id, Principal("phase-two-owner"))
+    phase_three_run = composition.phase_three._runs[script_build_id]
+    await phase_three_run
+    ledger = await task_store.load(ROOT)
+    root = ledger.tasks[ledger.root_task_id]
+    assert root.status is TaskStatus.COMPLETED
+    assert _task_kind(root) == "root-delivery"
+    assert llm.worker_kinds[-1] == "root-delivery"
+    assert llm.validator_kinds[-1] == "root-delivery"
+
+    assert composition.finalization is not None
+    result = await composition.finalization.finalize(script_build_id, Principal("phase-two-owner"))
+    assert result.committed is True
+    detail = await projection.project_authorized(
+        script_build_id, principal=Principal("phase-two-owner")
+    )
+    assert detail["build"]["status"] == BuildStatus.SUCCESS.value
+    assert projection.to_canonical(detail).canonical_sha256 == result.legacy_projection_digest
+    assert detail["paragraphs"]
+
+    async with sessions() as session:
+        publication = (
+            (
+                await session.execute(
+                    select(publication_table).where(
+                        publication_table.c.script_build_id == script_build_id,
+                        publication_table.c.publication_type == "final",
+                    )
+                )
+            )
+            .mappings()
+            .one()
+        )
+        binding = (
+            (
+                await session.execute(
+                    select(mission_binding_table).where(
+                        mission_binding_table.c.script_build_id == script_build_id
+                    )
+                )
+            )
+            .mappings()
+            .one()
+        )
+        candidate_ids_after = tuple(
+            (
+                await session.execute(
+                    select(script_build_paragraph.c.id).where(
+                        script_build_paragraph.c.script_build_id == script_build_id,
+                        script_build_paragraph.c.branch_id > 0,
+                    )
+                )
+            ).scalars()
+        )
+    assert publication["state"] == "published"
+    assert binding["accepted_root_artifact_version_id"] == publication["artifact_version_id"]
+    assert candidate_ids_after == candidate_ids_before
+
+
 def _phase_two_policy_count(messages: Sequence[Any]) -> int:
     return len(
         [

script_build_host/tests/test_phase_two_workspace.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
 import asyncio
+import os
 from dataclasses import replace
 
 import pytest
 from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
 from script_build_host.domain.artifacts import ArtifactKind, ArtifactState
 from script_build_host.domain.phase_two_artifacts import CandidateLineageV1, ParagraphArtifactV1
@@ -53,6 +55,37 @@ def _lineage(
     )
 
 
+@pytest.mark.mysql
+@pytest.mark.asyncio
+async def test_mysql_workspace_writes_only_through_candidate_view() -> None:
+    dsn = os.environ.get("SCRIPT_BUILD_TEST_MYSQL_DSN")
+    if not dsn:
+        pytest.skip("SCRIPT_BUILD_TEST_MYSQL_DSN is not configured")
+    engine = create_async_engine(dsn, isolation_level="READ COMMITTED", pool_pre_ping=True)
+    sessions = async_sessionmaker(engine, expire_on_commit=False)
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    repository = SqlAlchemyCandidateWorkspaceRepository(sessions, artifacts)
+    context = _context("mysql-candidate-view", build=920000)
+    try:
+        workspace = await repository.get_or_create(context, artifact_kind=ArtifactKind.PARAGRAPH)
+        paragraph_id = await repository.create_paragraph(
+            context,
+            paragraph_index=1,
+            name="candidate-only",
+            content_range={},
+        )
+        async with sessions() as session:
+            branch = await session.scalar(
+                select(script_build_paragraph.c.branch_id).where(
+                    script_build_paragraph.c.id == paragraph_id
+                )
+            )
+        assert branch == workspace.branch_id > 0
+        assert repository._paragraphs.name == "script_build_candidate_paragraph_v"
+    finally:
+        await engine.dispose()
+
+
 @pytest.mark.asyncio
 async def test_workspace_allocates_positive_branch_and_freezes_full_paragraph(database) -> None:
     _, sessions = database

script_build_host/tests/test_production_composition.py

@@ -92,6 +92,7 @@ def test_production_requires_decode_service_and_durable_trace_store(tmp_path: Pa
         environment="production",
         read_database_url="mysql+asyncmy://reader:read@db.example/app",
         write_database_url="mysql+asyncmy://writer:write@db.example/app",
+        final_database_url="mysql+asyncmy://publisher:publish@db.example/app",
     )
     with pytest.raises(RuntimeError, match="DECODE_ENDPOINT"):
         compose_production_host(production, _runtime(tmp_path))

script_build_host/tests/test_repositories.py

@@ -286,7 +286,7 @@ async def test_direction_and_publication_are_idempotent_and_error_is_redacted(
     )
     assert still_published.state is PublicationState.PUBLISHED
     assert still_published.publication_revision == 1
-    with pytest.raises(Exception, match="only an accepted direction"):
+    with pytest.raises(Exception, match="already bound to another publication"):
         await publications.prepare(
             script_build_id=1,
             publication_type=PublicationType.FINAL,