feat(token-capture): retire and delete external-staging capture ledgers

fern/versions/latest/pages/model-server/model-call-capture.mdx

@@ -293,6 +293,55 @@ digest-covered `media` summary. Deploy the paired RL changes with this adapter;
 checkpoints written in either older format need a separate migration. Supported
 processors and modalities are enforced by the framework's inference worker.
 
+## Retire capture ledgers
+
+With external staging, the model server keeps a capture ledger for each rollout: one `<rollout_id>.lineage.jsonl` file in the lineage store's directory, recording the model calls the rollout made without their token arrays.
+The training framework reads it once, through the manifest route, when the rollout ends.
+Gym can't tell when the framework is done with a rollout, so the framework removes each ledger itself, using the same two operations the complete-record token store has: retire and delete.
+
+### Retire a rollout the framework is done with
+
+Retiring deletes the rollout's ledger and leaves an empty `<rollout_id>.lineage.retired` file in its place.
+That file acts as a fence: if a model call for the rollout finishes later, Gym discards its row instead of starting a new ledger.
+Late calls are more common than they sound, because an agent the framework gave up on can keep running, and a client retry can run a finished rollout a second time.
+`TokenSource.drop` leaves the same kind of fence in the complete-record token store.
+After a rollout is retired, its manifest route returns HTTP 410 instead of an empty manifest, so a framework that reads it too late finds out rather than building an empty receipt.
+
+A call that starts after its rollout was retired is not admitted for capture, so it stages nothing.
+A call that was already in flight, however, may have staged its tokens before Gym discarded its row, and no manifest names those tokens.
+Gym logs a warning with the call's staging key, and the framework that staged the tokens has to remove them, for example by sweeping staged rows that no receipt references.
+
+Retire every rollout the framework is done with, whether it finished or was abandoned.
+A finished rollout can be retired once its manifest has been read and its receipt built, unless a crash before the receipt is saved would make the framework rerun it under the same ID; in that case, wait until the receipt is saved.
+An abandoned rollout should be retired as soon as the framework gives up on it.
+
+### Delete fences once nothing can still write
+
+Deleting removes a rollout's ledger together with its fence, which is only safe once no call from that rollout can still arrive.
+In practice that means the end of the run, or a point where every Gym server sharing the lineage directory has restarted.
+Until then, the fences accumulate at one empty file per retired rollout.
+Also delete a rollout ID before reusing it, so that the new run starts from an empty ledger, as `TokenCaptureStore.delete` does for the complete-record store.
+
+### Send the requests
+
+Send each request to a model server whose lineage directory holds the rollout's ledger, either the server that served the rollout or any other server configured with the same directory.
+A model server only sees the ledgers in its own lineage directory and reports any other rollout as `absent`.
+
+```http
+POST /training-token-capture/control/rollouts/retire
+POST /training-token-capture/control/rollouts/delete
+Authorization: Bearer <control auth token>
+
+{"rollout_ids": ["<rollout_id>", ...]}
+```
+
+Each request takes at most 4096 rollout IDs, and `RolloutControlClient.retire` and `RolloutControlClient.delete` split longer lists into batches for you.
+The reply lists the rollouts whose ledger was removed under `removed` and the rest under `absent`.
+Because repeating either operation changes nothing, a failed batch can simply be sent again.
+
+Neither operation touches staged token data.
+The framework that staged the data removes it, and it must keep any staged data that a checkpoint it may restore still refers to.
+
 ## Limitations
 
 The model-server boundary observes model HTTP requests and responses. It can record tool calls,

nemo_gym/base_responses_api_model.py

