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

from __future__ import annotations

from datetime import datetime, timedelta, timezone

import pytest
from agent.orchestration import (
    AgentRole,
    ArtifactRef,
    AttemptSubmission,
    CriterionResult,
    FrozenContextDocument,
    ResourceClaim,
    TaskStatus,
    ValidationVerdict,
)
from agent.orchestration.coordinator import TaskConflict
from agent.orchestration.models import ValidationEvidenceGrant, ValidationEvidenceSnapshot
from agent.orchestration.prompt_budget import PromptBudgetExceeded, RolePromptEnvelopePacker
from agent.orchestration.validation_evidence import (
    ValidationEvidenceConflict,
    ValidationEvidenceUnauthorized,
    complete_read,
    extend_snapshot,
    require_authorized_refs,
    require_complete,
    reserve_read,
    start_read_session,
)


def _grant(handle: str, uri: str, digest: str) -> ValidationEvidenceGrant:
    return ValidationEvidenceGrant(
        ArtifactRef(uri, "evidence", "1", digest),
        handle,
    )


def test_validation_snapshot_deduplicates_and_rejects_digest_conflict() -> None:
    original = _grant("handle-a", "evidence://a", "sha256:" + "1" * 64)
    snapshot = extend_snapshot(ValidationEvidenceSnapshot(), (original, original))
    assert snapshot.revision == 1
    assert snapshot.grants == (original,)
    with pytest.raises(ValidationEvidenceConflict):
        extend_snapshot(
            snapshot,
            (_grant("handle-b", "evidence://a", "sha256:" + "2" * 64),),
        )


def test_validation_snapshot_rejects_version_only_identity_conflict() -> None:
    first = ValidationEvidenceGrant(
        ArtifactRef("evidence://versioned", "evidence", "1"),
        "handle-v1",
    )
    second = ValidationEvidenceGrant(
        ArtifactRef("evidence://versioned", "evidence", "2"),
        "handle-v2",
    )
    snapshot = extend_snapshot(ValidationEvidenceSnapshot(), (first,))
    with pytest.raises(ValidationEvidenceConflict):
        extend_snapshot(snapshot, (second,))


def test_validation_submission_accepts_only_snapshot_refs() -> None:
    grant = _grant("handle-a", "evidence://a", "sha256:" + "1" * 64)
    snapshot = extend_snapshot(ValidationEvidenceSnapshot(), (grant,))
    require_authorized_refs(snapshot, (grant.artifact_ref,))
    with pytest.raises(ValidationEvidenceUnauthorized):
        require_authorized_refs(
            snapshot,
            (
                ArtifactRef(
                    "evidence://forged",
                    "evidence",
                    "1",
                    "sha256:" + "2" * 64,
                ),
            ),
        )


def test_validation_read_replay_is_core_session_semantics() -> None:
    snapshot = extend_snapshot(
        ValidationEvidenceSnapshot(),
        (_grant("handle-a", "evidence://a", "sha256:" + "1" * 64),),
    )
    session = start_read_session(snapshot, ("handle-a",))
    session, replay_result = reserve_read(
        session,
        evidence_revision=snapshot.revision,
        handle="handle-a",
        cursor=None,
    )
    assert replay_result is None
    session = complete_read(
        session,
        evidence_revision=snapshot.revision,
        handle="handle-a",
        cursor=None,
        next_cursor=None,
        exhausted=True,
        result={"content_fragment": "frozen", "exhausted": True},
    )
    require_complete(session)
    same, replay_result = reserve_read(
        session,
        evidence_revision=snapshot.revision,
        handle="handle-a",
        cursor=None,
    )
    assert replay_result == {"content_fragment": "frozen", "exhausted": True}
    assert same is session


def test_prompt_packer_counts_system_tools_history_and_payload() -> None:
    packer = RolePromptEnvelopePacker(token_limit=300, output_reserve=100)
    with pytest.raises(PromptBudgetExceeded):
        packer.pack(
            {"role_context": {"artifact": {"items": []}}},
            system_prompt="s" * 500,
            tool_schemas=({"name": "tool", "description": "x" * 500},),
            history=({"role": "user", "content": "h" * 500},),
        )


def test_prompt_packer_uses_the_same_final_envelope_for_planner_messages() -> None:
    packer = RolePromptEnvelopePacker(token_limit=2_000, output_reserve=100)
    packed = packer.pack_messages(
        (
            {"role": "system", "content": "planner policy"},
            {
                "role": "user",
                "content": (
                    '{"role_context":{"artifact":{"items":["body"]},'
                    '"context_bundle":{"cards":[{"id":"duplicate"}]}}}'
                ),
            },
        ),
        tool_schemas=({"name": "plan_script_tasks"},),
    )
    assert packed.packing_mode == "full"
    assert packed.deduplicated_segments == 1
    assert '"cards": []' in packed.messages[-1]["content"]


