feat(task): read recorded runs through task_instances and find_task_instance

.github/scripts/check_plugin_api.py

@@ -36,6 +36,9 @@
 USED = (
     "agent_env.task_step.context.TaskStepContext", "agent_env.plugins",
     "agent_env.providers.env_providers.env_provider.build_env_provider", "agent_env.env.store.register_env_instance",
+    # What a plugin reads of recorded runs.
+    "agent_env.task.store.task_instances", "agent_env.task.store.count_task_instances",
+    "agent_env.task.store.find_task_instance", "agent_env.task.store.TaskInstance",
     # What an environment provider reads to deploy a built-in env.
     "agent_env.env.envs.mcp_server.MCPServerEnv.docker_image_artifact", "agent_env.env.envs.mcp_server.MCPServerEnv.environment_name",
     "agent_env.env.envs.website.WebsiteEnv.backend_docker_image_artifact",

src/agent_env/explorer/routers/runs.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import asyncio
+import dataclasses
 import json
 import logging
 import uuid
@@ -20,8 +21,8 @@
 from agent_env.explorer.entity_ids import EntityId
 from agent_env.explorer.routers.common import PaginatedResponse, docs
 from agent_env.runner.runner import RunStatus
-from agent_env.store import Filter, Sort
-from agent_env.task.store import TaskStepStatus
+from agent_env.store import Filter
+from agent_env.task.store import TaskStepStatus, count_task_instances, find_task_instance, task_instances
 
 logger = logging.getLogger(__name__)
 
@@ -248,25 +249,18 @@ def list_instances(
     limit: int = Query(50, ge=1, le=500),
     offset: int = Query(0, ge=0),
 ) -> PaginatedResponse:
-    conditions = {"task_id": task_id}
-    if task_version is not None:
-        conditions["task_version"] = task_version
-    filt = Filter.of(**conditions)
-    store = docs()
-    items = store.query(TASK_INSTANCES_COLLECTION, filt,
-                        sort=Sort.by("created_at_utc", descending=True),
-                        limit=limit, offset=offset)
-    total = store.count(TASK_INSTANCES_COLLECTION, filt)
-    return PaginatedResponse(items=items, total=total, limit=limit, offset=offset,
+    items = task_instances(task_id, task_version=task_version, limit=limit, offset=offset)
+    total = count_task_instances(task_id, task_version=task_version)
+    return PaginatedResponse(items=[dataclasses.asdict(i) for i in items], total=total, limit=limit, offset=offset,
                              has_more=offset + len(items) < total)
 
 
 @router.get("/{task_id}/instances/{instance_id}")
 def get_instance(task_id: EntityId, instance_id: EntityId) -> dict:
-    doc = docs().find_one(TASK_INSTANCES_COLLECTION, Filter.of(instance_id=instance_id))
-    if doc is None:
+    instance = find_task_instance(instance_id)
+    if instance is None:
         raise HTTPException(status_code=404, detail=f"instance {instance_id} not found")
-    return doc
+    return dataclasses.asdict(instance)
 
 
 # --- run groups -------------------------------------------------------------

src/agent_env/task/store.py

@@ -16,7 +16,7 @@
 
 from agent_env.store.base import NotFoundError
 from agent_env.config import get_config
-from agent_env.store.document_store import DocumentStore, DuplicateKeyError, Filter, Lte, Ne, Sort, UpdateSpec, VersionedEntityStore, VersionedEntityStoreCache, compare_and_swap
+from agent_env.store.document_store import DocumentStore, DuplicateKeyError, Filter, Lte, Ne, Sort, SortKey, UpdateSpec, VersionedEntityStore, VersionedEntityStoreCache, compare_and_swap
 from agent_env.store.query import QueryBuilder, to_document_query
 from agent_env.task.step_journal import (
     _RESERVED_STEP_IDS,
@@ -257,6 +257,16 @@ def _new_instance_doc(
     }
 
 
+# A run records only the minute it started, so the instance id orders runs of one minute and pages never overlap.
+# Both keys descend so Mongo reads the sort from the (task_id, created_at_utc, instance_id) index scanned backward;
+# a mixed-direction sort matches no index and sorts every run's whole document in memory.
+_NEWEST_FIRST = Sort((SortKey("created_at_utc", descending=True), SortKey("instance_id", descending=True)))
+
+
+def _task_filter(task_id: str, task_version: int | None) -> Filter:
+    return Filter.of(task_id=task_id) if task_version is None else Filter.of(task_id=task_id, task_version=task_version)
+
+
 class TaskInstanceStore:
     def __init__(self) -> None:
         self._indexed = None
@@ -271,8 +281,8 @@ def _doc_store(self):
         if self._indexed is not store:
             store.ensure_index(TASK_INSTANCES_COLLECTION, ["instance_id"], unique=True)
             store.ensure_index(TASK_INSTANCES_COLLECTION, ["task_id"])
-            store.ensure_index(TASK_INSTANCES_COLLECTION, ["task_id", "created_at_utc"])
-            store.ensure_index(TASK_INSTANCES_COLLECTION, ["task_id", "task_version", "created_at_utc"])
+            store.ensure_index(TASK_INSTANCES_COLLECTION, ["task_id", "created_at_utc", "instance_id"])
+            store.ensure_index(TASK_INSTANCES_COLLECTION, ["task_id", "task_version", "created_at_utc", "instance_id"])
             store.ensure_index(TASK_STEP_JOURNAL_COLLECTION, ["instance_id", "step_id"], unique=True)
             store.ensure_index(TASK_STEP_JOURNAL_COLLECTION, ["instance_id", "seq"])
             self._indexed = store
@@ -375,6 +385,18 @@ def get(self, instance_id: str) -> TaskInstance:
             raise NotFoundError(f"TaskInstance '{instance_id}' not found")
         return TaskInstance.from_dict(doc)
 
+    def find(
+        self, task_id: str, *, task_version: int | None = None, limit: int | None = None, offset: int = 0,
+    ) -> list[TaskInstance]:
+        docs = self._doc_store.query(
+            TASK_INSTANCES_COLLECTION, _task_filter(task_id, task_version),
+            sort=_NEWEST_FIRST, limit=limit, offset=offset,
+        )
+        return [TaskInstance.from_dict(doc) for doc in docs]
+
+    def count(self, task_id: str, *, task_version: int | None = None) -> int:
+        return self._doc_store.count(TASK_INSTANCES_COLLECTION, _task_filter(task_id, task_version))
+
     def seed_context(self, instance_id: str, ops: "ContextUpdateOps") -> None:
         """Apply the run's initial context, then journal it as the replay base (even when empty)."""
         store = self._doc_store
@@ -763,6 +785,26 @@ def reset_task_instance_store() -> None:
     _task_instance_store = None
 
 
+def task_instances(
+    task_id: str, *, task_version: int | None = None, limit: int | None = None, offset: int = 0,
+) -> list[TaskInstance]:
+    """A task's recorded runs, newest first, from the configured document store. A run keeps only the minute it
+    started, so runs of one minute come in descending instance-id order, which keeps pages from overlapping."""
+    return get_task_instance_store().find(task_id, task_version=task_version, limit=limit, offset=offset)
+
+
+def count_task_instances(task_id: str, *, task_version: int | None = None) -> int:
+    return get_task_instance_store().count(task_id, task_version=task_version)
+
+
+def find_task_instance(instance_id: str) -> TaskInstance | None:
+    """One recorded run by its instance id, or None when there is no such run."""
+    try:
+        return get_task_instance_store().get(instance_id)
+    except NotFoundError:
+        return None
+
+
 def register_task_instance(
     task_id: str,
     task_version: int,

tst/unit/explorer/explorer_app_test.py

@@ -340,6 +340,29 @@ def test_a_bundle_runs_namespaced_instance_id_is_one_encoded_segment(client):
     assert len(client.get(f"/api/v1/task-instances/{instance}/conversations").json()["conversations"]) == 1
 
 
+def test_the_instance_routes_page_newest_first_and_answer_404_for_an_unknown_run(client):
+    task_id = "@local/~/triage/route-ticket"
+    store = get_config().get_document_store()
+    for minute in range(3):
+        for version in (1, 2):
+            store.insert("task_instances", {
+                "instance_id": f"{task_id}-v{version}m{minute}", "task_id": task_id, "task_version": version,
+                "status": "completed", "total_steps": 1, "current_step": 1, "completed_steps": [],
+                "created_at_utc": f"2026-10-08 10:0{minute} UTC", "journal_seq": 4,
+            })
+    base = f"/api/v1/tasks/{quote(task_id, safe='')}/instances"
+
+    page = client.get(base, params={"task_version": 2, "limit": 2}).json()
+    last = client.get(base, params={"offset": 5}).json()
+
+    assert [i["instance_id"] for i in page["items"]] == [f"{task_id}-v2m2", f"{task_id}-v2m1"]
+    assert (page["total"], page["has_more"]) == (3, True)
+    assert "journal_seq" not in page["items"][0]
+    assert (len(last["items"]), last["total"], last["has_more"]) == (1, 6, False)
+    assert client.get(f"{base}/{quote(f'{task_id}-v1m0', safe='')}").json()["task_version"] == 1
+    assert client.get(f"{base}/{quote(f'{task_id}-gone', safe='')}").status_code == 404
+
+
 def test_start_runs_caps_the_batch_size(client, monkeypatch):
     from agent_env.task import Task
 

tst/unit/store/test_task_instance_indexes.py

@@ -6,6 +6,10 @@
 which on large pass@k/batch tasks ran 60-300s and tripped ``socketTimeoutMS``.
 These tests pin that the store *declares* those compound indexes (via the
 backend-agnostic ``ensure_index``) so the regression can't silently come back.
+``task_instances`` also breaks ties within a minute by ``instance_id``, so each
+index ends in it: both sort keys descend and Mongo reads them from the index
+scanned backward, and the leading fields still serve a sort on
+``created_at_utc`` alone.
 
 Direction is no longer asserted: the store declares ASC compound indexes, which
 serve the DESC sort too (Mongo scans backward); a pre-existing DESC index is
@@ -33,12 +37,12 @@ def _declared_indexes(monkeypatch) -> list[tuple[list[str], bool]]:
 
 def test_task_instance_has_task_id_created_at_compound_index(monkeypatch):
     keys = [fields for fields, _ in _declared_indexes(monkeypatch)]
-    assert ["task_id", "created_at_utc"] in keys
+    assert ["task_id", "created_at_utc", "instance_id"] in keys
 
 
 def test_task_instance_has_task_id_version_created_at_compound_index(monkeypatch):
     keys = [fields for fields, _ in _declared_indexes(monkeypatch)]
-    assert ["task_id", "task_version", "created_at_utc"] in keys
+    assert ["task_id", "task_version", "created_at_utc", "instance_id"] in keys
 
 
 def test_task_instance_preserves_existing_indexes(monkeypatch):

tst/unit/task/task_instance_reads_test.py

@@ -0,0 +1,89 @@
+"""The recorded-run reads on the plugin surface: task_instances, count_task_instances and find_task_instance."""
+
+from __future__ import annotations
+
+import pytest
+
+from agent_env.config import set_document_store
+from agent_env.store import LocalSqliteDocumentStore
+from agent_env.task.store import (
+    TASK_INSTANCES_COLLECTION,
+    TaskInstance,
+    TaskStepStatus,
+    count_task_instances,
+    find_task_instance,
+    register_task_instance,
+    task_instances,
+)
+
+TASK = "@local/mycorp-demo/triage/route-ticket"
+
+
+@pytest.fixture
+def docs(tmp_path) -> LocalSqliteDocumentStore:
+    store = LocalSqliteDocumentStore(str(tmp_path / "runs.db"))
+    set_document_store(store)
+    return store
+
+
+def _run(docs, instance_id: str, created: str, *, task_id: str = TASK, version: int = 1, status: str = "completed"):
+    docs.insert(TASK_INSTANCES_COLLECTION, {
+        "instance_id": instance_id, "task_id": task_id, "task_version": version, "status": status,
+        "current_step": 1, "total_steps": 1, "created_at_utc": created,
+        "completed_steps": [{"step_id": "grade", "status": "success"}],
+        "context": {"metadata": {"verifications": {"grade": {"score": 1.0}}}},
+    })
+
+
+def test_a_tasks_runs_come_newest_first_and_only_that_tasks(docs):
+    _run(docs, "r1", "2026-10-08 09:00 UTC")
+    _run(docs, "r3", "2026-10-08 11:00 UTC")
+    _run(docs, "r2", "2026-10-08 10:00 UTC")
+    _run(docs, "other", "2026-10-08 12:00 UTC", task_id=f"{TASK}-v2")
+
+    runs = task_instances(TASK)
+
+    assert [r.instance_id for r in runs] == ["r3", "r2", "r1"]
+    assert all(isinstance(r, TaskInstance) for r in runs)
+    assert runs[0].completed_steps[0].status is TaskStepStatus.SUCCESS
+    assert runs[0].context["metadata"]["verifications"]["grade"]["score"] == 1.0
+    assert count_task_instances(TASK) == 3
+
+
+def test_a_version_filter_and_a_page(docs):
+    for minute in range(5):
+        _run(docs, f"v1-{minute}", f"2026-10-08 10:0{minute} UTC")
+        _run(docs, f"v2-{minute}", f"2026-10-08 10:0{minute} UTC", version=2)
+
+    assert [r.instance_id for r in task_instances(TASK, task_version=2, limit=2, offset=1)] == ["v2-3", "v2-2"]
+    assert count_task_instances(TASK, task_version=2) == 5
+    assert count_task_instances(TASK) == 10
+    assert task_instances(TASK, task_version=3) == []
+
+
+def test_one_run_by_id_or_none(docs):
+    _run(docs, "r1", "2026-10-08 09:00 UTC")
+
+    assert find_task_instance("r1").task_id == TASK
+    assert find_task_instance("missing") is None
+    assert task_instances("@local/mycorp-demo/triage/nothing") == []
+    assert count_task_instances("@local/mycorp-demo/triage/nothing") == 0
+
+
+def test_a_registered_run_reads_back(docs):
+    registered = register_task_instance(TASK, 4, total_steps=3)
+
+    found = find_task_instance(registered.instance_id)
+
+    assert (found.task_id, found.task_version, found.status, found.total_steps) == (TASK, 4, "running", 3)
+    assert [r.instance_id for r in task_instances(TASK)] == [registered.instance_id]
+
+
+def test_runs_of_one_minute_page_in_descending_instance_id_order_without_overlap(docs):
+    for instance_id in ("c", "a", "d", "b"):
+        _run(docs, instance_id, "2026-10-08 10:00 UTC")
+    _run(docs, "later", "2026-10-08 10:01 UTC")
+
+    pages = [task_instances(TASK, limit=2, offset=offset) for offset in (0, 2, 4)]
+
+    assert [[r.instance_id for r in page] for page in pages] == [["later", "d"], ["c", "b"], ["a"]]