测试：覆盖迁移、仓储和阶段一完整闭环

验证四张表升级回滚、MySQL 离线 SQL、快照与制品幂等、跨 build 归属和冻结后拒写。

使用真实 Runner、Coordinator 和 FileStore 覆盖 Decode 失败修订、Direction 接受、Root 阻塞、partial 投影、重复派发、停止竞态与重载可读性。

script_build_host/tests/test_migrations.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import hashlib
+import os
+from io import StringIO
+from pathlib import Path
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine, inspect, text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+
+def test_alembic_upgrade_and_downgrade(tmp_path: Path, monkeypatch: object) -> None:
+    root = Path(__file__).parents[1]
+    database = tmp_path / "migration.db"
+    url = f"sqlite+aiosqlite:///{database}"
+    monkeypatch.setenv("SCRIPT_BUILD_WRITE_DATABASE_URL", url)  # type: ignore[attr-defined]
+    config = Config(str(root / "alembic.ini"))
+    command.upgrade(config, "head")
+    sync_engine = create_engine(f"sqlite:///{database}")
+    expected = {
+        "script_build_mission_binding",
+        "script_build_input_snapshot",
+        "script_build_artifact_version",
+        "script_build_publication",
+    }
+    assert expected <= set(inspect(sync_engine).get_table_names())
+    command.downgrade(config, "base")
+    assert not (expected & set(inspect(sync_engine).get_table_names()))
+    sync_engine.dispose()
+
+
+def test_migration_checksum_manifest() -> None:
+    root = Path(__file__).parents[1]
+    manifest = root / "migrations" / "CHECKSUMS.sha256"
+    lines = [line.split(maxsplit=1) for line in manifest.read_text(encoding="utf-8").splitlines()]
+    assert lines
+    for expected, relative in lines:
+        path = root / relative
+        assert hashlib.sha256(path.read_bytes()).hexdigest() == expected
+
+
+def test_mysql_offline_sql_compiles_without_credentials(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)
+    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
+
+
+@pytest.mark.mysql
+@pytest.mark.asyncio
+async def test_explicit_mysql_dsn_is_reachable() -> 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)
+    try:
+        async with engine.connect() as connection:
+            assert await connection.scalar(text("SELECT 1")) == 1
+    finally:
+        await engine.dispose()

script_build_host/tests/test_phase_one_e2e.py

