feat(script-build): fence ownership and candidate writes

script_build_host/src/script_build_host/application/mission_service.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
+from contextvars import ContextVar
 from dataclasses import dataclass, field
 from typing import Any, Protocol
 from uuid import uuid4
@@ -48,6 +49,7 @@ from .mission_factory import ScriptMissionFactory
 
 PHASE_ONE_CAPABILITY_BOUNDARY = "PHASE_ONE_CAPABILITY_BOUNDARY"
 PHASE_TWO_CANDIDATE_PORTFOLIO_READY = "PHASE_TWO_CANDIDATE_PORTFOLIO_READY"
+_FENCED_BUILD_ID: ContextVar[int | None] = ContextVar("script_build_fenced_build", default=None)
 
 
 class BuildTransitionGate:
@@ -78,6 +80,7 @@ class StartScriptBuildCommand:
     runtime_prompt_manifest: tuple[dict[str, Any], ...] = ()
     datasource_manifest: dict[str, Any] = field(default_factory=dict)
     model_manifest: dict[str, Any] = field(default_factory=dict)
+    root_trace_id: str | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -254,6 +257,8 @@ class ScriptMissionService:
         phase_two_required_presets: tuple[str, ...] = (),
         phase_two_boundary_verifier: PhaseTwoBoundaryVerifier | None = None,
         phase_two_model_manifest: dict[str, Any] | None = None,
+        owner_lease_factory: Any | None = None,
+        fenced_command_gate: Any | None = None,
     ) -> None:
         self.runner = runner
         self.coordinator = coordinator
@@ -273,14 +278,17 @@ class ScriptMissionService:
             else BuildTransitionGate()
         )
         self.runtime_manifest_provider = runtime_manifest_provider
-        if enabled_phase not in {1, 2}:
-            raise ValueError("enabled_phase must be 1 or 2")
+        if enabled_phase not in {1, 2, 3}:
+            raise ValueError("enabled_phase must be 1, 2 or 3")
         self.enabled_phase = enabled_phase
         self.phase_two_prompt_requests = phase_two_prompt_requests
         self.phase_two_required_presets = frozenset(phase_two_required_presets)
         self.phase_two_boundary_verifier = phase_two_boundary_verifier
         self.phase_two_model_manifest = dict(phase_two_model_manifest or {})
+        self.owner_lease_factory = owner_lease_factory
+        self.fenced_command_gate = fenced_command_gate
         self._runs: dict[int, asyncio.Task[None]] = {}
+        self._owner_tokens: dict[int, Any] = {}
         self._phase_advance_locks: dict[int, asyncio.Lock] = {}
 
     async def start(self, command: StartScriptBuildCommand) -> ScriptMissionStartResult:
@@ -306,6 +314,7 @@ class ScriptMissionService:
             raise ProtocolViolation("agent_config must remain an object after redaction")
         if not isinstance(safe_strategies, dict):
             raise ProtocolViolation("strategies_config must remain an object after redaction")
+        root_trace_id = command.root_trace_id or str(uuid4())
         script_build_id = await self.legacy_state.create(
             execution_id=command.execution_id,
             topic_build_id=command.topic_build_id,
@@ -314,6 +323,7 @@ class ScriptMissionService:
             agent_config=safe_config,
             data_source_url=command.data_source_url,
             strategies_config=safe_strategies,
+            root_trace_id=root_trace_id,
         )
         try:
             runtime_datasources = (
@@ -342,7 +352,6 @@ class ScriptMissionService:
             snapshot = await self.input_snapshots.freeze(assembled)
             if self.enabled_phase >= 2:
                 self._require_phase_two_prompts(snapshot)
-            root_trace_id = str(uuid4())
             binding = await self.bindings.create(
                 script_build_id=script_build_id,
                 root_trace_id=root_trace_id,
@@ -358,8 +367,13 @@ class ScriptMissionService:
                 error_summary=type(exc).__name__,
             )
             raise
+        lease = None
+        if self.owner_lease_factory is not None:
+            lease = self.owner_lease_factory(script_build_id)
+            token = await lease.acquire(script_build_id)
+            self._owner_tokens[script_build_id] = token
         task = asyncio.create_task(
-            self.run(script_build_id), name=f"script-build:{script_build_id}"
+            self._run_owned(script_build_id, lease), name=f"script-build:{script_build_id}"
         )
         self._runs[script_build_id] = task
         task.add_done_callback(lambda done: self._discard_run(script_build_id, done))
@@ -370,9 +384,31 @@ class ScriptMissionService:
             input_snapshot_id=snapshot.snapshot_id,
         )
 
+    async def start_result_by_root(self, root_trace_id: str) -> ScriptMissionStartResult:
+        binding = await self.bindings.get_by_root(root_trace_id)
+        return ScriptMissionStartResult(
+            script_build_id=binding.script_build_id,
+            status=await self.legacy_state.get_status(binding.script_build_id),
+            root_trace_id=binding.root_trace_id,
+            input_snapshot_id=str(binding.input_snapshot_id),
+        )
+
     async def run(self, script_build_id: int) -> None:
         await self.run_to_enabled_phase(script_build_id)
 
