From 2cc67590fdacd4cf09b442c15addc34badce24d5 Mon Sep 17 00:00:00 2001 From: yixi ning Date: Sat, 17 Oct 2026 08:31:42 +0000 Subject: [PATCH] fix(server): preserve completion cursors when deleting rollouts --- agentlightning/server/routes/rollouts.py | 8 +++++- agentlightning/server/store.py | 4 ++- tests/server/conftest.py | 4 ++- tests/server/test_endpoints.py | 36 ++++++++++++++++++++++++ 4 files changed, 49 insertions(+), 3 deletions(-) diff --git a/agentlightning/server/routes/rollouts.py b/agentlightning/server/routes/rollouts.py index f4598ef04..0399e4319 100644 --- a/agentlightning/server/routes/rollouts.py +++ b/agentlightning/server/routes/rollouts.py @@ -23,7 +23,7 @@ RolloutPatch, RolloutState, ) -from agentlightning.server.store import _events, _rollouts, _terminal_order +from agentlightning.server.store import _events, _rollouts, _terminal_index, _terminal_order router = APIRouter(tags=["rollouts"]) @@ -157,6 +157,8 @@ async def list_terminal_rollouts(after: int = 0, limit: int = 1000) -> TerminalR slice_ids = _terminal_order[after : after + limit] items: list[TerminalRolloutItem] = [] for rid in slice_ids: + if rid is None: + continue rollout = _rollouts.get(rid) if rollout is None: continue @@ -209,6 +211,7 @@ async def patch_rollout(rollout_id: str, body: RolloutPatch) -> Rollout: _rollouts[rollout_id] = updated if "state" in updates and updated_status.state in TERMINAL_STATES: # One-way terminal transition (guarded above) => append exactly once. + _terminal_index[rollout_id] = len(_terminal_order) _terminal_order.append(rollout_id) return updated @@ -218,3 +221,6 @@ async def delete_rollout(rollout_id: str) -> None: """Delete a rollout and its events. Idempotent: missing id is a no-op.""" _rollouts.pop(rollout_id, None) _events.pop(rollout_id, None) + index = _terminal_index.pop(rollout_id, None) + if index is not None: + _terminal_order[index] = None diff --git a/agentlightning/server/store.py b/agentlightning/server/store.py index fcdb3a9f8..e3ed43f25 100644 --- a/agentlightning/server/store.py +++ b/agentlightning/server/store.py @@ -15,4 +15,6 @@ _models: dict[str, dict[str, Model]] = {} # Completion-ordered ids enable cursor pagination without rescanning rollouts. -_terminal_order: list[str] = [] +# Deleted entries become None so existing cursor positions stay valid. +_terminal_order: list[str | None] = [] +_terminal_index: dict[str, int] = {} diff --git a/tests/server/conftest.py b/tests/server/conftest.py index 00f9e2dd9..58b3ffde8 100644 --- a/tests/server/conftest.py +++ b/tests/server/conftest.py @@ -12,7 +12,7 @@ from fastapi.testclient import TestClient from agentlightning.server.app import create_app -from agentlightning.server.store import _events, _models, _rollouts, _terminal_order +from agentlightning.server.store import _events, _models, _rollouts, _terminal_index, _terminal_order AGL_KEY = "test-secret-key" MODEL_NAME = "test-model" @@ -23,11 +23,13 @@ def clean_store() -> Iterator[None]: _rollouts.clear() _events.clear() _models.clear() + _terminal_index.clear() _terminal_order.clear() yield _rollouts.clear() _events.clear() _models.clear() + _terminal_index.clear() _terminal_order.clear() diff --git a/tests/server/test_endpoints.py b/tests/server/test_endpoints.py index 55ad9f549..8ae736d8e 100644 --- a/tests/server/test_endpoints.py +++ b/tests/server/test_endpoints.py @@ -486,3 +486,39 @@ def test_terminal_rollouts_cursor_pagination(client: TestClient, auth_headers: d page3 = client.get("/api/rollouts/terminal", params={"after": 3}, headers=auth_headers) assert page3.json() == {"items": [], "next_after": 3, "total_terminal": 3} assert pending not in {it["rollout_id"] for it in body1["items"] + body2["items"]} + + +def test_terminal_rollouts_do_not_replay_deleted_ids(client: TestClient, auth_headers: dict[str, str]): + retired = _terminal(client, auth_headers, "old", is_train=True) + survivor = _terminal(client, auth_headers, "survivor", is_train=True) + first_page = client.get("/api/rollouts/terminal", params={"limit": 1}, headers=auth_headers).json() + cursor = first_page["next_after"] + assert cursor == 1 + + assert client.delete(f"/api/rollouts/{retired}", headers=auth_headers).status_code == 204 + after_delete = client.get("/api/rollouts/terminal", params={"after": cursor}, headers=auth_headers).json() + assert [item["rollout_id"] for item in after_delete["items"]] == [survivor] + recreated = client.post( + "/api/rollouts", + json=[{"rollout_id": retired, "input": {"data_id": "new"}}], + headers=auth_headers, + ) + assert recreated.status_code == 201 + assert recreated.json()[0]["status"]["state"] == "queuing" + + # A new rollout with a deleted ID has not completed yet. + pending_page = client.get("/api/rollouts/terminal", params={"limit": 1}, headers=auth_headers).json() + assert pending_page == {"items": [], "next_after": 1, "total_terminal": 2} + + for state in ("running", "succeeded"): + patched = client.patch(f"/api/rollouts/{retired}", json={"status": {"state": state}}, headers=auth_headers) + assert patched.status_code == 200 + + completed = client.get("/api/rollouts/terminal", headers=auth_headers).json() + assert [item["rollout_id"] for item in completed["items"]] == [survivor, retired] + assert completed["items"][-1]["data_id"] == "new" + assert completed["next_after"] == 3 + + # A cursor obtained before deletion must still see the following completions. + remaining = client.get("/api/rollouts/terminal", params={"after": cursor}, headers=auth_headers).json() + assert remaining == completed