script-build-and-framework-tmp/agent/tests/test_orchestration_store.py

import json

import pytest

from agent.orchestration.models import (
    ArtifactRef,
    AttemptSubmission,
    TaskLedger,
    TaskStatus,
)
from agent.orchestration.state_machine import InvalidTaskTransition, transition
from agent.orchestration.store import (
    FileSystemArtifactStore,
    FileSystemTaskStore,
    RevisionConflict,
)


@pytest.mark.asyncio
async def test_ledger_atomic_revision_and_roundtrip(tmp_path):
    store = FileSystemTaskStore(str(tmp_path))
    ledger = TaskLedger(root_trace_id="root", mission="mission")
    committed = await store.commit(ledger, expected_revision=-1)
    assert committed.revision == 0
    loaded = await store.load("root")
    assert loaded.mission == "mission"
    assert loaded.revision == 0

    stale = TaskLedger(root_trace_id="root", mission="stale", revision=-1)
    with pytest.raises(RevisionConflict):
        await store.commit(stale, expected_revision=-1)

    on_disk = json.loads((tmp_path / "root" / "orchestration" / "ledger.json").read_text())
    assert on_disk["revision"] == 0


@pytest.mark.asyncio
async def test_snapshot_is_immutable_and_hash_is_stable(tmp_path):
    store = FileSystemArtifactStore(str(tmp_path)).for_root("root")
    submission = AttemptSubmission(
        summary=" delivered ",
        artifact_refs=[ArtifactRef(uri="file:///result", version="v1")],
        evidence_refs=[ArtifactRef(uri="db://row/1", digest="abc")],
    )
    first = await store.freeze("attempt-1", submission)
    second = await store.freeze("attempt-2", submission)
    assert first.sha256 == second.sha256
    assert (await store.get(first.snapshot_id)).normalized_content["summary"] == "delivered"

    with pytest.raises(ValueError, match="version or digest"):
        ArtifactRef(uri="file:///mutable")


def test_state_machine_rejects_shortcut_to_completed():
    assert transition(TaskStatus.PENDING, TaskStatus.RUNNING) == TaskStatus.RUNNING
    with pytest.raises(InvalidTaskTransition):
        transition(TaskStatus.RUNNING, TaskStatus.COMPLETED)