script-build-and-framework-tmp/script_build_host/tests/test_mission_service_boundaries.py

from __future__ import annotations

import asyncio
from types import SimpleNamespace

import pytest
from agent.orchestration import (
    ArtifactRef,
    DecisionAction,
    OperationStatus,
    TaskStatus,
    ValidationVerdict,
)

from script_build_host.application.mission_service import (
    BuildTransitionGate,
    DirectionReconciler,
    ScriptMissionService,
    StartScriptBuildCommand,
)
from script_build_host.domain.artifacts import (
    ArtifactKind,
    ArtifactState,
    DirectionArtifact,
    DirectionGoal,
)
from script_build_host.domain.errors import (
    DirectionProjectionConflict,
    InputRelationMismatch,
    ProtocolViolation,
)
from script_build_host.domain.records import BuildStatus, Principal, PublicationState


class _RejectingInputs:
    async def validate_source(self, **_: object) -> None:
        raise InputRelationMismatch()


class _SourceAuthorizer:
    async def require_source_access(self, *_args: object, **_kwargs: object) -> None:
        return None

    async def require_access(self, *_args: object, **_kwargs: object) -> None:
        return None


class _LegacyState:
    def __init__(self) -> None:
        self.created = 0

    async def create(self, **_: object) -> int:
        self.created += 1
        return 1


@pytest.mark.asyncio
async def test_relation_mismatch_produces_no_build_or_binding_write() -> None:
    legacy = _LegacyState()
    bindings = SimpleNamespace(created=0)
    service = ScriptMissionService(
        runner=SimpleNamespace(),
        coordinator=SimpleNamespace(),
        factory=SimpleNamespace(),
        input_snapshots=_RejectingInputs(),  # type: ignore[arg-type]
        bindings=bindings,
        legacy_state=legacy,  # type: ignore[arg-type]
        authorizer=_SourceAuthorizer(),
        direction_reconciler=SimpleNamespace(),
    )
    with pytest.raises(InputRelationMismatch):
        await service.start(
            StartScriptBuildCommand(
                execution_id=1,
                topic_build_id=2,
                topic_id=3,
                principal=Principal("tester"),
            )
        )
    assert legacy.created == 0
    assert bindings.created == 0


@pytest.mark.asyncio
async def test_source_authorization_runs_before_input_or_build_access() -> None:
    legacy = _LegacyState()
    inputs = SimpleNamespace(validated=False)

    class Denied:
        async def require_source_access(self, *_args: object, **_kwargs: object) -> None:
            raise PermissionError("source forbidden")

    service = ScriptMissionService(
        runner=SimpleNamespace(),
        coordinator=SimpleNamespace(),
        factory=SimpleNamespace(),
        input_snapshots=inputs,
        bindings=SimpleNamespace(),
        legacy_state=legacy,  # type: ignore[arg-type]
        authorizer=Denied(),  # type: ignore[arg-type]
        direction_reconciler=SimpleNamespace(),
    )
    with pytest.raises(PermissionError, match="source forbidden"):
        await service.start(StartScriptBuildCommand(1, 2, 3, Principal("intruder")))
    assert legacy.created == 0
    assert inputs.validated is False


class _StopState:
    def __init__(self, status: BuildStatus) -> None:
        self.status = status

    async def get_status(self, _build: int) -> BuildStatus:
        return self.status

    async def set_status(self, _build: int, status: BuildStatus, **_: object) -> None:
        self.status = status


@pytest.mark.asyncio
async def test_stop_waits_for_planner_even_when_there_are_no_operations() -> None:
    state = _StopState(BuildStatus.RUNNING)
    pending: asyncio.Future[None] = asyncio.get_running_loop().create_future()
    trace = SimpleNamespace(status="running")
    runner = SimpleNamespace(
        stop=lambda _root: _async_value(True),
        trace_store=SimpleNamespace(get_trace=lambda _root: _async_value(trace)),
    )
    coordinator = SimpleNamespace(
        task_store=SimpleNamespace(load=lambda _root: _async_value(SimpleNamespace(operations={})))
    )
    service = ScriptMissionService(
        runner=runner,
        coordinator=coordinator,
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(
            get_by_build=lambda _build: _async_value(SimpleNamespace(root_trace_id="root"))
        ),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=SimpleNamespace(),
        stop_timeout_seconds=0.01,
    )
    service._runs[7] = pending  # type: ignore[assignment]
    result = await service.stop(7, Principal("owner"))
    assert result.status == BuildStatus.STOPPING
    assert state.status == BuildStatus.STOPPING
    pending.cancel()