+    async def _run_owned(self, script_build_id: int, lease: Any | None) -> None:
+        try:
+            token = self._owner_tokens.get(script_build_id)
+            if token is not None and self.fenced_command_gate is not None:
+                await self.fenced_command_gate.verify(token)
+            await self.run(script_build_id)
+            if token is not None and self.fenced_command_gate is not None:
+                await self.fenced_command_gate.verify(token)
+        finally:
+            self._owner_tokens.pop(script_build_id, None)
+            if lease is not None:
+                await lease.release()
+
     async def run_to_enabled_phase(self, script_build_id: int) -> None:
         await self.run_phase_one(script_build_id)
         if (
@@ -1087,7 +1123,9 @@ class ScriptMissionService:
                 return StopResult(script_build_id, BuildStatus.STOPPED)
             if status in {BuildStatus.FAILED, BuildStatus.SUCCESS}:
                 return StopResult(script_build_id, status)
-            if status is not BuildStatus.STOPPING:
+            if self.fenced_command_gate is not None:
+                await self.fenced_command_gate.request_stop(script_build_id)
+            elif status is not BuildStatus.STOPPING:
                 await self.legacy_state.set_status(script_build_id, BuildStatus.STOPPING)
         try:
             binding = await self.bindings.get_by_build(script_build_id)
@@ -1149,10 +1187,37 @@ class ScriptMissionService:
         )
 
     async def ensure_dispatch_allowed(self, script_build_id: int) -> None:
+        if _FENCED_BUILD_ID.get() == script_build_id:
+            return
+        token = self._owner_tokens.get(script_build_id)
+        if token is not None and self.fenced_command_gate is not None:
+            await self.fenced_command_gate.verify(token)
         status = await self.legacy_state.get_status(script_build_id)
         if status in {BuildStatus.STOPPING, BuildStatus.STOPPED}:
             raise ProtocolViolation("new dispatch is forbidden after stop intent")
 
