From a599af7af7ba3255d537b95f4dfe80433984bd6e Mon Sep 17 00:00:00 2001 From: Alexander Schwartzman Date: Tue, 20 Oct 2026 16:52:41 -0400 Subject: [PATCH 1/3] test(runner): a conformance kit every [runner] must pass, run against LocalRunner --- AGENTS.md | 3 +- tst/runner/__init__.py | 0 tst/runner/conformance.py | 117 +++++++++++++++++++++++++++ tst/unit/runner/local_runner_test.py | 14 ++++ 4 files changed, 133 insertions(+), 1 deletion(-) create mode 100644 tst/runner/__init__.py create mode 100644 tst/runner/conformance.py diff --git a/AGENTS.md b/AGENTS.md index 4b0ce2dc..92b13e41 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..56c60d46 --- /dev/null +++ b/tst/runner/conformance.py @@ -0,0 +1,117 @@ +"""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 + + +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) + assert (record.status, record.run_id, record.instance_id, record.task_id) == ( + RunStatus.COMPLETED, handle.run_id, handle.instance_id, task.id) + assert run_store.get_run(handle.run_id).status is RunStatus.COMPLETED + 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) + assert record.status is RunStatus.FAILED and "conformance step needs ['never']" in record.error + assert run_store.get_run(handle.run_id).status is RunStatus.FAILED + + +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(0.05) + 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() From d0d8b212d60e0ab9c16a40db8e10e150f9659ff0 Mon Sep 17 00:00:00 2001 From: Alexander Schwartzman Date: Tue, 20 Oct 2026 16:59:38 -0400 Subject: [PATCH 2/3] test(runner): the kit checks the run index saved a failed run's error, and names its poll interval --- tst/runner/conformance.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tst/runner/conformance.py b/tst/runner/conformance.py index 56c60d46..a9ca21b8 100644 --- a/tst/runner/conformance.py +++ b/tst/runner/conformance.py @@ -17,6 +17,7 @@ from agent_env.task_step.task_step import TaskStep ENDED_WITHIN_SECONDS = 60 +POLL_SECONDS = 0.05 class Step(TaskStep): @@ -60,8 +61,9 @@ 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) - assert record.status is RunStatus.FAILED and "conformance step needs ['never']" in record.error - assert run_store.get_run(handle.run_id).status is RunStatus.FAILED + 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): @@ -113,5 +115,5 @@ 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(0.05) + await asyncio.sleep(POLL_SECONDS) return record From d06241a65f33545fc05143d107b38fff87e4f7cb Mon Sep 17 00:00:00 2001 From: Alexander Schwartzman Date: Tue, 20 Oct 2026 18:43:00 -0400 Subject: [PATCH 3/3] test(runner): validate saved run identity --- tst/runner/conformance.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tst/runner/conformance.py b/tst/runner/conformance.py index a9ca21b8..7b5d29d9 100644 --- a/tst/runner/conformance.py +++ b/tst/runner/conformance.py @@ -50,9 +50,10 @@ 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) - assert (record.status, record.run_id, record.instance_id, record.task_id) == ( - RunStatus.COMPLETED, handle.run_id, handle.instance_id, task.id) - assert run_store.get_run(handle.run_id).status is RunStatus.COMPLETED + 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)