fix(host): fence phase-three state transitions

script_build_host/src/script_build_host/infrastructure/ownership.py

@@ -167,6 +167,108 @@ class FencedCommandGate:
             await self.verify_in_session(session, token)
             return await mutation(session)
 
+    async def compare_and_set_input_snapshot(
+        self,
+        token: MissionOwnerToken,
+        *,
+        expected_snapshot_id: int,
+        new_snapshot_id: int,
+    ) -> None:
+        """Move the binding to one immutable snapshot under the owner fence."""
+
+        async with self._sessions() as session, session.begin():
+            await self.verify_in_session(session, token)
+            result = await session.execute(
+                update(mission_binding_table)
+                .where(
+                    mission_binding_table.c.script_build_id == token.script_build_id,
+                    mission_binding_table.c.input_snapshot_id == expected_snapshot_id,
+                )
+                .values(input_snapshot_id=new_snapshot_id, updated_at=datetime.now(UTC))
+            )
+            if result.rowcount != 1:
+                current = await session.scalar(
+                    select(mission_binding_table.c.input_snapshot_id).where(
+                        mission_binding_table.c.script_build_id == token.script_build_id
+                    )
+                )
+                if current != new_snapshot_id:
+                    raise MissionFencingTokenStale()
+
+    async def begin_phase(self, token: MissionOwnerToken) -> None:
+        """Project the legacy build to running without a stop-overwrite window."""
+
+        await self._update_runtime_state(
+            token,
+            expected_statuses=(BuildStatus.PARTIAL.value, BuildStatus.RUNNING.value),
+            status=BuildStatus.RUNNING.value,
+            end_time=None,
+            error_message=None,
+            summary=None,
+        )
+
+    async def set_checkpoint(
+        self,
+        token: MissionOwnerToken,
+        *,
+        checkpoint_code: str,
+        summary: str,
+    ) -> None:
+        """Persist the Phase 3 checkpoint in the same transaction as fencing."""
+
+        await self._update_runtime_state(
+            token,
+            expected_statuses=(BuildStatus.RUNNING.value,),
+            idempotent_checkpoint=checkpoint_code[:255],
+            status=BuildStatus.PARTIAL.value,
+            error_message=checkpoint_code[:255],
+            summary=summary[:2000],
+            reson_trace_id=token.root_trace_id[:200],
+            end_time=datetime.now(UTC),
+        )
+
+    async def _update_runtime_state(
+        self,
+        token: MissionOwnerToken,
+        *,
+        expected_statuses: tuple[str, ...],
+        idempotent_checkpoint: str | None = None,
+        **values: Any,
+    ) -> None:
+        async with self._sessions() as session, session.begin():
+            await self.verify_in_session(session, token)
+            result = await session.execute(
+                update(self._runtime_record)
+                .where(
+                    self._runtime_record.c.id == token.script_build_id,
+                    self._runtime_record.c.is_deleted.is_(False),
+                    self._runtime_record.c.status.in_(expected_statuses),
+                )
+                .values(**values)
+            )
+            if result.rowcount != 1:
+                current = (
+                    (
+                        await session.execute(
+                            select(
+                                script_build_record.c.status,
+                                script_build_record.c.error_message,
+                            ).where(script_build_record.c.id == token.script_build_id)
+                        )
+                    )
+                    .mappings()
+                    .one_or_none()
+                )
+                if current is None:
+                    raise BuildNotFound()
+                if (
+                    idempotent_checkpoint is not None
+                    and current["status"] == BuildStatus.PARTIAL.value
+                    and current["error_message"] == idempotent_checkpoint
+                ):
+                    return
+                raise MissionFencingTokenStale()
+
     async def verify_in_session(
         self,
         session: AsyncSession,
@@ -204,6 +306,8 @@ class FencedCommandGate:
             raise BuildNotFound()
         if status in {BuildStatus.STOPPING.value, BuildStatus.STOPPED.value}:
             raise MissionStopRequested()
+        if status == BuildStatus.FAILED.value:
+            raise MissionFencingTokenStale()
         if status == BuildStatus.SUCCESS.value and not allow_success:
             raise MissionFencingTokenStale()
 

script_build_host/tests/test_phase_three_ownership.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from pathlib import Path
 
 import pytest
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
 
 from script_build_host.domain.errors import (
@@ -11,7 +12,9 @@ from script_build_host.domain.errors import (
     MissionStopRequested,
 )
 from script_build_host.domain.records import BuildStatus
+from script_build_host.infrastructure.legacy_tables import script_build_record
 from script_build_host.infrastructure.ownership import FencedCommandGate, OwnerLease
+from script_build_host.infrastructure.tables import mission_binding_table
 from script_build_host.repositories.legacy_state import SqlAlchemyLegacyBuildStateRepository
 from script_build_host.repositories.sqlalchemy import SqlAlchemyMissionBindingRepository
 
@@ -78,3 +81,86 @@ async def test_stop_epoch_fences_current_owner_without_taking_process_lease(
         is BuildStatus.STOPPING
     )
     await lease.release()
+
+
+@pytest.mark.asyncio
+async def test_phase_three_state_and_snapshot_mutations_share_the_owner_fence(
+    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="phase-three-owner")
+    token = await lease.acquire(build_id)
+    gate = FencedCommandGate(sessions)
+    try:
+        await gate.compare_and_set_input_snapshot(token, expected_snapshot_id=1, new_snapshot_id=2)
+        await gate.begin_phase(token)
+        await gate.set_checkpoint(
+            token,
+            checkpoint_code="PHASE_THREE_ROOT_ACCEPTED",
+            summary="accepted",
+        )
+        await gate.set_checkpoint(
+            token,
+            checkpoint_code="PHASE_THREE_ROOT_ACCEPTED",
+            summary="accepted",
+        )
+        async with sessions() as session:
+            assert (
+                await session.scalar(
+                    select(mission_binding_table.c.input_snapshot_id).where(
+                        mission_binding_table.c.script_build_id == build_id
+                    )
+                )
+                == 2
+            )
+            row = (
+                (
+                    await session.execute(
+                        select(script_build_record).where(script_build_record.c.id == build_id)
+                    )
+                )
+                .mappings()
+                .one()
+            )
+            assert row["status"] == BuildStatus.PARTIAL.value
+            assert row["error_message"] == "PHASE_THREE_ROOT_ACCEPTED"
+
+        await gate.request_stop(build_id)
+        with pytest.raises(MissionStopRequested):
+            await gate.begin_phase(token)
+        with pytest.raises(MissionStopRequested):
+            await gate.compare_and_set_input_snapshot(
+                token, expected_snapshot_id=2, new_snapshot_id=3
+            )
+    finally:
+        await lease.release()
+
+
+@pytest.mark.asyncio
+async def test_failed_build_cannot_be_resurrected_by_an_owner_token(
+    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="failed-owner")
+    token = await lease.acquire(build_id)
+    state = SqlAlchemyLegacyBuildStateRepository(sessions)
+    await state.set_status(build_id, BuildStatus.FAILED, error_summary="failed")
+    gate = FencedCommandGate(sessions)
+    try:
+        with pytest.raises(MissionFencingTokenStale):
+            await gate.begin_phase(token)
+        with pytest.raises(MissionFencingTokenStale):
+            await gate.set_checkpoint(
+                token,
+                checkpoint_code="PHASE_THREE_ROOT_ACCEPTED",
+                summary="must not restore",
+            )
+        with pytest.raises(MissionFencingTokenStale):
+            await gate.compare_and_set_input_snapshot(
+                token, expected_snapshot_id=1, new_snapshot_id=2
+            )
+        assert await state.get_status(build_id) is BuildStatus.FAILED
+    finally:
+        await lease.release()