+    async def execute_fenced(self, script_build_id: int, mutation: Any) -> Any:
+        token = self._owner_tokens.get(script_build_id)
+        if token is None or self.fenced_command_gate is None:
+            await self.ensure_dispatch_allowed(script_build_id)
+            return await mutation()
+
+        async def execute(_: Any) -> Any:
+            marker = _FENCED_BUILD_ID.set(script_build_id)
+            try:
+                return await mutation()
+            finally:
+                _FENCED_BUILD_ID.reset(marker)
+
+        return await self.fenced_command_gate.execute(token, execute)
+
+    def register_owner_token(self, token: Any) -> None:
+        self._owner_tokens[token.script_build_id] = token
+
+    def unregister_owner_token(self, token: Any) -> None:
+        if self._owner_tokens.get(token.script_build_id) == token:
+            self._owner_tokens.pop(token.script_build_id, None)
+
     async def _wait_stopped(
         self,
         root_trace_id: str,

script_build_host/src/script_build_host/infrastructure/config.py

@@ -30,6 +30,7 @@ class ScriptBuildSettings(BaseSettings):
     environment: str = "production"
     read_database_url: SecretStr = Field(repr=False)
     write_database_url: SecretStr = Field(repr=False)
+    final_database_url: SecretStr | None = Field(default=None, repr=False)
     agent_data_root: Path
     persona_root: Path
     section_pattern_root: Path
@@ -57,7 +58,7 @@ class ScriptBuildSettings(BaseSettings):
     max_image_bytes: int = Field(default=10_000_000, gt=0)
     max_total_image_bytes: int = Field(default=50_000_000, gt=0)
     enable_image_understanding: bool = False
-    enabled_phase: int = Field(default=2, ge=1, le=2)
+    enabled_phase: int = Field(default=3, ge=1, le=3)
     max_upload_bytes: int = Field(default=5_000_000, gt=0)
     max_upload_points: int = Field(default=500, gt=0, le=10_000)
     max_upload_items: int = Field(default=5_000, gt=0, le=100_000)
@@ -90,11 +91,20 @@ class ScriptBuildSettings(BaseSettings):
     def validate_database_separation(self) -> Self:
         read_url = make_url(self.read_database_url.get_secret_value())
         write_url = make_url(self.write_database_url.get_secret_value())
+        final_url = make_url(
+            self.final_database_url.get_secret_value()
+            if self.final_database_url is not None
+            else self.write_database_url.get_secret_value()
+        )
         if self.environment.lower() != "test":
             if read_url.get_backend_name() == "sqlite" or write_url.get_backend_name() == "sqlite":
                 raise ValueError("SQLite database URLs are permitted only in the test environment")
             if read_url.username == write_url.username:
                 raise ValueError("production read and write database users must be different")
+            if self.final_database_url is None:
+                raise ValueError("production final publisher database URL is required")
+            if final_url.username in {read_url.username, write_url.username}:
+                raise ValueError("production final publisher database user must be dedicated")
         if self.max_total_image_bytes < self.max_image_bytes:
             raise ValueError("max_total_image_bytes must not be smaller than max_image_bytes")
         return self
@@ -105,6 +115,11 @@ class ScriptBuildSettings(BaseSettings):
     def write_dsn(self) -> str:
         return self.write_database_url.get_secret_value()
 
+    def final_dsn(self) -> str:
+        if self.final_database_url is None:
+            return self.write_dsn()
+        return self.final_database_url.get_secret_value()
+
     def phase_two_limits(self) -> PhaseTwoLimits:
         return PhaseTwoLimits(
             max_tasks=self.phase_two_max_tasks,

script_build_host/src/script_build_host/infrastructure/db.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 
+from sqlalchemy.engine import make_url
 from sqlalchemy.ext.asyncio import (
     AsyncEngine,
     AsyncSession,
@@ -16,21 +17,34 @@ from .config import ScriptBuildSettings
 class DatabaseSessions:
     read_engine: AsyncEngine
     write_engine: AsyncEngine
+    final_engine: AsyncEngine
     read: async_sessionmaker[AsyncSession]
     write: async_sessionmaker[AsyncSession]
+    final: async_sessionmaker[AsyncSession]
 
     async def dispose(self) -> None:
         await self.read_engine.dispose()
         if self.write_engine is not self.read_engine:
             await self.write_engine.dispose()
+        if self.final_engine not in {self.read_engine, self.write_engine}:
+            await self.final_engine.dispose()
 
 
 def create_database_sessions(settings: ScriptBuildSettings) -> DatabaseSessions:
-    read_engine = create_async_engine(settings.read_dsn(), pool_pre_ping=True)
-    write_engine = create_async_engine(settings.write_dsn(), pool_pre_ping=True)
+    read_engine = _create_engine(settings.read_dsn())
+    write_engine = _create_engine(settings.write_dsn())
+    final_engine = _create_engine(settings.final_dsn())
     return DatabaseSessions(
         read_engine=read_engine,
         write_engine=write_engine,
+        final_engine=final_engine,
         read=async_sessionmaker(read_engine, expire_on_commit=False),
         write=async_sessionmaker(write_engine, expire_on_commit=False),
+        final=async_sessionmaker(final_engine, expire_on_commit=False),
     )
+
+
+def _create_engine(url: str) -> AsyncEngine:
+    if make_url(url).get_backend_name() == "mysql":
+        return create_async_engine(url, pool_pre_ping=True, isolation_level="READ COMMITTED")
+    return create_async_engine(url, pool_pre_ping=True)

script_build_host/src/script_build_host/infrastructure/ownership.py

@@ -0,0 +1,250 @@
+"""Single-host mission ownership and database fencing.
+
+The file descriptor lock remains held for the lifetime of an owner.  Database
+epochs make every later command independently reject a stale process.
+"""
+
+from __future__ import annotations
+
+import fcntl
+import os
+from collections.abc import Awaitable, Callable
+from contextlib import AbstractAsyncContextManager
+from datetime import UTC, datetime
+from pathlib import Path
+from types import TracebackType
+from typing import Any
+from uuid import uuid4
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from script_build_host.domain.errors import (
+    BuildNotFound,
+    MissionAlreadyOwned,
+    MissionFencingTokenStale,
+    MissionStopRequested,
+)
+from script_build_host.domain.records import BuildStatus, MissionOwnerToken
+from script_build_host.infrastructure.legacy_tables import (
+    script_build_record,
+    script_build_runtime_record,
+)
+from script_build_host.infrastructure.tables import mission_binding_table
+
+
+class OwnerLease(AbstractAsyncContextManager[MissionOwnerToken]):
+    def __init__(
+        self,
+        sessions: async_sessionmaker[AsyncSession],
+        lock_root: Path,
+        *,
+        owner_instance_id: str | None = None,
+    ) -> None:
+        self._sessions = sessions
+        self._lock_root = lock_root.resolve()
+        self._owner_instance_id = owner_instance_id or str(uuid4())
+        self._fd: int | None = None
+        self._token: MissionOwnerToken | None = None
+
+    async def acquire(self, script_build_id: int) -> MissionOwnerToken:
+        if self._fd is not None:
+            if self._token is None or self._token.script_build_id != script_build_id:
+                raise MissionAlreadyOwned()
+            return self._token
+        self._lock_root.mkdir(parents=True, exist_ok=True)
+        path = self._lock_root / f"script-build-{script_build_id}.owner.lock"
+        fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as exc:
+            os.close(fd)
+            raise MissionAlreadyOwned() from exc
+        try:
+            async with self._sessions() as session, session.begin():
+                row = (
+                    (
+                        await session.execute(
+                            select(mission_binding_table)
+                            .where(mission_binding_table.c.script_build_id == script_build_id)
+                            .with_for_update()
+                        )
+                    )
+                    .mappings()
+                    .one_or_none()
+                )
+                if row is None:
+                    raise BuildNotFound()
+                epoch = int(row["owner_epoch"] or 0) + 1
+                now = datetime.now(UTC)
+                await session.execute(
+                    update(mission_binding_table)
+                    .where(mission_binding_table.c.id == row["id"])
+                    .values(
+                        owner_epoch=epoch,
+                        owner_instance_id=self._owner_instance_id,
+                        owner_acquired_at=now,
+                        updated_at=now,
+                    )
+                )
+                token = MissionOwnerToken(
+                    script_build_id=script_build_id,
+                    root_trace_id=str(row["root_trace_id"]),
+                    owner_epoch=epoch,
+                    stop_epoch=int(row["stop_epoch"] or 0),
+                    owner_instance_id=self._owner_instance_id,
+                )
+        except BaseException:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+            os.close(fd)
+            raise
+        self._fd = fd
+        self._token = token
+        return token
+
+    async def release(self) -> None:
+        fd, token = self._fd, self._token
+        if fd is None:
+            return
+        try:
+            if token is not None:
+                async with self._sessions() as session, session.begin():
+                    await session.execute(
+                        update(mission_binding_table)
+                        .where(
+                            mission_binding_table.c.script_build_id == token.script_build_id,
+                            mission_binding_table.c.owner_epoch == token.owner_epoch,
+                            mission_binding_table.c.owner_instance_id == token.owner_instance_id,
+                        )
+                        .values(
+                            owner_instance_id=None,
+                            owner_acquired_at=None,
+                            updated_at=datetime.now(UTC),
+                        )
+                    )
+        finally:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+            os.close(fd)
+            self._fd = None
+            self._token = None
+
+    async def __aenter__(self) -> MissionOwnerToken:
+        if self._token is None:
+            raise RuntimeError("OwnerLease.acquire() must be called before entering the lease")
+        return self._token
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        await self.release()
+
+
+class FencedCommandGate:
+    """Validate one owner token while holding binding/build row locks."""
+
+    def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
+        self._sessions = sessions
+        bind = sessions.kw.get("bind")
+        self._runtime_record = (
+            script_build_runtime_record
+            if getattr(getattr(bind, "dialect", None), "name", None) == "mysql"
+            else script_build_record
+        )
+
+    async def verify(self, token: MissionOwnerToken) -> None:
+        async with self._sessions() as session, session.begin():
+            await self.verify_in_session(session, token)
+
+    async def execute(
+        self,
+        token: MissionOwnerToken,
+        mutation: Callable[[AsyncSession], Awaitable[Any]],
+    ) -> Any:
+        async with self._sessions() as session, session.begin():
+            await self.verify_in_session(session, token)
+            return await mutation(session)
+
+    async def verify_in_session(
+        self,
+        session: AsyncSession,
+        token: MissionOwnerToken,
+        *,
+        allow_success: bool = False,
+    ) -> None:
+        binding = (
+            (
+                await session.execute(
+                    select(mission_binding_table)
+                    .where(mission_binding_table.c.script_build_id == token.script_build_id)
+                    .with_for_update()
+                )
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if binding is None:
+            raise BuildNotFound()
+        if (
+            str(binding["root_trace_id"]) != token.root_trace_id
+            or int(binding["owner_epoch"] or 0) != token.owner_epoch
+            or binding["owner_instance_id"] != token.owner_instance_id
+        ):
+            raise MissionFencingTokenStale()
+        if int(binding["stop_epoch"] or 0) != token.stop_epoch:
+            raise MissionStopRequested()
+        status = await session.scalar(
+            select(script_build_record.c.status)
+            .where(script_build_record.c.id == token.script_build_id)
+            .with_for_update()
+        )
+        if status is None:
+            raise BuildNotFound()
+        if status in {BuildStatus.STOPPING.value, BuildStatus.STOPPED.value}:
+            raise MissionStopRequested()
+        if status == BuildStatus.SUCCESS.value and not allow_success:
+            raise MissionFencingTokenStale()
+
+    async def request_stop(self, script_build_id: int) -> int:
+        """Persist stop intent without acquiring the process OwnerLease."""
+
+        async with self._sessions() as session, session.begin():
+            binding = (
+                (
+                    await session.execute(
+                        select(mission_binding_table)
+                        .where(mission_binding_table.c.script_build_id == script_build_id)
+                        .with_for_update()
+                    )
+                )
+                .mappings()
+                .one_or_none()
+            )
+            if binding is None:
+                raise BuildNotFound()
+            status = await session.scalar(
+                select(script_build_record.c.status)
+                .where(script_build_record.c.id == script_build_id)
+                .with_for_update()
+            )
+            if status is None:
+                raise BuildNotFound()
+            if status in {BuildStatus.SUCCESS.value, BuildStatus.STOPPED.value}:
+                return int(binding["stop_epoch"] or 0)
+            stop_epoch = int(binding["stop_epoch"] or 0) + 1
+            await session.execute(
+                update(mission_binding_table)
+                .where(mission_binding_table.c.id == binding["id"])
+                .values(stop_epoch=stop_epoch, updated_at=datetime.now(UTC))
+            )
+            await session.execute(
+                update(self._runtime_record)
+                .where(self._runtime_record.c.id == script_build_id)
+                .values(status=BuildStatus.STOPPING.value, end_time=None)
+            )
+            return stop_epoch
+
+
+__all__ = ["FencedCommandGate", "OwnerLease"]

script_build_host/src/script_build_host/repositories/legacy_state.py

@@ -1,20 +1,30 @@
 from __future__ import annotations
 
 from datetime import UTC, datetime
-from typing import Any, cast
+from typing import Any
+from uuid import uuid4
 
-from sqlalchemy import CursorResult, insert, select, update
+from sqlalchemy import insert, select, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from script_build_host.domain.errors import BuildNotFound, ProtocolViolation
 from script_build_host.domain.records import BuildStatus
-from script_build_host.infrastructure.legacy_tables import script_build_record
+from script_build_host.infrastructure.legacy_tables import (
+    script_build_record,
+    script_build_runtime_record,
+)
 from script_build_host.infrastructure.redaction import redact_text
 
 
 class SqlAlchemyLegacyBuildStateRepository:
     def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
         self._sessions = sessions
+        bind = sessions.kw.get("bind")
+        self._runtime_record = (
+            script_build_runtime_record
+            if getattr(getattr(bind, "dialect", None), "name", None) == "mysql"
+            else script_build_record
+        )
 
     async def create(
         self,
@@ -26,10 +36,11 @@ class SqlAlchemyLegacyBuildStateRepository:
         agent_config: dict[str, Any] | None,
         data_source_url: str | None,
         strategies_config: dict[str, Any],
+        root_trace_id: str | None = None,
     ) -> int:
         async with self._sessions() as session, session.begin():
             result = await session.execute(
-                insert(script_build_record).values(
+                insert(self._runtime_record).values(
                     execution_id=execution_id,
                     topic_build_id=topic_build_id,
                     topic_id=topic_id,
@@ -37,13 +48,14 @@ class SqlAlchemyLegacyBuildStateRepository:
                     agent_config=agent_config,
                     data_source_url=data_source_url,
                     strategies_config=strategies_config,
+                    reson_trace_id=root_trace_id or str(uuid4()),
                     status=BuildStatus.RUNNING.value,
                     is_deleted=False,
                     is_favorited=False,
                     start_time=datetime.now(UTC),
                 )
             )
-            primary_key = cast(CursorResult[Any], result).inserted_primary_key
+            primary_key = result.inserted_primary_key
             if primary_key is None or primary_key[0] is None:
                 raise BuildNotFound()
             return int(primary_key[0])
@@ -66,14 +78,14 @@ class SqlAlchemyLegacyBuildStateRepository:
             values["end_time"] = datetime.now(UTC)
         async with self._sessions() as session, session.begin():
             result = await session.execute(
-                update(script_build_record)
+                update(self._runtime_record)
                 .where(
-                    script_build_record.c.id == script_build_id,
-                    script_build_record.c.is_deleted.is_(False),
+                    self._runtime_record.c.id == script_build_id,
+                    self._runtime_record.c.is_deleted.is_(False),
                 )
                 .values(**values)
             )
-            if cast(CursorResult[Any], result).rowcount != 1:
+            if result.rowcount != 1:
                 raise BuildNotFound()
 
     async def get_status(self, script_build_id: int) -> BuildStatus:
@@ -91,14 +103,14 @@ class SqlAlchemyLegacyBuildStateRepository:
     async def project_direction(self, script_build_id: int, legacy_markdown: str) -> None:
         async with self._sessions() as session, session.begin():
             result = await session.execute(
-                update(script_build_record)
+                update(self._runtime_record)
                 .where(
-                    script_build_record.c.id == script_build_id,
-                    script_build_record.c.is_deleted.is_(False),
+                    self._runtime_record.c.id == script_build_id,
+                    self._runtime_record.c.is_deleted.is_(False),
                 )
                 .values(script_direction=legacy_markdown)
             )
-            if cast(CursorResult[Any], result).rowcount != 1:
+            if result.rowcount != 1:
                 raise BuildNotFound()
 
     async def get_direction(self, script_build_id: int) -> str | None:
@@ -128,14 +140,14 @@ class SqlAlchemyLegacyBuildStateRepository:
         }
         async with self._sessions() as session, session.begin():
             result = await session.execute(
-                update(script_build_record)
+                update(self._runtime_record)
                 .where(
-                    script_build_record.c.id == script_build_id,
-                    script_build_record.c.is_deleted.is_(False),
+                    self._runtime_record.c.id == script_build_id,
+                    self._runtime_record.c.is_deleted.is_(False),
                 )
                 .values(**values)
             )
-            if cast(CursorResult[Any], result).rowcount != 1:
+            if result.rowcount != 1:
                 raise BuildNotFound()
 
     async def begin_phase(self, script_build_id: int) -> None:

script_build_host/src/script_build_host/repositories/workspace.py

@@ -36,6 +36,9 @@ from script_build_host.domain.workspaces import (
 )
 from script_build_host.infrastructure.canonical_json import canonical_sha256, normalize_json
 from script_build_host.infrastructure.legacy_tables import (
+    script_build_candidate_element,
+    script_build_candidate_paragraph,
+    script_build_candidate_paragraph_element,
     script_build_element,
     script_build_paragraph,
     script_build_paragraph_element,
@@ -71,6 +74,15 @@ class SqlAlchemyCandidateWorkspaceRepository:
     ) -> None:
         self._sessions = sessions
         self._artifacts = artifacts
+        bind = sessions.kw.get("bind")
+        use_views = getattr(getattr(bind, "dialect", None), "name", None) == "mysql"
+        self._paragraphs = script_build_candidate_paragraph if use_views else script_build_paragraph
+        self._elements = script_build_candidate_element if use_views else script_build_element
+        self._links = (
+            script_build_candidate_paragraph_element
+            if use_views
+            else script_build_paragraph_element
+        )
 
     async def get_or_create(
         self,
@@ -214,7 +226,7 @@ class SqlAlchemyCandidateWorkspaceRepository:
                         "LEGACY_REFERENCE_INVALID", "child content range exceeds its parent"
                     )
             result = await session.execute(
-                insert(script_build_paragraph).values(
+                insert(self._paragraphs).values(
                     script_build_id=workspace.script_build_id,
                     branch_id=workspace.branch_id,
                     base_ref_id=None,
@@ -356,11 +368,11 @@ class SqlAlchemyCandidateWorkspaceRepository:
                         )
                 if values.get("is_active") is False:
                     active_child = await session.scalar(
-                        select(script_build_paragraph.c.id).where(
-                            script_build_paragraph.c.script_build_id == workspace.script_build_id,
-                            script_build_paragraph.c.branch_id == workspace.branch_id,
-                            script_build_paragraph.c.parent_id == paragraph_id,
-                            script_build_paragraph.c.is_active.is_(True),
+                        select(self._paragraphs.c.id).where(
+                            self._paragraphs.c.script_build_id == workspace.script_build_id,
+                            self._paragraphs.c.branch_id == workspace.branch_id,
+                            self._paragraphs.c.parent_id == paragraph_id,
+                            self._paragraphs.c.is_active.is_(True),
                         )
                     )
                     if active_child is not None:
@@ -374,21 +386,20 @@ class SqlAlchemyCandidateWorkspaceRepository:
                 prepared.append((paragraph_id, values))
             for paragraph_id, values in prepared:
                 await session.execute(
-                    update(script_build_paragraph)
+                    update(self._paragraphs)
                     .where(
-                        script_build_paragraph.c.id == paragraph_id,
-                        script_build_paragraph.c.script_build_id == workspace.script_build_id,
-                        script_build_paragraph.c.branch_id == workspace.branch_id,
+                        self._paragraphs.c.id == paragraph_id,
+                        self._paragraphs.c.script_build_id == workspace.script_build_id,
+                        self._paragraphs.c.branch_id == workspace.branch_id,
                     )
                     .values(**values)
                 )
                 if values.get("is_active") is False:
                     await session.execute(
-                        delete(script_build_paragraph_element).where(
-                            script_build_paragraph_element.c.script_build_id
-                            == workspace.script_build_id,
-                            script_build_paragraph_element.c.branch_id == workspace.branch_id,
-                            script_build_paragraph_element.c.paragraph_id == paragraph_id,
+                        delete(self._links).where(
+                            self._links.c.script_build_id == workspace.script_build_id,
+                            self._links.c.branch_id == workspace.branch_id,
+                            self._links.c.paragraph_id == paragraph_id,
                         )
                     )
             return tuple(item[0] for item in prepared)
@@ -419,8 +430,8 @@ class SqlAlchemyCandidateWorkspaceRepository:
             dimensions = {str(item.get("维度", "")) for item in existing}
             added = [item for item in normalized if item["维度"] not in dimensions]
             await session.execute(
-                update(script_build_paragraph)
-                .where(script_build_paragraph.c.id == paragraph_id)
+                update(self._paragraphs)
+                .where(self._paragraphs.c.id == paragraph_id)
                 .values(**{column: [*existing, *added], "updated_at": _now()})
             )
             return len(added)
@@ -460,8 +471,8 @@ class SqlAlchemyCandidateWorkspaceRepository:
             if deleted_count == 0:
                 raise WorkspaceError("LEGACY_REFERENCE_INVALID", "paragraph atom does not exist")
             await session.execute(
-                update(script_build_paragraph)
-                .where(script_build_paragraph.c.id == paragraph_id)
+                update(self._paragraphs)
+                .where(self._paragraphs.c.id == paragraph_id)
                 .values(**{column: kept, "updated_at": _now()})
             )
             return deleted_count
@@ -487,13 +498,12 @@ class SqlAlchemyCandidateWorkspaceRepository:
             existing = (
                 (
                     await session.execute(
-                        select(script_build_element.c.id).where(
-                            script_build_element.c.script_build_id == workspace.script_build_id,
-                            script_build_element.c.branch_id == workspace.branch_id,
-                            script_build_element.c.name == name.strip(),
-                            script_build_element.c.dimension_primary == dimension_primary,
-                            script_build_element.c.dimension_secondary
-                            == dimension_secondary.strip(),
+                        select(self._elements.c.id).where(
+                            self._elements.c.script_build_id == workspace.script_build_id,
+                            self._elements.c.branch_id == workspace.branch_id,
+                            self._elements.c.name == name.strip(),
+                            self._elements.c.dimension_primary == dimension_primary,
+                            self._elements.c.dimension_secondary == dimension_secondary.strip(),
                         )
                     )
                 )
@@ -503,7 +513,7 @@ class SqlAlchemyCandidateWorkspaceRepository:
             if existing is not None:
                 return int(existing)
             result = await session.execute(
-                insert(script_build_element).values(
+                insert(self._elements).values(
                     script_build_id=workspace.script_build_id,
                     branch_id=workspace.branch_id,
                     base_ref_id=None,
@@ -582,21 +592,20 @@ class SqlAlchemyCandidateWorkspaceRepository:
                 raise WorkspaceError("LEGACY_WRITE_INVALID", "element is_active must be boolean")
             normalized["updated_at"] = _now()
             await session.execute(
-                update(script_build_element)
+                update(self._elements)
                 .where(
-                    script_build_element.c.id == element_id,
-                    script_build_element.c.script_build_id == workspace.script_build_id,
-                    script_build_element.c.branch_id == workspace.branch_id,
+                    self._elements.c.id == element_id,
+                    self._elements.c.script_build_id == workspace.script_build_id,
+                    self._elements.c.branch_id == workspace.branch_id,
                 )
                 .values(**normalized)
             )
             if normalized.get("is_active") is False:
                 await session.execute(
-                    delete(script_build_paragraph_element).where(
-                        script_build_paragraph_element.c.script_build_id
-                        == workspace.script_build_id,
-                        script_build_paragraph_element.c.branch_id == workspace.branch_id,
-                        script_build_paragraph_element.c.element_id == element_id,
+                    delete(self._links).where(
+                        self._links.c.script_build_id == workspace.script_build_id,
+                        self._links.c.branch_id == workspace.branch_id,
+                        self._links.c.element_id == element_id,
                     )
                 )
 
@@ -630,12 +639,11 @@ class SqlAlchemyCandidateWorkspaceRepository:
                 exists = (
                     (
                         await session.execute(
-                            select(script_build_paragraph_element.c.id).where(
-                                script_build_paragraph_element.c.script_build_id
-                                == workspace.script_build_id,
-                                script_build_paragraph_element.c.branch_id == workspace.branch_id,
-                                script_build_paragraph_element.c.paragraph_id == paragraph_id,
-                                script_build_paragraph_element.c.element_id == element_id,
+                            select(self._links.c.id).where(
+                                self._links.c.script_build_id == workspace.script_build_id,
+                                self._links.c.branch_id == workspace.branch_id,
+                                self._links.c.paragraph_id == paragraph_id,
+                                self._links.c.element_id == element_id,
                             )
                         )
                     )
@@ -644,7 +652,7 @@ class SqlAlchemyCandidateWorkspaceRepository:
                 )
                 if exists is None:
                     await session.execute(
-                        insert(script_build_paragraph_element).values(
+                        insert(self._links).values(
                             script_build_id=workspace.script_build_id,
                             branch_id=workspace.branch_id,
                             paragraph_id=paragraph_id,
@@ -681,10 +689,9 @@ class SqlAlchemyCandidateWorkspaceRepository:
             rows = (
                 (
                     await session.execute(
-                        select(script_build_paragraph_element).where(
-                            script_build_paragraph_element.c.script_build_id
-                            == workspace.script_build_id,
-                            script_build_paragraph_element.c.branch_id == workspace.branch_id,
+                        select(self._links).where(
+                            self._links.c.script_build_id == workspace.script_build_id,
+                            self._links.c.branch_id == workspace.branch_id,
                         )
                     )
                 )
@@ -699,11 +706,10 @@ class SqlAlchemyCandidateWorkspaceRepository:
                 for item in (
                     (
                         await session.execute(
-                            select(script_build_paragraph.c.id).where(
-                                script_build_paragraph.c.script_build_id
-                                == workspace.script_build_id,
-                                script_build_paragraph.c.branch_id == workspace.branch_id,
-                                script_build_paragraph.c.is_active.is_(True),
+                            select(self._paragraphs.c.id).where(
+                                self._paragraphs.c.script_build_id == workspace.script_build_id,
+                                self._paragraphs.c.branch_id == workspace.branch_id,
+                                self._paragraphs.c.is_active.is_(True),
                             )
                         )
                     )
@@ -714,10 +720,10 @@ class SqlAlchemyCandidateWorkspaceRepository:
             active_elements = set(
                 (
                     await session.execute(
-                        select(script_build_element.c.id).where(
-                            script_build_element.c.script_build_id == workspace.script_build_id,
-                            script_build_element.c.branch_id == workspace.branch_id,
-                            script_build_element.c.is_active.is_(True),
+                        select(self._elements.c.id).where(
+                            self._elements.c.script_build_id == workspace.script_build_id,
+                            self._elements.c.branch_id == workspace.branch_id,
+                            self._elements.c.is_active.is_(True),
                         )
                     )
                 ).scalars()
@@ -737,11 +743,7 @@ class SqlAlchemyCandidateWorkspaceRepository:
                 )
             ]
             if delete_ids:
-                await session.execute(
-                    delete(script_build_paragraph_element).where(
-                        script_build_paragraph_element.c.id.in_(delete_ids)
-                    )
-                )
+                await session.execute(delete(self._links).where(self._links.c.id.in_(delete_ids)))
             return len(delete_ids)
 
     async def freeze(
@@ -984,41 +986,41 @@ class SqlAlchemyCandidateWorkspaceRepository:
             )
         return workspace
 
-    @staticmethod
     async def _paragraph(
+        self,
         session: AsyncSession,
         workspace: CandidateWorkspace,
         paragraph_id: int,
         *,
         active: bool | None = None,
     ) -> Mapping[str, Any] | None:
-        statement = select(script_build_paragraph).where(
-            script_build_paragraph.c.id == paragraph_id,
-            script_build_paragraph.c.script_build_id == workspace.script_build_id,
-            script_build_paragraph.c.branch_id == workspace.branch_id,
+        statement = select(self._paragraphs).where(
+            self._paragraphs.c.id == paragraph_id,
+            self._paragraphs.c.script_build_id == workspace.script_build_id,
+            self._paragraphs.c.branch_id == workspace.branch_id,
         )
         if active is not None:
-            statement = statement.where(script_build_paragraph.c.is_active.is_(active))
+            statement = statement.where(self._paragraphs.c.is_active.is_(active))
         return cast(
             Mapping[str, Any] | None,
             (await session.execute(statement)).mappings().one_or_none(),
         )
 
-    @staticmethod
     async def _element(
+        self,
         session: AsyncSession,
         workspace: CandidateWorkspace,
         element_id: int,
         *,
         active: bool | None = None,
     ) -> Mapping[str, Any] | None:
-        statement = select(script_build_element).where(
-            script_build_element.c.id == element_id,
-            script_build_element.c.script_build_id == workspace.script_build_id,
-            script_build_element.c.branch_id == workspace.branch_id,
+        statement = select(self._elements).where(
+            self._elements.c.id == element_id,
+            self._elements.c.script_build_id == workspace.script_build_id,
+            self._elements.c.branch_id == workspace.branch_id,
         )
         if active is not None:
-            statement = statement.where(script_build_element.c.is_active.is_(active))
+            statement = statement.where(self._elements.c.is_active.is_(active))
         return cast(
             Mapping[str, Any] | None,
             (await session.execute(statement)).mappings().one_or_none(),
@@ -1030,14 +1032,14 @@ class SqlAlchemyCandidateWorkspaceRepository:
         paragraph_rows = (
             (
                 await session.execute(
-                    select(script_build_paragraph)
+                    select(self._paragraphs)
                     .where(
-                        script_build_paragraph.c.script_build_id == workspace.script_build_id,
-                        script_build_paragraph.c.branch_id == workspace.branch_id,
+                        self._paragraphs.c.script_build_id == workspace.script_build_id,
+                        self._paragraphs.c.branch_id == workspace.branch_id,
                     )
                     .order_by(
-                        script_build_paragraph.c.paragraph_index,
-                        script_build_paragraph.c.id,
+                        self._paragraphs.c.paragraph_index,
+                        self._paragraphs.c.id,
                     )
                 )
             )
@@ -1047,12 +1049,12 @@ class SqlAlchemyCandidateWorkspaceRepository:
         element_rows = (
             (
                 await session.execute(
-                    select(script_build_element)
+                    select(self._elements)
                     .where(
-                        script_build_element.c.script_build_id == workspace.script_build_id,
-                        script_build_element.c.branch_id == workspace.branch_id,
+                        self._elements.c.script_build_id == workspace.script_build_id,
+                        self._elements.c.branch_id == workspace.branch_id,
                     )
-                    .order_by(script_build_element.c.id)
+                    .order_by(self._elements.c.id)
                 )
             )
             .mappings()
@@ -1061,15 +1063,14 @@ class SqlAlchemyCandidateWorkspaceRepository:
         link_rows = (
             (
                 await session.execute(
-                    select(script_build_paragraph_element)
+                    select(self._links)
                     .where(
-                        script_build_paragraph_element.c.script_build_id
-                        == workspace.script_build_id,
-                        script_build_paragraph_element.c.branch_id == workspace.branch_id,
+                        self._links.c.script_build_id == workspace.script_build_id,
+                        self._links.c.branch_id == workspace.branch_id,
                     )
                     .order_by(
-                        script_build_paragraph_element.c.paragraph_id,
-                        script_build_paragraph_element.c.element_id,
+                        self._links.c.paragraph_id,
+                        self._links.c.element_id,
                     )
                 )
             )
@@ -1164,7 +1165,7 @@ class SqlAlchemyCandidateWorkspaceRepository:
         for item in sorted(paragraphs, key=lambda value: (value.level, value.paragraph_index)):
             parent_id = paragraph_map.get(item.parent_id) if item.parent_id is not None else None
             result = await session.execute(
-                insert(script_build_paragraph).values(
+                insert(self._paragraphs).values(
                     script_build_id=workspace.script_build_id,
                     branch_id=workspace.branch_id,
                     base_ref_id=item.paragraph_id,
@@ -1192,7 +1193,7 @@ class SqlAlchemyCandidateWorkspaceRepository:
         element_map: dict[int, int] = {}
         for item in elements:
             result = await session.execute(
-                insert(script_build_element).values(
+                insert(self._elements).values(
                     script_build_id=workspace.script_build_id,
                     branch_id=workspace.branch_id,
                     base_ref_id=item.element_id,
@@ -1214,7 +1215,7 @@ class SqlAlchemyCandidateWorkspaceRepository:
             element_id = element_map.get(item.element_id)
             if paragraph_id is not None and element_id is not None:
                 await session.execute(
-                    insert(script_build_paragraph_element).values(
+                    insert(self._links).values(
                         script_build_id=workspace.script_build_id,
                         branch_id=workspace.branch_id,
                         paragraph_id=paragraph_id,