@pytest.mark.asyncio
async def test_stop_is_idempotent_and_publication_is_gated_by_durable_status() -> None:
    state = _StopState(BuildStatus.STOPPED)
    service = ScriptMissionService(
        runner=SimpleNamespace(),
        coordinator=SimpleNamespace(),
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=SimpleNamespace(),
    )
    result = await service.stop(7, Principal("owner"))
    assert result.status == BuildStatus.STOPPED

    state.status = BuildStatus.STOPPING
    publications = SimpleNamespace(prepared=False)
    reconciler = DirectionReconciler(
        coordinator=SimpleNamespace(),
        bindings=SimpleNamespace(),
        artifacts=SimpleNamespace(),
        publications=publications,
        legacy_state=state,
    )
    with pytest.raises(ProtocolViolation, match="forbidden while stopping"):
        await reconciler.reconcile(7, "root")
    assert publications.prepared is False


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("publication_state", "has_projection", "active"),
    [
        (PublicationState.PENDING, True, None),
        (PublicationState.FAILED, True, 3),
        (PublicationState.PUBLISHED, True, 3),
    ],
)
async def test_direction_reconciler_reenters_each_publication_crash_point(
    publication_state: PublicationState,
    has_projection: bool,
    active: int | None,
) -> None:
    ref = ArtifactRef(
        "script-build://artifact-versions/3",
        ArtifactKind.DIRECTION.value,
        "3",
        "sha256:" + "a" * 64,
    )
    direction = DirectionArtifact(
        goals=(
            DirectionGoal(
                "goal", "statement", "the statement is testable", success_criteria=("done",)
            ),
        ),
        evidence_refs=("script-build://artifact-versions/2",),
    )
    projected = direction.legacy_markdown if has_projection else None
    version = SimpleNamespace(
        artifact_version_id=3,
        canonical_sha256=ref.digest,
        state=(
            ArtifactState.PUBLISHED
            if publication_state is PublicationState.PUBLISHED
            else ArtifactState.FROZEN
        ),
        artifact=direction,
    )
    task = SimpleNamespace(
        task_id="direction-task",
        current_spec=SimpleNamespace(context_refs=("script-build://task-kinds/direction",)),
        status=TaskStatus.COMPLETED,
        decision_ids=["accept"],
    )
    attempt = SimpleNamespace(
        attempt_id="attempt",
        snapshot_id="snapshot",
        submission=SimpleNamespace(artifact_refs=(ref,)),
    )
    ledger = SimpleNamespace(
        tasks={task.task_id: task},
        decisions={
            "accept": SimpleNamespace(
                action=DecisionAction.ACCEPT,
                decision_id="accept",
                attempt_id="attempt",
                validation_id="validation",
            )
        },
        attempts={"attempt": attempt},
        validations={
            "validation": SimpleNamespace(
                attempt_id="attempt",
                verdict=ValidationVerdict.PASSED,
                snapshot_id="snapshot",
            )
        },
    )

    class Bindings:
        current = active

        async def get_by_build(self, _build: int):
            return SimpleNamespace(active_direction_artifact_version_id=self.current)

        async def set_active_direction(self, *, artifact_version_id: int, **_kwargs):
            self.current = artifact_version_id

    class Publications:
        state = publication_state
        published_calls = 0

        async def prepare(self, **_kwargs):
            return SimpleNamespace(publication_id=1, state=self.state)

        async def mark_published(self, _publication_id: int):
            self.state = PublicationState.PUBLISHED
            self.published_calls += 1

        async def mark_failed(self, *_args, **_kwargs):
            self.state = PublicationState.FAILED

    class State(_StopState):
        direction = projected

        async def get_direction(self, _build: int):
            return self.direction

        async def project_direction(self, _build: int, value: str):
            self.direction = value

    bindings = Bindings()
    publications = Publications()
    state = State(BuildStatus.PARTIAL)
    reconciler = DirectionReconciler(
        coordinator=SimpleNamespace(
            task_store=SimpleNamespace(load=lambda _root: _async_value(ledger))
        ),
        bindings=bindings,
        artifacts=SimpleNamespace(read_by_ref=lambda *_args, **_kwargs: _async_value(version)),
        publications=publications,
        legacy_state=state,
    )

    assert await reconciler.reconcile(7, "root") == 3
    assert state.direction == direction.legacy_markdown
    assert bindings.current == 3
    assert publications.state is PublicationState.PUBLISHED
    assert publications.published_calls == (
        0 if publication_state is PublicationState.PUBLISHED else 1
    )


