fix(server): preserve completion cursors when deleting rollouts

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

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] = {}

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()
 
 

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