@@ -0,0 +1,534 @@
+from __future__ import annotations
+
+import json
+from dataclasses import replace
+from datetime import UTC, datetime
+from types import SimpleNamespace
+
+import pytest
+from agent import AgentRunner, FileSystemArtifactStore, FileSystemTaskStore, FileSystemTraceStore
+from agent.orchestration import ArtifactRef, TaskStatus, ValidationVerdict
+
+from script_build_host.composition import HostDependencies, compose_host
+from script_build_host.domain.artifacts import (
+    ArtifactKind,
+    ArtifactState,
+    ArtifactVersion,
+    EvidenceRecordV1,
+)
+from script_build_host.domain.input_snapshot import ScriptBuildInputSnapshotV1
+from script_build_host.domain.records import (
+    BuildStatus,
+    MissionBinding,
+    Principal,
+    PublicationState,
+)
+from script_build_host.infrastructure.canonical_json import canonical_sha256
+from script_build_host.tools.gateway import RetrievalResult
+
+
+def _tool(call_id: str, name: str, arguments: dict[str, object]) -> dict[str, object]:
+    return {
+        "content": "",
+        "tool_calls": [
+            {
+                "id": call_id,
+                "type": "function",
+                "function": {"name": name, "arguments": json.dumps(arguments)},
+            }
+        ],
+        "finish_reason": "tool_calls",
+    }
+
+
+def _prompt(messages: list[dict[str, object]]) -> dict[str, object]:
+    for message in reversed(messages):
+        if message.get("role") != "user" or not isinstance(message.get("content"), str):
+            continue
+        try:
+            value = json.loads(message["content"])
+        except ValueError:
+            continue
+        if isinstance(value, dict) and "task_spec" in value:
+            return value
+    raise AssertionError("missing role prompt")
+
+
+def _last_tool_json(messages: list[dict[str, object]]) -> dict[str, object] | None:
+    for message in reversed(messages):
+        if message.get("role") != "tool":
+            continue
+        content = message.get("content")
+        if isinstance(content, str):
+            try:
+                value = json.loads(content)
+                return value if isinstance(value, dict) else None
+            except ValueError:
+                return None
+    return None
+
+
+class _ScriptedLLM:
+    def __init__(self, store: FileSystemTaskStore, root: str) -> None:
+        self.store = store
+        self.root = root
+        self.calls = 0
+        self.decode_versions: list[int] = []
+        self.validated_artifact_kinds: list[str] = []
+        self.replayed_dispatch = False
+
+    def call(self, name: str, arguments: dict[str, object]) -> dict[str, object]:
+        self.calls += 1
+        return _tool(f"call-{self.calls}", name, arguments)
+
+    def call_with_id(
+        self,
+        call_id: str,
+        name: str,
+        arguments: dict[str, object],
+    ) -> dict[str, object]:
+        self.calls += 1
+        return _tool(call_id, name, arguments)
+
+    async def __call__(self, messages, tools, **_kwargs):
+        names = {item["function"]["name"] for item in tools or []}
+        if "task_plan" in names:
+            return await self._planner()
+        if "submit_validation" in names:
+            return self._validator(messages)
+        if "submit_attempt" in names:
+            return self._worker(messages, names)
+        raise AssertionError(sorted(names))
+
+    async def _planner(self):
+        ledger = await self.store.load(self.root)
+        root = ledger.tasks[ledger.root_task_id]
+        direction = next(
+            (
+                item
+                for item in ledger.tasks.values()
+                if "script-build://task-kinds/direction" in item.current_spec.context_refs
+            ),
+            None,
+        )
+        if direction is None:
+            return self.call(
+                "task_plan",
+                {
+                    "operation": "create",
+                    "tasks": [
+                        {
+                            "objective": "build direction",
+                            "acceptance_criteria": [
+                                {
+                                    "criterion_id": "direction-ready",
+                                    "description": "direction is grounded",
+                                    "hard": True,
+                                }
+                            ],
+                            "context_refs": ["script-build://task-kinds/direction"],
+                        }
+                    ],
+                },
+            )
+        retrieval = next(
+            (
+                item
+                for item in ledger.tasks.values()
+                if "script-build://task-kinds/decode-retrieval" in item.current_spec.context_refs
+            ),
+            None,
+        )
+        if retrieval is None:
+            return self.call(
+                "task_plan",
+                {
+                    "operation": "create",
+                    "parent_task_id": direction.task_id,
+                    "tasks": [
+                        {
+                            "objective": "decode evidence",
+                            "acceptance_criteria": [
+                                {
+                                    "criterion_id": "evidence-ready",
+                                    "description": "evidence is relevant",
+                                    "hard": True,
+                                }
+                            ],
+                            "context_refs": ["script-build://task-kinds/decode-retrieval"],
+                        }
+                    ],
+                },
+            )
+        if retrieval.status in {TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN}:
+            return self.call_with_id(
+                f"decode-dispatch-v{retrieval.current_spec_version}",
+                "dispatch_script_tasks",
+                {"task_ids": [retrieval.task_id]},
+            )
+        if retrieval.status == TaskStatus.AWAITING_DECISION:
+            if retrieval.current_spec_version == 1 and not self.replayed_dispatch:
+                self.replayed_dispatch = True
+                return self.call_with_id(
+                    "decode-dispatch-v1",
+                    "dispatch_script_tasks",
+                    {"task_ids": [retrieval.task_id]},
+                )
+            validation = ledger.validations[retrieval.validation_ids[-1]]
+            if validation.verdict == ValidationVerdict.FAILED:
+                return self.call(
+                    "task_decide",
+                    {
+                        "task_id": retrieval.task_id,
+                        "validation_id": validation.validation_id,
+                        "action": "revise",
+                        "reason": "make decode query precise",
+                        "payload": {"objective": "decode evidence revised"},
+                    },
+                )
+            return self.call(
+                "task_decide",
+                {
+                    "task_id": retrieval.task_id,
+                    "validation_id": validation.validation_id,
+                    "action": "accept",
+                    "reason": "decode evidence passed",
+                },
+            )
+        if direction.status in {TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN}:
+            return self.call("dispatch_script_tasks", {"task_ids": [direction.task_id]})
+        if direction.status == TaskStatus.AWAITING_DECISION:
+            validation = ledger.validations[direction.validation_ids[-1]]
+            return self.call(
+                "task_decide",
+                {
+                    "task_id": direction.task_id,
+                    "validation_id": validation.validation_id,
+                    "action": "accept",
+                    "reason": "direction passed",
+                },
+            )
+        assert direction.status == TaskStatus.COMPLETED
+        if root.current_spec_version == 1:
+            return self.call(
+                "task_decide",
+                {
+                    "task_id": root.task_id,
+                    "action": "revise",
+                    "reason": "record accepted direction in root objective",
+                    "payload": {
+                        "objective": root.current_spec.objective + " using accepted direction"
+                    },
+                },
+            )
+        return self.call(
+            "task_decide",
+            {
+                "task_id": root.task_id,
+                "action": "block",
+                "reason": "PHASE_ONE_CAPABILITY_BOUNDARY",
+            },
+        )
+
+    def _worker(self, messages, names):
+        prompt = _prompt(messages)
+        spec = prompt["task_spec"]
+        last = _last_tool_json(messages)
+        if "search_script_decode_case" in names:
+            if last is None or "artifact_ref" not in last:
+                self.decode_versions.append(spec["version"])
+                return self.call(
+                    "search_script_decode_case",
+                    {"return_field": "主脉络", "keyword": "focus", "top_k": 3},
+                )
+            return self.call(
+                "submit_attempt",
+                {
+                    "summary": "decode evidence",
+                    "artifact_refs": [],
+                    "evidence_refs": [last["artifact_ref"]],
+                },
+            )
+        if last is None or "artifact_ref" not in last:
+            accepted = prompt["accepted_child_results"]
+            evidence_uri = accepted[0]["submission"]["evidence_refs"][0]["uri"]
+            return self.call(
+                "save_direction_candidate",
+                {
+                    "goals": [
+                        {
+                            "goal_id": "g1",
+                            "statement": "make one grounded direction",
+                            "rationale": "accepted evidence",
+                        }
+                    ],
+                    "criteria": [
+                        {
+                            "criterion_id": "c1",
+                            "description": "grounded",
+                        }
+                    ],
+                    "legacy_markdown": "# accepted direction",
+                    "evidence_refs": [evidence_uri],
+                },
+            )
+        return self.call(
+            "submit_attempt",
+            {
+                "summary": "direction candidate",
+                "artifact_refs": [last["artifact_ref"]],
+                "evidence_refs": [],
+            },
+        )
+
+    def _validator(self, messages):
+        prompt = _prompt(messages)
+        spec = prompt["task_spec"]
+        kind = (
+            "direction"
+            if "script-build://task-kinds/direction" in spec["context_refs"]
+            else "evidence"
+        )
+        evidence = _last_tool_json(messages)
+        if evidence is None or "items" not in evidence:
+            return self.call(
+                "query_validation_evidence",
+                {"query": kind, "limit": 5},
+            )
+        assert evidence["items"], "validator must read the frozen artifact body"
+        artifact = evidence["items"][0]["artifact"]
+        assert isinstance(artifact, dict)
+        self.validated_artifact_kinds.append(kind)
+        failed = spec["objective"] == "decode evidence" and spec["version"] == 1
+        verdict = "failed" if failed else "passed"
+        return self.call(
+            "submit_validation",
+            {
+                "verdict": verdict,
+                "criterion_results": [
+                    {
+                        "criterion_id": item["criterion_id"],
+                        "verdict": verdict,
+                        "reason": "checked frozen snapshot",
+                    }
+                    for item in spec["acceptance_criteria"]
+                ],
+                "summary": verdict,
+                "recommendation": "revise" if failed else "accept",
+            },
+        )
+
+
+class _Bindings:
+    def __init__(self, binding: MissionBinding) -> None:
+        self.binding = binding
+        self.active = None
+
+    async def get_by_build(self, _: int) -> MissionBinding:
+        return self.binding
+
+    async def get_by_root(self, _: str) -> MissionBinding:
+        return self.binding
+
+    async def set_active_direction(self, **values) -> MissionBinding:
+        self.active = values["artifact_version_id"]
+        return replace(self.binding, active_direction_artifact_version_id=self.active)
+
+
+class _Snapshots:
+    def __init__(self, snapshot: ScriptBuildInputSnapshotV1) -> None:
+        self.snapshot = snapshot
+
+    async def get(self, *_args, **_kwargs) -> ScriptBuildInputSnapshotV1:
+        return self.snapshot
+
+
+class _Artifacts:
+    def __init__(self) -> None:
+        self.values = {}
+        self.next_id = 1
+
+    async def freeze(self, *, artifact, script_build_id, task_id, attempt_id, spec_version):
+        payload = artifact.content_payload()
+        digest = canonical_sha256(payload).wire
+        if isinstance(artifact, EvidenceRecordV1):
+            artifact = replace(artifact, content_sha256=digest)
+            kind = ArtifactKind.EVIDENCE
+        else:
+            artifact = replace(artifact, canonical_sha256=digest)
+            kind = ArtifactKind.DIRECTION
+        value = ArtifactVersion(
+            self.next_id,
+            script_build_id,
+            task_id,
+            attempt_id,
+            spec_version,
+            kind,
+            digest,
+            ArtifactState.FROZEN,
+            artifact,
+            datetime.now(UTC),
+            datetime.now(UTC),
+        )
+        ref = ArtifactRef(
+            uri=f"script-build://artifact-versions/{self.next_id}",
+            kind=kind.value,
+            version=str(self.next_id),
+            digest=digest,
+            metadata={
+                "script_build_id": script_build_id,
+                "task_id": task_id,
+                "attempt_id": attempt_id,
+            },
+        )
+        self.values[self.next_id] = (value, ref)
+        self.next_id += 1
+        return value, ref
+
+    async def read_by_ref(self, ref, **_kwargs):
+        return self.values[int(ref.version)][0]
+
+
+class _Legacy:
+    def __init__(self) -> None:
+        self.status = BuildStatus.RUNNING
+        self.direction = None
+
+    async def get_status(self, _: int) -> BuildStatus:
+        return self.status
+
+    async def set_status(self, _id, status, **_kwargs):
+        self.status = status
+
+    async def project_direction(self, _id, value):
+        assert self.status != BuildStatus.STOPPING
+        self.direction = value
+
+
+class _Publications:
+    def __init__(self) -> None:
+        self.state = PublicationState.PENDING
+
+    async def prepare(self, **values):
+        return SimpleNamespace(publication_id=1, state=self.state, **values)
+
+    async def mark_published(self, _id):
+        self.state = PublicationState.PUBLISHED
+
+    async def mark_failed(self, _id, **_values):
+        self.state = PublicationState.FAILED
+
+
+class _Decode:
+    async def retrieve(self, *, query, snapshot):
+        del snapshot
+        return RetrievalResult(
+            source_refs=(f"decode:{query['keyword']}",),
+            summary="evidence",
+            supports=("direction",),
+            confidence="high",
+            metadata={"index_sha256": "sha256:" + "1" * 64, "ranks": [1]},
+        )
+
+
+class _PrincipalProvider:
+    async def current(self, request_context=None):
+        del request_context
+        return Principal("owner")
+
+
+class _Authorizer:
+    async def require_source_access(self, _principal, **_source):
+        return None
+
+    async def require_access(self, _principal, _build):
+        return None
+
+
+@pytest.mark.asyncio
+async def test_real_runner_phase_one_fail_revise_pass_direction_accept_then_root_block(tmp_path):
+    now = datetime.now(UTC)
+    binding = MissionBinding(1, 7, "phase-one-root", 11, None, None, "test", "v1", now, now)
+    model_presets = {
+        name: {"model": "fake-model", "temperature": 0, "max_iterations": 50}
+        for name in (
+            "script_decode_retrieval_worker",
+            "script_direction_worker",
+            "script_retrieval_validator",
+            "script_candidate_validator",
+        )
+    }
+    snapshot = ScriptBuildInputSnapshotV1(
+        "11",
+        7,
+        1,
+        2,
+        3,
+        {"topic": {"id": 3, "result": "topic"}},
+        {"account_name": "acct"},
+        (),
+        (),
+        (),
+        (),
+        {},
+        {"presets": model_presets},
+        "sha256:" + "a" * 64,
+        now,
+    )
+    task_store = FileSystemTaskStore(str(tmp_path / "ledger"))
+    llm = _ScriptedLLM(task_store, binding.root_trace_id)
+    runner = AgentRunner(trace_store=FileSystemTraceStore(str(tmp_path / "trace")), llm_call=llm)
+    bindings = _Bindings(binding)
+    snapshots = _Snapshots(snapshot)
+    artifacts = _Artifacts()
+    legacy = _Legacy()
+    publications = _Publications()
+    composition = compose_host(
+        HostDependencies(
+            runner=runner,
+            task_store=task_store,
+            framework_artifact_store=FileSystemArtifactStore(str(tmp_path / "artifacts")),
+            input_snapshot_service=SimpleNamespace(get=snapshots.get),
+            input_snapshots=snapshots,
+            bindings=bindings,
+            business_artifacts=artifacts,
+            publications=publications,
+            legacy_state=legacy,
+            principal_provider=_PrincipalProvider(),
+            build_authorizer=_Authorizer(),
+            retrieval_adapters={"decode": _Decode()},
+        )
+    )
+    await composition.mission_service.run(7)
+    ledger = await task_store.load(binding.root_trace_id)
+    root = ledger.tasks[ledger.root_task_id]
+    assert root.status == TaskStatus.BLOCKED
+    assert root.blocked_reason == "PHASE_ONE_CAPABILITY_BOUNDARY"
+    assert root.current_spec_version == 2
+    assert llm.decode_versions == [1, 2]
+    assert llm.replayed_dispatch
+    assert len(ledger.operations) == 3
+    assert llm.validated_artifact_kinds == ["evidence", "evidence", "direction"]
+    assert legacy.status == BuildStatus.PARTIAL
+    assert legacy.direction == "# accepted direction"
+    assert publications.state == PublicationState.PUBLISHED
+    assert bindings.active is not None
+    assert not hasattr(bindings.binding, "branch_id")
+
+    model_calls = llm.calls
+    reloaded_ledger = await FileSystemTaskStore(str(tmp_path / "ledger")).load(
+        binding.root_trace_id
+    )
+    assert reloaded_ledger.tasks[reloaded_ledger.root_task_id].status == TaskStatus.BLOCKED
+    reloaded_trace = await FileSystemTraceStore(str(tmp_path / "trace")).get_trace(
+        binding.root_trace_id
+    )
+    assert reloaded_trace is not None
+    submitted = next(
+        attempt for attempt in reloaded_ledger.attempts.values() if attempt.snapshot_id
+    )
+    assert await FileSystemArtifactStore(str(tmp_path / "artifacts")).get(
+        binding.root_trace_id,
+        submitted.snapshot_id,
+    )
+    assert llm.calls == model_calls