@pytest.mark.asyncio
async def test_direction_reconciler_fails_closed_on_projection_conflict() -> None:
    ref = ArtifactRef(
        "script-build://artifact-versions/3",
        ArtifactKind.DIRECTION.value,
        "3",
        "sha256:" + "a" * 64,
    )
    direction = DirectionArtifact(
        goals=(
            DirectionGoal(
                "goal", "statement", "the statement is testable", success_criteria=("done",)
            ),
        ),
        evidence_refs=("script-build://artifact-versions/2",),
    )
    task = SimpleNamespace(
        task_id="direction-task",
        current_spec=SimpleNamespace(context_refs=("script-build://task-kinds/direction",)),
        status=TaskStatus.COMPLETED,
        decision_ids=["accept"],
    )
    ledger = SimpleNamespace(
        tasks={task.task_id: task},
        decisions={
            "accept": SimpleNamespace(
                action=DecisionAction.ACCEPT,
                decision_id="accept",
                attempt_id="attempt",
                validation_id="validation",
            )
        },
        attempts={
            "attempt": SimpleNamespace(
                attempt_id="attempt",
                snapshot_id="snapshot",
                submission=SimpleNamespace(artifact_refs=(ref,)),
            )
        },
        validations={
            "validation": SimpleNamespace(
                attempt_id="attempt",
                verdict=ValidationVerdict.PASSED,
                snapshot_id="snapshot",
            )
        },
    )

    class Publications:
        failed = False

        async def prepare(self, **_kwargs):
            return SimpleNamespace(publication_id=1, state=PublicationState.PENDING)

        async def mark_failed(self, *_args, **_kwargs):
            self.failed = True

    publications = Publications()
    reconciler = DirectionReconciler(
        coordinator=SimpleNamespace(
            task_store=SimpleNamespace(load=lambda _root: _async_value(ledger))
        ),
        bindings=SimpleNamespace(),
        artifacts=SimpleNamespace(
            read_by_ref=lambda *_args, **_kwargs: _async_value(
                SimpleNamespace(
                    artifact_version_id=3,
                    canonical_sha256=ref.digest,
                    state=ArtifactState.FROZEN,
                    artifact=direction,
                )
            )
        ),
        publications=publications,
        legacy_state=SimpleNamespace(
            get_status=lambda _build: _async_value(BuildStatus.PARTIAL),
            get_direction=lambda _build: _async_value("# conflicting direction"),
        ),
    )

    with pytest.raises(DirectionProjectionConflict):
        await reconciler.reconcile(7, "root")
    assert publications.failed is True


@pytest.mark.asyncio
@pytest.mark.parametrize("status", [BuildStatus.FAILED, BuildStatus.SUCCESS])
async def test_stop_does_not_rewrite_terminal_failed_or_success(status: BuildStatus) -> None:
    state = _StopState(status)
    service = ScriptMissionService(
        runner=SimpleNamespace(),
        coordinator=SimpleNamespace(),
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=SimpleNamespace(),
    )
    result = await service.stop(7, Principal("owner"))
    assert result.status is status
    assert state.status is status