def test_resource_claim_requires_a_nonblank_canonical_identity() -> None:
    assert ResourceClaim("script-build://resources/workspace/a").exclusive
    with pytest.raises(ValueError):
        ResourceClaim(" ")


@pytest.mark.asyncio
async def test_agent_validation_cannot_submit_before_evidence_initialization(
    tmp_path,
) -> None:
    from test_coordinator_integration import (
        FakeExecutor,
        create_task,
        make_coordinator,
    )

    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
    task_id = await create_task(coordinator, "evidence must be initialized")
    created = await coordinator._create_attempt("root", task_id, "worker", None)
    ledger = await store.load("root")
    attempt = ledger.attempts[created["attempt_id"]]
    await coordinator.submit_attempt(
        {
            "role": AgentRole.WORKER.value,
            "root_trace_id": "root",
            "task_id": task_id,
            "attempt_id": attempt.attempt_id,
            "spec_version": attempt.spec_version,
            "trace_id": attempt.worker_trace_id,
            "tool_call_id": "submit-attempt",
        },
        AttemptSubmission("done", [ArtifactRef("memory://artifact", version="1")]),
    )
    ledger = await store.load("root")
    attempt = ledger.attempts[attempt.attempt_id]
    validation_id = await coordinator._start_validation(
        "root", task_id, attempt.attempt_id
    )
    ledger = await store.load("root")
    report = ledger.validations[validation_id]

    with pytest.raises(TaskConflict, match="evidence was not initialized"):
        await coordinator.submit_validation(
            {
                "role": AgentRole.VALIDATOR.value,
                "root_trace_id": "root",
                "task_id": task_id,
                "attempt_id": attempt.attempt_id,
                "validation_id": validation_id,
                "snapshot_id": report.snapshot_id,
                "trace_id": report.validator_trace_id,
                "tool_call_id": "premature-validation",
            },
            ValidationVerdict.PASSED,
            [CriterionResult("c1", ValidationVerdict.PASSED, "checked")],
            "passed",
            [],
            [],
            [],
            "accept",
        )


@pytest.mark.asyncio
async def test_durable_lease_renewal_preserves_epoch_and_rejects_stale_owner(
    tmp_path,
) -> None:
    from test_coordinator_integration import FakeExecutor, make_coordinator

    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
    reserved = await coordinator.reserve_durable_lease(
        "root", "retrieval:test", owner="worker-a", lease_seconds=30
    )
    await coordinator.renew_durable_lease(
        "root",
        "retrieval:test",
        owner="worker-a",
        epoch=reserved["epoch"],
        lease_seconds=60,
    )
    ledger = await store.load("root")
    renewed = ledger.durable_leases["retrieval:test"]
    assert renewed.epoch == reserved["epoch"]
    assert datetime.fromisoformat(renewed.expires_at) > datetime.now(
        timezone.utc
    ) + timedelta(seconds=50)
    with pytest.raises(TaskConflict, match="ownership is stale"):
        await coordinator.renew_durable_lease(
            "root",
            "retrieval:test",
            owner="worker-b",
            epoch=reserved["epoch"],
            lease_seconds=60,
        )


@pytest.mark.asyncio
async def test_host_contract_revision_and_document_freeze_are_one_mutation(
    tmp_path,
) -> None:
    from test_coordinator_integration import FakeExecutor, make_coordinator

    coordinator, store, _ = await make_coordinator(tmp_path, FakeExecutor([]))
    root_task_id = (await store.load("root")).root_task_id
    first = await coordinator.create_task_graph(
        "root",
        (
            {
                "_task_id": "container",
                "_parent_task_id": root_task_id,
                "objective": "open",
                "acceptance_criteria": (
                    {"criterion_id": "closed", "description": "closed", "hard": True},
                ),
                "context_refs": ("contract://v1",),
            },
        ),
        idempotency_key="open",
        context_documents=(
            FrozenContextDocument(
                "contract://v1",
                "sha256:" + "1" * 64,
                {"schema_version": "test/v1"},
            ),
        ),
    )
    assert first["tasks"][0]["task_id"] == "container"
    ledger = await store.load("root")
    ledger.tasks["container"].status = TaskStatus.NEEDS_REPLAN
    await store.commit(ledger, ledger.revision)

    await coordinator.create_task_graph(
        "root",
        (
            {
                "_task_id": "container",
                "_revision_task_id": "container",
                "_parent_task_id": root_task_id,
                "objective": "closed",
                "acceptance_criteria": (
                    {"criterion_id": "closed", "description": "closed", "hard": True},
                ),
                "context_refs": ("contract://v2",),
            },
        ),
        idempotency_key="close",
        context_documents=(
            FrozenContextDocument(
                "contract://v2",
                "sha256:" + "2" * 64,
                {"schema_version": "test/v2"},
            ),
        ),
    )
    revised = await store.load("root")
    task = revised.tasks["container"]
    assert task.status is TaskStatus.PENDING
    assert task.current_spec_version == 2
    assert task.current_spec.objective == "closed"
    assert "contract://v2" in revised.context_documents