@@ -1883,7 +1883,7 @@ def install_model_call_capture(
         if not isinstance(lineage_store, CaptureLedger):
             raise ValueError(
                 "token_id_capture.external_staging requires the lineage store to implement "
-                "the CaptureLedger protocol (record, record_failure, manifest, has_rows)"
+                "the CaptureLedger protocol (record, record_failure, manifest, has_rows, retire, delete)"
             )
         capture_ledger: CaptureLedger = lineage_store
         install_rollout_control_routes(

nemo_gym/token_id_capture/control_routes.py

@@ -1,26 +1,40 @@
 # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-"""Expose rollout capture manifests over HTTP.
+"""Expose rollout capture manifests and ledger removal over HTTP.
 
 The training framework reads one manifest for each rollout.
 The endpoint returns call metadata and does not read staged token data.
 The framework builds receipts and removes staged data after use.
+It then retires the rollout's ledger, which Gym cannot otherwise know is no longer needed.
 """
 
 from __future__ import annotations
 
 import asyncio
 import hmac
+from collections.abc import Sequence
 from typing import Any
 
 from fastapi import APIRouter, Header, HTTPException
+from pydantic import BaseModel, ConfigDict, Field
 
-from nemo_gym.token_id_capture.protocols import CaptureLedger
-from nemo_gym.token_id_capture.staging.records import RolloutManifest
+from nemo_gym.token_id_capture.protocols import CaptureLedger, RolloutRemovalPayload, RolloutRetiredError
+from nemo_gym.token_id_capture.staging.records import RolloutManifest, RolloutRemoval
 
 
 CONTROL_ROUTE_PREFIX = "/training-token-capture/control"
+# Bounds one retire or delete request so it finishes well within a control timeout.
+# The client splits larger batches.
+MAX_LEDGER_BATCH = 4096
+
+
+class LedgerBatchRequest(BaseModel):
+    """Body of the retire and delete routes."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    rollout_ids: list[str] = Field(min_length=1, max_length=MAX_LEDGER_BATCH)
 
 
 def install_rollout_control_routes(
@@ -29,7 +43,7 @@ def install_rollout_control_routes(
     *,
     auth_token: str,
 ) -> None:
-    """Install the bearer-protected manifest route."""
+    """Install the bearer-protected manifest, retire, and delete routes."""
     if not auth_token:
         raise ValueError("rollout control routes require a non-empty auth token")
     expected = f"Bearer {auth_token}"
@@ -50,14 +64,39 @@ async def rollout_manifest(
         check_auth(authorization)
         try:
             return await lineage_store.manifest(rollout_id)
+        except RolloutRetiredError as error:
+            # Gone, not empty: an empty manifest would read as a rollout that made no calls.
+            raise HTTPException(status_code=410, detail=str(error)) from error
+        except ValueError as error:
+            raise HTTPException(status_code=400, detail=str(error)) from error
+
+    @router.post("/rollouts/retire")
+    async def retire_rollouts(
+        body: LedgerBatchRequest,
+        authorization: str | None = Header(default=None),
+    ) -> RolloutRemovalPayload:
+        check_auth(authorization)
+        try:
+            return await lineage_store.retire(body.rollout_ids)
+        except ValueError as error:
+            raise HTTPException(status_code=400, detail=str(error)) from error
+
+    @router.post("/rollouts/delete")
+    async def delete_rollouts(
+        body: LedgerBatchRequest,
+        authorization: str | None = Header(default=None),
+    ) -> RolloutRemovalPayload:
+        check_auth(authorization)
+        try:
+            return await lineage_store.delete(body.rollout_ids)
         except ValueError as error:
             raise HTTPException(status_code=400, detail=str(error)) from error
 
     app.include_router(router)
 
 
 class RolloutControlClient:
-    """Framework-side client for the read-only manifest route."""
+    """Framework-side client for the manifest, retire, and delete routes."""
 
     def __init__(
         self,
@@ -73,12 +112,50 @@ def __init__(
         self._request_timeout_s = request_timeout_s
 
     async def manifest(self, rollout_id: str) -> RolloutManifest:
+        """Fetch a rollout's manifest. A retired rollout raises ``RolloutRetiredError``."""
+        from nemo_gym.token_id_capture.store import validate_rollout_id
+
+        # Validate before building the URL: dot segments are normalized, so "../rollouts/r1" would fetch r1.
+        validate_rollout_id(rollout_id)
         response = await self._request("GET", f"/rollouts/{rollout_id}/manifest")
+        if response.status == 410:
+            raise RolloutRetiredError(f"rollout {rollout_id} is retired: {await response.text()}")
         if response.status != 200:
             raise RuntimeError(
                 f"rollout {rollout_id} manifest fetch failed: HTTP {response.status} {await response.text()}"
             )
-        return RolloutManifest.model_validate(await response.json())
+        manifest = RolloutManifest.model_validate(await response.json())
+        if manifest.rollout_id != rollout_id:
+            raise ValueError(f"asked for the manifest of rollout {rollout_id} but got rollout {manifest.rollout_id}")
+        return manifest
+
+    async def retire(self, rollout_ids: Sequence[str]) -> RolloutRemoval:
+        """Retire ledgers; see ``CaptureLedger.retire``. Retrying after a failure is safe."""
+        return await self._remove("retire", rollout_ids)
+
+    async def delete(self, rollout_ids: Sequence[str]) -> RolloutRemoval:
+        """Delete ledgers and their fences; see ``CaptureLedger.delete``. Retrying after a failure is safe."""
+        return await self._remove("delete", rollout_ids)
+
+    async def _remove(self, action: str, rollout_ids: Sequence[str]) -> RolloutRemoval:
+        from nemo_gym.token_id_capture.store import validate_rollout_ids
+
+        # Validate and deduplicate the whole list before the first request. Batch by batch, a bare string
+        # would become one-character IDs, an invalid ID in a later batch would fail after earlier batches
+        # had changed state, and an ID repeated across batches would be reported both removed and absent.
+        rollout_ids = validate_rollout_ids(rollout_ids)
+        result = RolloutRemoval()
+        for start in range(0, len(rollout_ids), MAX_LEDGER_BATCH):
+            batch = rollout_ids[start : start + MAX_LEDGER_BATCH]
+            response = await self._request("POST", f"/rollouts/{action}", json={"rollout_ids": batch})
+            if response.status != 200:
+                raise RuntimeError(
+                    f"{action} of {len(batch)} rollout ledgers failed: HTTP {response.status} {await response.text()}"
+                )
+            removal = RolloutRemoval.model_validate(await response.json())
+            result.removed.extend(removal.removed)
+            result.absent.extend(removal.absent)
+        return result
 
     async def _request(self, method: str, path: str, **kwargs: Any):
         # Deferred because server_utils loads Gym's aiohttp/server stack.