@pytest.mark.asyncio
async def test_stop_intent_and_direction_reconcile_are_serialized() -> None:
    gate = BuildTransitionGate()
    entered = asyncio.Event()
    release = asyncio.Event()
    state = _StopState(BuildStatus.RUNNING)

    class BlockingReconciler:
        transition_gate = gate

        async def reconcile(self, script_build_id: int, _root: str) -> int:
            async with gate.hold(script_build_id):
                entered.set()
                await release.wait()
                assert state.status is BuildStatus.RUNNING
                return 1

    runner = SimpleNamespace(
        stop=lambda _root: _async_value(True),
        trace_store=SimpleNamespace(get_trace=lambda _root: _async_value(None)),
    )
    coordinator = SimpleNamespace(
        task_store=SimpleNamespace(load=lambda _root: _async_value(SimpleNamespace(operations={})))
    )
    service = ScriptMissionService(
        runner=runner,
        coordinator=coordinator,
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(
            get_by_build=lambda _build: _async_value(SimpleNamespace(root_trace_id="root"))
        ),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=BlockingReconciler(),  # type: ignore[arg-type]
        transition_gate=gate,
    )
    reconcile = asyncio.create_task(service.direction_reconciler.reconcile(7, "root"))
    await entered.wait()
    stop = asyncio.create_task(service.stop(7, Principal("owner")))
    await asyncio.sleep(0)
    assert state.status is BuildStatus.RUNNING
    release.set()
    assert await reconcile == 1
    assert (await stop).status is BuildStatus.STOPPED


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "initial_status",
    [BuildStatus.RUNNING, BuildStatus.STOPPING, BuildStatus.PARTIAL],
)
async def test_stop_closes_all_nonterminal_states_when_no_execution_remains(
    initial_status: BuildStatus,
) -> None:
    state = _StopState(initial_status)
    service = ScriptMissionService(
        runner=SimpleNamespace(
            stop=lambda _root: _async_value(True),
            trace_store=SimpleNamespace(get_trace=lambda _root: _async_value(None)),
        ),
        coordinator=SimpleNamespace(
            task_store=SimpleNamespace(
                load=lambda _root: _async_value(SimpleNamespace(operations={}))
            )
        ),
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(
            get_by_build=lambda _build: _async_value(SimpleNamespace(root_trace_id="root"))
        ),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=SimpleNamespace(),
    )

    result = await service.stop(7, Principal("owner"))

    assert result.status is BuildStatus.STOPPED
    assert state.status is BuildStatus.STOPPED


@pytest.mark.asyncio
async def test_stop_timeout_keeps_intent_and_reports_active_operation_ids() -> None:
    state = _StopState(BuildStatus.RUNNING)
    operation = SimpleNamespace(
        operation_id="operation-1",
        status=OperationStatus.RUNNING,
    )
    service = ScriptMissionService(
        runner=SimpleNamespace(
            stop=lambda _root: _async_value(True),
            trace_store=SimpleNamespace(get_trace=lambda _root: _async_value(None)),
        ),
        coordinator=SimpleNamespace(
            task_store=SimpleNamespace(
                load=lambda _root: _async_value(
                    SimpleNamespace(operations={"operation-1": operation})
                )
            ),
            stop_operation=lambda *_args, **_kwargs: _async_value(None),
            get_operation=lambda *_args, **_kwargs: _async_value(operation),
        ),
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(
            get_by_build=lambda _build: _async_value(SimpleNamespace(root_trace_id="root"))
        ),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=SimpleNamespace(),
        stop_timeout_seconds=0.01,
    )

    result = await service.stop(7, Principal("owner"))

    assert result.status is BuildStatus.STOPPING
    assert result.stopped_operation_ids == ("operation-1",)
    assert state.status is BuildStatus.STOPPING


@pytest.mark.asyncio
async def test_stop_with_missing_binding_records_terminal_failure() -> None:
    class RecordingState(_StopState):
        error_summary: str | None = None

        async def set_status(
            self,
            _build: int,
            status: BuildStatus,
            *,
            error_summary: str | None = None,
        ) -> None:
            self.status = status
            self.error_summary = error_summary

    state = RecordingState(BuildStatus.PARTIAL)

    async def missing_binding(_build: int):
        raise LookupError("missing")

    service = ScriptMissionService(
        runner=SimpleNamespace(),
        coordinator=SimpleNamespace(),
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(get_by_build=missing_binding),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=SimpleNamespace(),
    )

    with pytest.raises(LookupError, match="missing"):
        await service.stop(7, Principal("owner"))

    assert state.status is BuildStatus.FAILED
    assert state.error_summary == "MISSION_BINDING_MISSING"


async def _async_value(value):
    return value