script_build_host/tests/test_repositories.py

@@ -0,0 +1,289 @@
+from __future__ import annotations
+
+import asyncio
+from datetime import UTC, datetime
+
+import pytest
+from agent.orchestration import ArtifactRef, EvidenceQuery
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+
+from script_build_host.agents.validation import ScriptBuildArtifactEvidenceReader
+from script_build_host.domain.artifacts import (
+    ArtifactState,
+    DirectionGoal,
+    EvidenceRecordV1,
+    ScriptDirectionArtifactV1,
+)
+from script_build_host.domain.errors import (
+    ArtifactAlreadyFrozen,
+    ArtifactDigestMismatch,
+    ArtifactNotFound,
+    ArtifactOwnershipMismatch,
+    InputHashConflict,
+)
+from script_build_host.domain.input_snapshot import ScriptBuildInput
+from script_build_host.domain.records import BuildStatus, PublicationState, PublicationType
+from script_build_host.repositories.legacy_state import SqlAlchemyLegacyBuildStateRepository
+from script_build_host.repositories.sqlalchemy import (
+    SqlAlchemyInputSnapshotRepository,
+    SqlAlchemyMissionBindingRepository,
+    SqlAlchemyPublicationRepository,
+    SqlAlchemyScriptBusinessArtifactRepository,
+)
+
+
+def _input(build_id: int = 1, *, topic_result: str = "result") -> ScriptBuildInput:
+    return ScriptBuildInput(
+        script_build_id=build_id,
+        execution_id=10,
+        topic_build_id=20,
+        topic_id=30,
+        topic={"topic": {"id": 30, "result": topic_result}, "points": [], "sources": []},
+        account={"account_name": "account", "source": "personal_config"},
+        prompt_manifest=({"biz_type": "planner", "content_sha256": "sha256:" + "0" * 64},),
+        datasource_manifest={"pattern": {"version": "fixture"}},
+        model_manifest={"planner": {"model": "fake"}},
+    )
+
+
+def _evidence() -> EvidenceRecordV1:
+    return EvidenceRecordV1(
+        evidence_id="evidence-1",
+        source_type="decode",
+        tool_name="search_script_decode_case",
+        query={"return_field": "summary", "top_k": 3},
+        source_refs=("decode://case/1",),
+        raw_artifact_ref=None,
+        summary="source-backed summary",
+        supports=("criterion-1",),
+        confidence="medium",
+        limitations=("fixture",),
+        content_sha256="",
+        created_at=datetime(2026, 7, 19, tzinfo=UTC),
+    )
+
+
+@pytest.mark.asyncio
+async def test_snapshot_freeze_is_idempotent_and_conflicts_by_version(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyInputSnapshotRepository(sessions)
+    first = await repository.freeze(_input())
+    second = await repository.freeze(_input())
+    assert first.snapshot_id == second.snapshot_id
+    assert first.canonical_sha256 == second.canonical_sha256
+    same_hash_new_version = await repository.freeze(_input(), version=2)
+    assert same_hash_new_version.snapshot_id == first.snapshot_id
+    with pytest.raises(InputHashConflict):
+        await repository.freeze(_input(topic_result="changed"))
+    with pytest.raises(Exception) as cross_build:
+        await repository.get(first.snapshot_id, script_build_id=999)
+    assert getattr(cross_build.value, "code", None) == "BUILD_NOT_FOUND"
+
+
+@pytest.mark.asyncio
+async def test_concurrent_same_snapshot_freeze_returns_one_version(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyInputSnapshotRepository(sessions)
+    first, second = await asyncio.gather(repository.freeze(_input()), repository.freeze(_input()))
+    assert first.snapshot_id == second.snapshot_id
+
+
+@pytest.mark.asyncio
+async def test_binding_create_is_idempotent_and_active_pointer_is_build_scoped(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyMissionBindingRepository(sessions)
+    values = {
+        "script_build_id": 1,
+        "root_trace_id": "root-1",
+        "input_snapshot_id": 2,
+        "engine_version": "0.4.0",
+        "schema_version": "phase-one/v1",
+    }
+    first, second = await asyncio.gather(repository.create(**values), repository.create(**values))
+    assert first.binding_id == second.binding_id
+    updated = await repository.set_active_direction(script_build_id=1, artifact_version_id=99)
+    assert updated.active_direction_artifact_version_id == 99
+
+
+@pytest.mark.asyncio
+async def test_artifact_freeze_ref_ownership_digest_and_spec_replay(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    version, reference = await repository.freeze(
+        script_build_id=1,
+        task_id="task-1",
+        attempt_id="attempt-1",
+        spec_version=1,
+        artifact=_evidence(),
+    )
+    assert version.state is ArtifactState.FROZEN
+    assert reference.uri == f"script-build://artifact-versions/{version.artifact_version_id}"
+    assert reference.kind == "evidence"
+    assert reference.version == str(version.artifact_version_id)
+    assert reference.digest and reference.digest.startswith("sha256:")
+    loaded = await repository.read_by_ref(
+        reference, script_build_id=1, task_id="task-1", attempt_id="attempt-1"
+    )
+    assert loaded == version
+    assert await repository.get_by_id(version.artifact_version_id, script_build_id=1) == version
+    with pytest.raises(ArtifactNotFound):
+        await repository.get_by_id(version.artifact_version_id, script_build_id=2)
+    with pytest.raises(ArtifactOwnershipMismatch):
+        await repository.read_by_ref(reference, script_build_id=2)
+    tampered = ArtifactRef(
+        uri=reference.uri,
+        kind=reference.kind,
+        version=reference.version,
+        digest="sha256:" + "f" * 64,
+    )
+    with pytest.raises(ArtifactDigestMismatch):
+        await repository.read_by_ref(tampered, script_build_id=1)
+    assert not await repository.verify_digest(tampered, script_build_id=1)
+    with pytest.raises(ArtifactAlreadyFrozen):
+        await repository.freeze(
+            script_build_id=1,
+            task_id="task-1",
+            attempt_id="attempt-1",
+            spec_version=2,
+            artifact=_evidence(),
+        )
+
+    class Bindings:
+        async def get_by_root(self, _root: str):
+            return type("Binding", (), {"script_build_id": 1})()
+
+    reader = ScriptBuildArtifactEvidenceReader(repository, Bindings())  # type: ignore[arg-type]
+    with pytest.raises(ArtifactOwnershipMismatch):
+        await reader.read(
+            reference,
+            EvidenceQuery(
+                root_trace_id="root-1",
+                task_id="other-task",
+                attempt_id="other-attempt",
+                snapshot_id="snapshot-1",
+                validation_id="validation-1",
+                query="evidence",
+                limit=1,
+            ),
+        )
+
+
+@pytest.mark.asyncio
+async def test_artifact_same_attempt_and_digest_is_concurrently_idempotent(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+
+    async def freeze() -> tuple[object, ArtifactRef]:
+        return await repository.freeze(
+            script_build_id=1,
+            task_id="task-1",
+            attempt_id="attempt-1",
+            spec_version=1,
+            artifact=_evidence(),
+        )
+
+    first, second = await asyncio.gather(freeze(), freeze())
+    assert first[1] == second[1]
+
+
+@pytest.mark.asyncio
+async def test_direction_and_publication_are_idempotent_and_error_is_redacted(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    artifacts = SqlAlchemyScriptBusinessArtifactRepository(sessions)
+    direction, _ = await artifacts.freeze(
+        script_build_id=1,
+        task_id="direction-task",
+        attempt_id="direction-attempt",
+        spec_version=1,
+        artifact=ScriptDirectionArtifactV1(
+            goals=(DirectionGoal("goal-1", "make the topic clear", "source-backed"),),
+            evidence_refs=("script-build://artifact-versions/1",),
+            legacy_markdown="# Direction",
+        ),
+    )
+    publications = SqlAlchemyPublicationRepository(sessions)
+    first = await publications.prepare(
+        script_build_id=1,
+        publication_type=PublicationType.DIRECTION,
+        accept_decision_id="decision-1",
+        artifact_version_id=direction.artifact_version_id,
+        expected_sha256=direction.canonical_sha256,
+    )
+    second = await publications.prepare(
+        script_build_id=1,
+        publication_type=PublicationType.DIRECTION,
+        accept_decision_id="decision-1",
+        artifact_version_id=direction.artifact_version_id,
+        expected_sha256=direction.canonical_sha256,
+    )
+    assert first.publication_id == second.publication_id
+    failed = await publications.mark_failed(
+        first.publication_id,
+        error_code="UPSTREAM",
+        error_summary="mysql://reader:password@db.example/app Bearer raw-token",
+    )
+    assert failed.state is PublicationState.FAILED
+    assert "password" not in (failed.last_error_summary or "")
+    assert "raw-token" not in (failed.last_error_summary or "")
+    published = await publications.mark_published(first.publication_id)
+    assert published.state is PublicationState.PUBLISHED
+    assert published.publication_revision == 1
+    replayed = await publications.mark_published(first.publication_id)
+    assert replayed.publication_revision == 1
+    still_published = await publications.mark_failed(
+        first.publication_id,
+        error_code="LATE_FAILURE",
+        error_summary="must not downgrade a committed publication",
+    )
+    assert still_published.state is PublicationState.PUBLISHED
+    assert still_published.publication_revision == 1
+    with pytest.raises(Exception, match="only an accepted direction"):
+        await publications.prepare(
+            script_build_id=1,
+            publication_type=PublicationType.FINAL,
+            accept_decision_id="decision-1",
+            artifact_version_id=direction.artifact_version_id,
+            expected_sha256=direction.canonical_sha256,
+        )
+    assert (
+        await artifacts.get_by_id(direction.artifact_version_id, script_build_id=1)
+    ).state is ArtifactState.PUBLISHED
+    observed = await publications.get_by_build(1, publication_type=PublicationType.DIRECTION)
+    assert observed == published
+
+
+@pytest.mark.asyncio
+async def test_legacy_status_read_write_and_direction_projection(
+    database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
+) -> None:
+    _, sessions = database
+    repository = SqlAlchemyLegacyBuildStateRepository(sessions)
+    build_id = await repository.create(
+        execution_id=10,
+        topic_build_id=20,
+        topic_id=30,
+        agent_type="AigcAgent",
+        agent_config={"model": "fake"},
+        data_source_url=None,
+        strategies_config={"always_on": [7], "on_demand": []},
+    )
+    assert await repository.get_status(build_id) is BuildStatus.RUNNING
+    await repository.set_status(build_id, BuildStatus.STOPPING)
+    assert await repository.get_status(build_id) is BuildStatus.STOPPING
+    await repository.project_direction(build_id, "# accepted direction")
+    await repository.set_status(build_id, BuildStatus.PARTIAL)
+    assert await repository.get_status(build_id) is BuildStatus.PARTIAL
+    with pytest.raises(Exception, match="cannot project script build success"):
+        await repository.set_status(build_id, BuildStatus.SUCCESS)