from __future__ import annotations

import asyncio
from datetime import UTC, datetime

import pytest
from agent.orchestration import ArtifactRef, EvidenceQuery
from sqlalchemy import select, update
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,
    InputDigestMismatch,
    InputHashConflict,
)
from script_build_host.domain.input_snapshot import ScriptBuildInput
from script_build_host.domain.records import BuildStatus, PublicationState, PublicationType
from script_build_host.infrastructure.legacy_tables import script_build_record
from script_build_host.infrastructure.tables import input_snapshot_table
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_snapshot_get_recomputes_digest_and_rejects_database_tampering(database) -> None:
    _, sessions = database
    repository = SqlAlchemyInputSnapshotRepository(sessions)
    frozen = await repository.freeze(_input())
    async with sessions() as session, session.begin():
        await session.execute(
            update(input_snapshot_table)
            .where(input_snapshot_table.c.id == int(frozen.snapshot_id))
            .values(canonical_json={**frozen.to_input().canonical_payload(), "topic_id": 999})
        )
    with pytest.raises(InputDigestMismatch):
        await repository.get(frozen.snapshot_id, script_build_id=1)


@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
    advanced = await repository.compare_and_set_input_snapshot(
        script_build_id=1,
        expected_snapshot_id=2,
        new_snapshot_id=3,
    )
    assert advanced.input_snapshot_id == 3
    replayed = await repository.compare_and_set_input_snapshot(
        script_build_id=1,
        expected_snapshot_id=2,
        new_snapshot_id=3,
    )
    assert replayed.input_snapshot_id == 3
    with pytest.raises(Exception, match="changed concurrently"):
        await repository.compare_and_set_input_snapshot(
            script_build_id=1,
            expected_snapshot_id=2,
            new_snapshot_id=4,
        )


@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
    await repository.set_checkpoint(
        build_id,
        checkpoint_code="PHASE_ONE_CAPABILITY_BOUNDARY",
        summary="direction ready",
        root_trace_id="root-1",
    )
    async with sessions() as session:
        row = (
            (
                await session.execute(
                    select(script_build_record).where(script_build_record.c.id == build_id)
                )
            )
            .mappings()
            .one()
        )
    assert row["status"] == "partial"
    assert row["error_message"] == "PHASE_ONE_CAPABILITY_BOUNDARY"
    assert row["summary"] == "direction ready"
    assert row["reson_trace_id"] == "root-1"
    assert row["end_time"] is not None
    await repository.set_status(build_id, BuildStatus.STOPPING)
    async with sessions() as session:
        stopping_row = (
            (
                await session.execute(
                    select(script_build_record).where(script_build_record.c.id == build_id)
                )
            )
            .mappings()
            .one()
        )
    assert stopping_row["status"] == "stopping"
    assert stopping_row["error_message"] is None
    assert stopping_row["summary"] is None
    assert stopping_row["end_time"] is None
    await repository.begin_phase(build_id)
    async with sessions() as session:
        row = (
            (
                await session.execute(
                    select(script_build_record).where(script_build_record.c.id == build_id)
                )
            )
            .mappings()
            .one()
        )
    assert row["status"] == "running"
    assert row["end_time"] is None
    assert row["summary"] is None
    with pytest.raises(Exception, match="cannot project script build success"):
        await repository.set_status(build_id, BuildStatus.SUCCESS)