diff --git a/AGENTS.md b/AGENTS.md index 25ae1d16..48debac0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -227,7 +227,8 @@ orchestrators call it the same way. What callers rely on: `default_a2a_agent`, `mcp_server_sources`, `gnu_stat`). CI rejects any other skip reason; a broken configuration must fail loudly, not skip. - The suite is backend-agnostic: a test about one particular backend lives with that backend, not here. A custom store must pass `tst/store/{conformance,object_conformance,image_conformance,secret_conformance}.py`, - the same kits the built-ins pass. + the same kits the built-ins pass, and a custom `[runner]` must pass `tst/runner/conformance.py`, which + `LocalRunner` passes. - Shared helpers live in `tst/util/` (`config.py` installs a config document, `a2a_test_agent.py` builds the echo A2A agent, `capabilities.py`, `image_cache.py`); fixtures and mock services in `tst/data/`. Nothing test-only ships under `src/`. diff --git a/tst/runner/__init__.py b/tst/runner/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tst/runner/conformance.py b/tst/runner/conformance.py new file mode 100644 index 00000000..7b5d29d9 --- /dev/null +++ b/tst/runner/conformance.py @@ -0,0 +1,120 @@ +"""Backend-neutral Runner conformance: what ``task run``, ``eval run`` and the explorer rely on from any ``[runner]``. + +Each case is ``await case(runner)`` on a runner the caller never started, as the CLI holds one, and stops after. +The cases store their tasks with ``Task.put`` and read the run index and the task instances, so the runner's worker +must share the caller's stores and have ``Step`` in its task-step registry. Every runner must pass every case in +``CASES``. +""" + +import asyncio +import time +import uuid + +from agent_env.runner import store as run_store +from agent_env.runner.runner import RunStatus +from agent_env.task import Task +from agent_env.task.store import find_task_instance +from agent_env.task_step.task_step import TaskStep + +ENDED_WITHIN_SECONDS = 60 +POLL_SECONDS = 0.05 + + +class Step(TaskStep): + """Naps ``nap`` seconds, fails unless the run's metadata sets each key in ``needs``, and records that it ran + under its id.""" + + type = "runner_conformance" + entity_refs = () + + def __init__(self, nap=0, needs=(), **base): + super().__init__(**base) + self.nap, self.needs = nap, list(needs) + + @classmethod + def from_dict(cls, data): + return cls(**cls._base_from_dict(data), nap=data.get("nap", 0), needs=data.get("needs", [])) + + def to_dict(self): + return {**super().to_dict(), "nap": self.nap, "needs": self.needs} + + async def execute(self, context): + await asyncio.sleep(self.nap) + if missing := [key for key in self.needs if not context.metadata.get(key)]: + raise RuntimeError(f"conformance step needs {missing}") + context.metadata[self.id] = True + return context + + +async def a_run_completes_and_is_recorded_under_its_handle(runner): + task = _put(Step(id="a", version=None)) + handle = await runner.submit(task.id, task.version, metadata={"carried": "yes"}) + record = await _until_ended(runner, handle.run_id) + saved = run_store.get_run(handle.run_id) + for record in (record, saved): + assert (record.status, record.run_id, record.instance_id, record.task_id) == ( + RunStatus.COMPLETED, handle.run_id, handle.instance_id, task.id) + metadata = find_task_instance(handle.instance_id).context["metadata"] + assert (metadata["workflow_id"], metadata["carried"], metadata["a"]) == (handle.run_id, "yes", True) + + +async def a_run_whose_step_fails_fails_with_its_error(runner): + task = _put(Step(id="a", version=None, needs=["never"])) + handle = await runner.submit(task.id, task.version) + record = await _until_ended(runner, handle.run_id) + saved = run_store.get_run(handle.run_id) + for record in (record, saved): + assert record.status is RunStatus.FAILED and "conformance step needs ['never']" in record.error + + +async def a_cancelled_run_ends_canceled(runner): + task = _put(Step(id="a", version=None, nap=10 * ENDED_WITHIN_SECONDS)) + handle = await runner.submit(task.id, task.version) + await _until(runner, handle.run_id, lambda record: record.status is not RunStatus.QUEUED) + assert await runner.cancel(handle.run_id) is True + assert (await _until_ended(runner, handle.run_id)).status is RunStatus.CANCELED + assert run_store.get_run(handle.run_id).status is RunStatus.CANCELED + + +async def a_run_given_an_instance_continues_it_from_start_step(runner): + task = _put(Step(id="a", version=None, needs=["first"]), Step(id="b", version=None, needs=["a", "resumed"])) + first = await runner.submit(task.id, task.version, metadata={"first": True}) + assert (await _until_ended(runner, first.run_id)).status is RunStatus.FAILED + handle = await runner.submit(task.id, task.version, instance_id=first.instance_id, + metadata={"start_step": 1, "first": False, "resumed": True}) + assert handle.instance_id == first.instance_id + assert (await _until_ended(runner, handle.run_id)).status is RunStatus.COMPLETED + + +async def a_run_it_does_not_have_or_that_ended_cannot_be_cancelled(runner): + task = _put(Step(id="a", version=None)) + handle = await runner.submit(task.id, task.version) + await _until_ended(runner, handle.run_id) + assert await runner.cancel(handle.run_id) is False + assert await runner.cancel(f"no-such-run-{uuid.uuid4().hex}") is False + assert await runner.status(f"no-such-run-{uuid.uuid4().hex}") is None + + +CASES = [ + a_run_completes_and_is_recorded_under_its_handle, + a_run_whose_step_fails_fails_with_its_error, + a_cancelled_run_ends_canceled, + a_run_given_an_instance_continues_it_from_start_step, + a_run_it_does_not_have_or_that_ended_cannot_be_cancelled, +] + + +def _put(*steps) -> Task: + return Task.put(id=f"runner-conformance-{uuid.uuid4().hex[:12]}", steps=list(steps)) + + +async def _until_ended(runner, run_id): + return await _until(runner, run_id, lambda record: record.status.is_terminal) + + +async def _until(runner, run_id, done): + deadline = time.monotonic() + ENDED_WITHIN_SECONDS + while (record := await runner.status(run_id)) is None or not done(record): + assert time.monotonic() < deadline, f"run {run_id} is still {record and record.status}" + await asyncio.sleep(POLL_SECONDS) + return record diff --git a/tst/unit/runner/local_runner_test.py b/tst/unit/runner/local_runner_test.py index 59d8219a..1b94fafe 100644 --- a/tst/unit/runner/local_runner_test.py +++ b/tst/unit/runner/local_runner_test.py @@ -13,10 +13,12 @@ import pytest from agent_env.config import configure, reset_config, set_document_store +from agent_env.config.runtime import Config from agent_env.runner import store as run_store from agent_env.runner.local_runner import LocalRunner from agent_env.runner.runner import KEEP_KEY, RunRecord, RunStatus from agent_env.store.document_store.sqlite_document_store import LocalSqliteDocumentStore +from tst.runner import conformance @pytest.fixture @@ -680,3 +682,15 @@ async def stubborn(_): assert f"Gave up tearing down {handle.run_id}" in caplog.text await asyncio.sleep(1.1) # let the abandoned run end before the loop closes + + +@pytest.mark.asyncio +@pytest.mark.parametrize("case", conformance.CASES, ids=lambda case: case.__name__) +async def test_conformance(case, local_stores, monkeypatch): + steps = {**Config().task_step_registry(), conformance.Step.type: conformance.Step} + monkeypatch.setattr(Config, "task_step_registry", lambda self: steps) + runner = LocalRunner() + try: + await case(runner) + finally: + await runner.stop()