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)