feat(token-capture): retire and delete complete-record captures

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,

fern/versions/latest/pages/training-tutorials/external-agent-harnesses.mdx

@@ -353,6 +353,11 @@ async def deliver_rollout(
 
 Exclude any result with `mask_sample: true` from the loss. Call `retire_rollout_token_capture` only after durable downstream storage has accepted the rebuilt result. If the framework retires the records first and the write then fails, the captured token ids are lost.
 
+`retire_rollout_token_capture` only removes rollouts that were delivered, so it keeps masked and failed captures around for inspection and never sees an attempt you abandoned. To remove those captures, use the source's two batched operations, which match the ones the external-staging capture ledger provides.
+
+- `await source.retire(rollout_ids)` removes the rollouts' records and leaves a fence, so any later write for those rollouts fails with `TokenCaptureRetiredError` and is dropped. Late writes come from an abandoned attempt that is still running, or from a second run of a finished rollout after a client retry. Retire masked or failed captures once you no longer need them, and retire abandoned attempts as soon as you give up on them.
+- `await source.delete(rollout_ids)` removes the records together with the fence. Because the fence is what stops late writes, delete retired rollouts only once no write from them can still arrive, for example at the end of the run. Also delete a rollout ID before reusing it.
+
 ### Check a custom transport
 
 Run `run_conformance` from `nemo_gym.token_id_capture.conformance` before using a custom transport for training. It takes factories rather than instances, because several checks need a fresh client. Each factory call must return a new client connected to the backend. The function is async, returns the names of the checks that passed, and raises `ConformanceError` on the first failure:

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/__init__.py

@@ -66,6 +66,7 @@
     LineageResolution,
     LineageResolver,
     TokenCaptureFrozenError,
+    TokenCaptureRetiredError,
     TokenCaptureSnapshot,
     TokenSink,
     TokenSource,
@@ -134,6 +135,7 @@
     "LineageResolution",
     "LineageResolver",
     "TokenCaptureFrozenError",
+    "TokenCaptureRetiredError",
     "install_lineage_store",
     "installed_lineage_store",
     "TokenCaptureSnapshot",

nemo_gym/token_id_capture/conformance.py

@@ -30,7 +30,7 @@
 from typing import Awaitable, Callable
 
 from nemo_gym.token_id_capture.lineage import stamp_continuation
-from nemo_gym.token_id_capture.protocols import LineageResolver, TokenSink, TokenSource
+from nemo_gym.token_id_capture.protocols import LineageResolver, TokenCaptureRetiredError, TokenSink, TokenSource
 from nemo_gym.token_id_capture.records import (
     ParentResolutionStatus,
     TokenEntry,
@@ -128,6 +128,8 @@ def lineage() -> LineageResolver:
         ("freeze_idempotency", lambda r: _check_freeze_idempotency(sink(), source, r)),
         ("post_freeze_write_safety", lambda r: _check_post_freeze_write(sink(), source(), r)),
         ("conditional_retirement", lambda r: _check_conditional_retirement(sink(), source(), r)),
+        ("unconditional_retirement", lambda r: _check_unconditional_retirement(sink(), source(), r)),
+        ("delete_clears_retirement", lambda r: _check_delete_clears_retirement(sink(), source(), r)),
     ]
     if lineage_factory is not None:
         checks.append(
@@ -263,6 +265,63 @@ async def _check_conditional_retirement(sink: TokenSink, src: TokenSource, rollo
     _require(retired, name, "retiring the exact consumed snapshot failed")
 
 
+async def _visible_entries(src: TokenSource, rollout_id: str) -> int:
+    """Entries a consumer could build from; a retired rollout may refuse to freeze instead."""
+    try:
+        return len((await src.freeze(rollout_id)).entries)
+    except TokenCaptureRetiredError:
+        return 0
+
+
+async def _require_fenced(name: str, call: Awaitable[object], what: str) -> None:
+    """A write to a retired rollout must fail with ``TokenCaptureRetiredError``, as ``TokenSource.retire`` requires.
+
+    The capture sink treats that error as a late write; any other error would be reported as a capture failure.
+    """
+    try:
+        await call
+    except TokenCaptureRetiredError:
+        return
+    except Exception as error:  # noqa: BLE001 - any other error is reported as a capture failure.
+        raise ConformanceError(
+            name, f"{what} after retirement raised {type(error).__name__}, not TokenCaptureRetiredError"
+        ) from error
+    raise ConformanceError(name, f"{what} after retirement was accepted")
+
+
+async def _check_unconditional_retirement(sink: TokenSink, src: TokenSource, rollout_id: str) -> None:
+    name = "unconditional_retirement"
+    entry = _make_entry(rollout_id, "call-1", prompt=[11, 12], generation=[13, 14], request_items=_REQUEST, text="a")
+    await sink.put(entry)
+    # No freeze first: retire works whatever the rollout's state.
+    first = await src.retire([rollout_id])
+    _require(first.get("removed") == [rollout_id], name, f"retire did not report the rollout as removed: {first}")
+    late = _make_entry(
+        rollout_id, "call-2", prompt=[11, 12, 13, 14], generation=[15], request_items=_REQUEST, text="b"
+    )
+    await _require_fenced(name, sink.put(late), "a late put")
+    mark = getattr(sink, "mark_incomplete", None)
+    if mark is not None:
+        await _require_fenced(name, mark(rollout_id, "call-3"), "a late mark_incomplete")
+    begin = getattr(sink, "begin_call", None)
+    if begin is not None:
+        await _require_fenced(name, begin(rollout_id, "call-4"), "a late begin_call")
+    _require(await _visible_entries(src, rollout_id) == 0, name, "a write after retirement became visible")
+    again = await src.retire([rollout_id])
+    _require(again.get("absent") == [rollout_id], name, f"retiring again was not a no-op: {again}")
+
+
+async def _check_delete_clears_retirement(sink: TokenSink, src: TokenSource, rollout_id: str) -> None:
+    name = "delete_clears_retirement"
+    entry = _make_entry(rollout_id, "call-1", prompt=[11, 12], generation=[13, 14], request_items=_REQUEST, text="a")
+    await sink.put(entry)
+    await src.retire([rollout_id])
+    await src.delete([rollout_id])
+    # The rollout ID is reusable: a new execution starts empty and its writes are visible.
+    await sink.put(entry)
+    _require(await _visible_entries(src, rollout_id) == 1, name, "a write after delete was not visible")
+
+
 async def _check_lineage_visibility(
     sink: TokenSink,
     lineage_factory: Callable[[], LineageResolver],

nemo_gym/token_id_capture/consumer.py

@@ -83,12 +83,9 @@ def clear_token_captures_for_rollouts(records: list, token_capture_dirs: list[Pa
         return
     from nemo_gym.base_responses_api_model import maybe_rollout_id_from_run_body
 
+    rollout_ids = [rollout_id for record in records if (rollout_id := maybe_rollout_id_from_run_body(record))]
     for directory in token_capture_dirs:
-        store = TokenCaptureStore(directory)
-        for record in records:
-            rollout_id = maybe_rollout_id_from_run_body(record)
-            if rollout_id:
-                store.delete(rollout_id)
+        TokenCaptureStore(directory).delete_now(rollout_ids)
 
 
 def _assemble(

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.