diff --git a/nemo_gym/base_responses_api_model.py b/nemo_gym/base_responses_api_model.py
index 062181e59c..5d323dad33 100644
--- a/nemo_gym/base_responses_api_model.py
+++ b/nemo_gym/base_responses_api_model.py
@@ -72,6 +72,7 @@
     CaptureContext,
     capture_tokens,
     installed_token_sink,
+    register_call_intent,
     reset_token_sink,
     set_token_sink,
 )
@@ -209,6 +210,7 @@ async def _invoke_chat_completions(
         # chat_completions() signatures vary across servers: some take a leading `request`, some
         # only `body`. Dispatch on whichever this server declares so the shared dispatch works for
         # all of them.
+        await register_call_intent()
         if "request" in inspect.signature(self.chat_completions).parameters:
             completion = await self.chat_completions(request=request, body=params)
         else:
@@ -242,6 +244,7 @@ async def _invoke_responses(
         # responses() signatures vary across servers: some take a leading `request`, some only
         # `body`. Dispatch on whichever this server declares so the default messages() works for
         # all of them.
+        await register_call_intent()
         if "request" in inspect.signature(self.responses).parameters:
             response = await self.responses(request=request, body=params)
         else:
@@ -1107,6 +1110,19 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None
         # Installed sinks are resolved for each request.
         token_sink = self._configured_sink or installed_token_sink() or self._token_store
         capture_wanted = token_capture_requested and (token_sink is not None or self._token_capture_enabled)
+        if token_capture_requested and dialect is None and token_sink is not None:
+            # This call cannot produce a capture record.
+            # Its output may still feed a later prompt.
+            # Mark the rollout incomplete before forwarding the request.
+            try:
+                await token_sink.mark_incomplete(rollout_from_path, "")
+            except Exception:
+                logger.warning(
+                    "Could not mark rollout %s incomplete for unobserved path %s.",
+                    rollout_from_path,
+                    path,
+                    exc_info=True,
+                )
         if (self._store is None and not capture_wanted) or rollout_from_path is None or dialect is None:
             await self._app(scope, receive, send)
             return
diff --git a/nemo_gym/rollout_collection.py b/nemo_gym/rollout_collection.py
index 3399a6ca7a..1b78515061 100644
--- a/nemo_gym/rollout_collection.py
+++ b/nemo_gym/rollout_collection.py
@@ -85,6 +85,20 @@
     setup_server_client as setup_server_client_utils,
 )
 from nemo_gym.skills import SkillsConfig, load_skill_directory
+from nemo_gym.token_id_capture import (
+    TokenCaptureStore,
+    TokenIdCaptureConfig,
+    clear_token_captures_for_rollouts,
+    installed_token_source,
+    token_id_capture_dirs_from_config,
+)
+from nemo_gym.token_id_capture.config import token_id_capture_enabled_for_agent
+from nemo_gym.token_id_capture.delivery import (
+    MASK_SAMPLE_KEY,
+    capture_build_can_retire,
+    finalize_rollout_token_capture,
+    retire_rollout_token_capture,
+)
 
 
 logger = logging.getLogger(__name__)
@@ -790,13 +804,53 @@ async def run_from_config(self, config: RolloutCollectionConfig) -> Tuple[List[D
 
         # Resolve capture dirs once so each rollout's captured model calls can be folded
         # into its record below (uniform across agents; no-op when capture is off / dirs absent).
-        capture_dirs = model_call_capture_dirs_from_config(get_global_config_dict())
+        global_config = get_global_config_dict()
+        capture_dirs = model_call_capture_dirs_from_config(global_config)
+        # Resolve the training-token store directory once.
+        # Training capture is independent of evaluation capture.
+        # An empty result disables training-token capture.
+        token_capture_dirs = token_id_capture_dirs_from_config(global_config)
+        # The finalizer reads and freezes records through this source.
+        # The source is absent when capture or response rebuilding is disabled.
+        # A framework-owned transport may rebuild through its own source.
+        # The sink still records captures when Gym does not rebuild.
+        # Reruns still clear deterministic rollout ids before dispatch.
+        token_source = None
+        owned_token_source = None
+        token_capture_config = TokenIdCaptureConfig.model_validate(global_config)
+        if token_capture_config.enabled and token_capture_config.token_id_capture.rebuild_response:
+            token_source = installed_token_source()
+            if token_source is None and token_capture_dirs:
+                token_source = TokenCaptureStore(token_capture_dirs[0])
+            if isinstance(token_source, TokenCaptureStore):
+                owned_token_source = token_source
 
         # Clear only rows about to be dispatched, after resume has assigned retry suffixes. This also
         # removes a kill-shaped attempt's partial capture when its rollout-attempt id is reused.
         if capture_dirs:
             print("Clearing existing model-call captures for rollouts being dispatched")
             clear_model_call_captures_for_rollouts(input_rows, capture_dirs)
+        token_capture_rows = [
+            row
+            for row in input_rows
+            if token_id_capture_enabled_for_agent(global_config, (row.get(AGENT_REF_KEY_NAME) or {}).get("name"))
+        ]
+        if token_capture_config.token_id_capture.rebuild_response and token_capture_rows and token_source is None:
+            raise ValueError(
+                "Token capture response rebuilding requires a TokenSource in the rollout-collector process. "
+                "Call install_token_source before starting collection or configure token_id_capture.dir."
+            )
+        if token_capture_dirs and token_capture_rows:
+            # Token stores append under deterministic rollout ids.
+            # Clear stale records to avoid merging different attempts.
+            print("Clearing existing token captures for rollouts being dispatched")
+            clear_token_captures_for_rollouts(token_capture_rows, token_capture_dirs)
+
+        # Stop a run that produces mostly masked captures.
+        finalized_count = 0
+        masked_count = 0
+        mask_reasons: Counter = Counter()
+        warned_malformed_rollout_id = False
 
         # Intermediate status printing
         pcts_to_print = list(range(1, 100)) + [99.5]
@@ -834,6 +888,43 @@ async def run_from_config(self, config: RolloutCollectionConfig) -> Tuple[List[D
             if "ng_model_call_capture" in result or "ng_agent_observations" in result or NG_TRAJECTORY_KEY in result:
                 _attach_trajectory_record(row, result)
 
+            # Freeze and rebuild tokens only for participating agents.
+            # This step does not retire the frozen snapshot.
+            # It leaves harness output and reward unchanged.
+            # Direct callers of run_examples finalize each record themselves.
+            token_capture_build = None
+            if token_id_capture_enabled_for_agent(
+                global_config,
+                (row.get(AGENT_REF_KEY_NAME) or {}).get("name"),
+            ):
+                token_capture_build = await finalize_rollout_token_capture(result, token_source)
+                if token_capture_build is not None:
+                    finalized_count += 1
+                    if token_capture_build.get(MASK_SAMPLE_KEY):
+                        masked_count += 1
+                        # Aggregate available reasons for the abort message.
+                        build_metrics = token_capture_build.get("metrics") or {}
+                        if build_metrics.get("capture_incomplete"):
+                            mask_reasons["capture_incomplete"] += 1
+                        if build_metrics.get("unresolved_parent_calls"):
+                            mask_reasons["unresolved_parent_calls"] += 1
+                        build_error = token_capture_build.get("error") or build_metrics.get("error")
+                        if build_error:
+                            mask_reasons[str(build_error)] += 1
+                    settings = token_capture_config.token_id_capture
+                    if (
+                        settings.max_mask_fraction is not None
+                        and finalized_count >= settings.mask_fraction_min_samples
+                        and masked_count / finalized_count > settings.max_mask_fraction
+                    ):
+                        raise RuntimeError(
+                            f"{masked_count}/{finalized_count} finalized rollouts "
+                            f"({masked_count / finalized_count:.1%}) are masked, exceeding "
+                            f"token_id_capture.max_mask_fraction={settings.max_mask_fraction}. "
+                            f"Mask reasons: {dict(mask_reasons)}. Aborting instead of collecting "
+                            "mostly token-less data."
+                        )
+
             no_persist = bool(result.get(NG_NO_PERSIST_KEY))
             failure_class = result.get(NG_FAILURE_CLASS_KEY)
 
@@ -856,6 +947,21 @@ async def run_from_config(self, config: RolloutCollectionConfig) -> Tuple[List[D
                 results_file.flush()
                 persisted_rows.append(row)
                 persisted_results.append(result)
+                try:
+                    rollout_id = maybe_rollout_id_from_run_body(result)
+                except (TypeError, ValueError) as error:
+                    # Preserve capture evidence when the rollout id is invalid.
+                    rollout_id = None
+                    if not warned_malformed_rollout_id:
+                        warned_malformed_rollout_id = True
+                        warnings.warn(
+                            f"a result carries a malformed rollout id ({error}); "
+                            "its token capture will not be retired.",
+                            stacklevel=2,
+                        )
+                if rollout_id is not None and capture_build_can_retire(token_capture_build):
+                    os.fsync(results_file.fileno())
+                    await retire_rollout_token_capture(rollout_id, token_source, token_capture_build)
 
             counts_left[row[AGENT_REF_KEY_NAME]["name"]] -= 1
             if counts_left[row[AGENT_REF_KEY_NAME]["name"]] <= 0:
@@ -893,6 +999,8 @@ async def run_from_config(self, config: RolloutCollectionConfig) -> Tuple[List[D
 
         results_file.close()
         failures_file.close()
+        if owned_token_source is not None:
+            await owned_token_source.close()
 
         if config.upload_rollouts and get_exporters():  # pragma: no cover
             print("Uploading rollouts. This may take a few minutes if your data is large.")
diff --git a/nemo_gym/token_id_capture/__init__.py b/nemo_gym/token_id_capture/__init__.py
index 229b00e972..de1b3cf97f 100644
--- a/nemo_gym/token_id_capture/__init__.py
+++ b/nemo_gym/token_id_capture/__init__.py
@@ -25,6 +25,15 @@
 Framework transports may provide their own sink and source.
 There is no HTTP token reader.
 This leaf package avoids imports from Gym's server stack.
+The rollout-record finalizer needs Gym's server stack.
+It is deliberately not re-exported here.
+Import ``nemo_gym.token_id_capture.delivery`` from server-side code.
+The incomplete state prevents training on a rollout that lost a model call.
+Finalization freezes and rebuilds the rollout.
+Finalization does not retire the snapshot.
+The caller retires it only after durable handoff.
+Retirement uses the frozen ``snapshot_id`` and version.
+Failed or masked builds retain their capture evidence.
 """
 
 from nemo_gym.token_id_capture.builder import (
@@ -63,6 +72,7 @@
     capture_tokens,
     commit_entry,
     current_capture_context,
+    register_call_intent,
     reset_token_sink,
     set_token_sink,
 )
@@ -89,6 +99,7 @@
     "CaptureContext",
     "set_token_sink",
     "reset_token_sink",
+    "register_call_intent",
     "capture_tokens",
     "commit_entry",
     "current_capture_context",
diff --git a/nemo_gym/token_id_capture/config.py b/nemo_gym/token_id_capture/config.py
index 119d06857d..cc0ed87ff3 100644
--- a/nemo_gym/token_id_capture/config.py
+++ b/nemo_gym/token_id_capture/config.py
@@ -48,11 +48,24 @@
 They do not inherit a sink installed by a launcher.
 Configure the sink here so each worker builds its own.
 Programmatic installation must occur inside the serving process.
+
+Choosing who reads them back
+----------------------------
+``rebuild_response`` controls whether Gym rebuilds a finished rollout.
+Gym freezes captured records before rebuilding ``response.output``.
+Rebuilding does not retire the frozen snapshot.
+Gym retires a successful build only after durable handoff.
+Retirement uses the frozen ``snapshot_id`` and version.
+Failed or masked builds retain their capture evidence.
+Set it to false when a framework reads through its own ``TokenSource``.
+Gym then stops after the write.
+Read ownership is independent of write ownership.
 """
 
 from __future__ import annotations
 
 import logging
+from collections.abc import Mapping
 from importlib import import_module
 from pathlib import Path
 from typing import Any
@@ -88,8 +101,14 @@ class TokenIdCaptureSettings(BaseModel):
     # A real transport needs explicit endpoint, client, or credential wiring.
     # Use ``${oc.env:VAR}`` for secrets instead of writing them here.
     sink_kwargs: dict[str, Any] = Field(default_factory=dict)
-    # Rebuild opaque-harness responses from captured records after the run.
+    # Whether Gym freezes capture records and rebuilds the response.
+    # Finalization does not retire the frozen snapshot.
+    # Durable delivery permits retirement by snapshot id and version.
     rebuild_response: bool = True
+    # Abort once enough finalized rollouts exceed this masked fraction.
+    # ``None`` disables the limit.
+    max_mask_fraction: float | None = None
+    mask_fraction_min_samples: int = 50
 
 
 class TokenIdCaptureConfig(BaseModel):
@@ -179,3 +198,26 @@ def _build_endpoint(target: str, kwargs: dict[str, Any], protocol: type, kind: s
 def token_id_capture_config(global_config_dict: Any) -> TokenIdCaptureConfig:
     """Read the capture settings out of a global config dict."""
     return TokenIdCaptureConfig.model_validate(global_config_dict or {})
+
+
+def token_id_capture_enabled_for_agent(global_config_dict: Any, agent_name: str | None) -> bool:
+    """Return whether one configured agent participates in capture."""
+    config = token_id_capture_config(global_config_dict)
+    settings = config.token_id_capture
+    if not settings.enabled:
+        return False
+    if settings.all_agents:
+        return True
+    if not agent_name or not isinstance(global_config_dict, Mapping):
+        return False
+    server_entry = global_config_dict.get(agent_name)
+    if not isinstance(server_entry, Mapping):
+        return False
+    agents = server_entry.get("responses_api_agents")
+    if not isinstance(agents, Mapping):
+        return False
+    return any(
+        bool(agent_config.get("token_id_capture", False))
+        for agent_config in agents.values()
+        if isinstance(agent_config, Mapping)
+    )
diff --git a/nemo_gym/token_id_capture/delivery.py b/nemo_gym/token_id_capture/delivery.py
new file mode 100644
index 0000000000..ef4a0910f9
--- /dev/null
+++ b/nemo_gym/token_id_capture/delivery.py
@@ -0,0 +1,227 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Build a token-bearing record from one finished rollout.
+
+The finalizer freezes the rollout's captured model calls.
+It rebuilds ``response.output`` from that frozen snapshot.
+It does not retire the snapshot.
+The caller may retire it only after durable handoff.
+Retirement uses the frozen ``snapshot_id`` and version.
+
+The caller provides both the rollout record and its ``TokenSource``.
+Gym resolves its source from Gym configuration.
+A training framework may provide a source from its own data plane.
+Training correlation must preserve ``/ng-rollout/<id>/training-token-capture``.
+
+Existing token ids are the policy's sampled data.
+The finalizer leaves a rollout containing any token ids unchanged.
+Failed or masked builds retain their capture evidence.
+"""
+
+from __future__ import annotations
+
+import warnings
+
+from nemo_gym.rollout_correlation import maybe_rollout_id_from_run_body
+from nemo_gym.token_id_capture.consumer import trajectories_from_source
+from nemo_gym.token_id_capture.protocols import TokenSource
+
+
+# Attach token-capture health to each rollout record.
+# It reports build losses and masking reasons.
+TOKEN_CAPTURE_KEY = "_ng_token_capture"
+
+# A consumer reads this top-level field to exclude a rollout from the loss.
+# The field stays outside TOKEN_CAPTURE_KEY for direct access.
+MASK_SAMPLE_KEY = "mask_sample"
+_REDUNDANT_CAPTURE_KEY = "_redundant_capture"
+
+
+def rollout_carries_token_ids(result: dict) -> bool:
+    """Whether this rollout already holds what training needs.
+
+    Return true when any output item carries generated token ids.
+    These ids are what the policy sampled.
+    They take precedence over a reconstruction that may differ.
+    Partial token coverage must remain visible instead of being overwritten.
+    """
+    response = result.get("response")
+    if not isinstance(response, dict):
+        return False
+    return any(isinstance(item, dict) and item.get("generation_token_ids") for item in (response.get("output") or []))
+
+
+def _unusable(result: dict, error: str, message: str) -> dict:
+    """Mask a rollout that needed token ids and could not get them.
+
+    An unmasked rollout would appear healthy until it reaches the trainer.
+    The record retains the reason for aggregate reporting.
+    """
+    warnings.warn(message, stacklevel=3)
+    metrics = {"n_calls": 0, "error": error}
+    result[MASK_SAMPLE_KEY] = True
+    result[TOKEN_CAPTURE_KEY] = metrics
+    return {"rebuilt_response": None, MASK_SAMPLE_KEY: True, "error": error, "metrics": metrics}
+
+
+async def finalize_rollout_token_capture(result: dict, source: TokenSource | None) -> dict | None:
+    """Rebuild one finished rollout record's ``response.output`` from its recorded token ids.
+
+    Call this after the harness and verifier finish the record.
+    The function mutates ``result`` in place.
+    It replaces only ``response.output``.
+    It preserves the reward and all other harness and verifier output.
+
+    The function freezes capture records through ``source``.
+    It rebuilds from that frozen snapshot.
+    It never retires the snapshot.
+    A ``None`` source means this caller does not rebuild.
+    A rollout that already carries token ids is left unchanged.
+    Its redundant frozen capture remains eligible for retirement after handoff.
+
+    The function never raises.
+    Missing or ambiguous tokens cause masking.
+    Failed or masked builds retain their frozen evidence.
+
+    Return the build with its rebuilt response, metrics, and optional error.
+    Return ``None`` when no source exists.
+    An unusable build has no rebuilt response and sets ``mask_sample``.
+    """
+    if source is None:
+        return None
+
+    try:
+        rollout_id = maybe_rollout_id_from_run_body(result)
+    except (TypeError, ValueError) as error:
+        # A malformed explicit id cannot be looked up.
+        # Mask the rollout instead of raising.
+        return _unusable(
+            result,
+            f"malformed rollout id: {error}",
+            f"a rollout result carries a malformed id ({error}), so its recorded token ids could "
+            "not be looked up and it will be token-less.",
+        )
+    if rollout_carries_token_ids(result):
+        # Re-finalization must return the frozen snapshot.
+        # The caller must be able to retire on every path.
+        if rollout_id is None:
+            return None
+        try:
+            snapshot = await source.freeze(rollout_id)
+        except Exception:
+            warnings.warn(f"could not freeze redundant records for rollout {rollout_id}.", stacklevel=2)
+            return None
+        return {
+            "rebuilt_response": None,
+            MASK_SAMPLE_KEY: False,
+            _REDUNDANT_CAPTURE_KEY: True,
+            "_capture_snapshot": {
+                "snapshot_id": snapshot.snapshot_id,
+                "version": snapshot.version,
+            },
+        }
+
+    if rollout_id is None:
+        # No correlation key was preserved on the finished record.
+        return _unusable(
+            result,
+            "no capture key",
+            "a rollout result carries no id and no task/rollout indices, so its recorded token ids "
+            "could not be looked up and it will be token-less.",
+        )
+
+    response = result.get("response") if isinstance(result.get("response"), dict) else {}
+    try:
+        built = await trajectories_from_source(rollout_id, source, model=str(response.get("model") or ""))
+    except Exception as error:
+        # A transport failure may be unrelated to this rollout.
+        # Mask this rollout instead of failing the entire batch.
+        return _unusable(
+            result,
+            f"{type(error).__name__}: {error}",
+            f"could not read the records for rollout {rollout_id}: {type(error).__name__}: {error}. "
+            "It will be token-less.",
+        )
+    if built is None:
+        # Correlation failed between the agent and capture middleware.
+        # An external harness or proxy may have dropped ``/ng-rollout/<id>/training-token-capture``.
+        return _unusable(
+            result,
+            "nothing recorded",
+            f"rollout {rollout_id} has no token ids and none were recorded for it, so it was not "
+            "rebuilt and will be token-less. Its model calls likely did not reach the capture "
+            "middleware correlated.",
+        )
+
+    projected = built["rebuilt_response"]
+    if projected is not None:
+        if isinstance(result.get("response"), dict):
+            result["response"]["output"] = projected["output"]
+        else:
+            result["response"] = projected
+
+    # Record build losses so partial trajectories remain visible.
+    record_metrics = dict(built.get("metrics") or {})
+    if built.get("error"):
+        record_metrics["error"] = built["error"]
+    if built.get(MASK_SAMPLE_KEY):
+        # Keep the masking verdict at the top level.
+        # Retain its reasons in the metrics.
+        result[MASK_SAMPLE_KEY] = True
+        warnings.warn(
+            f"rollout {rollout_id} was captured incompletely or ambiguously ({record_metrics}); "
+            "it is marked for masking rather than trained on.",
+            stacklevel=2,
+        )
+    result[TOKEN_CAPTURE_KEY] = record_metrics
+
+    return built
+
+
+async def retire_rollout_token_capture(
+    rollout_id: str,
+    source: TokenSource | None,
+    built: dict | None,
+) -> bool:
+    """Retire a frozen snapshot after durable handoff.
+
+    The caller owns the durability boundary.
+    Call this only after downstream acceptance or a local fsync.
+    Retirement uses the frozen ``snapshot_id`` and version.
+    Failed or masked builds remain as diagnostic evidence.
+    """
+    if source is None or not capture_build_can_retire(built):
+        return False
+    snapshot = built.get("_capture_snapshot")
+    if not isinstance(snapshot, dict):
+        warnings.warn(f"rollout {rollout_id} has no frozen capture identity to retire.", stacklevel=2)
+        return False
+    try:
+        return await source.drop(
+            rollout_id,
+            snapshot_id=str(snapshot["snapshot_id"]),
+            version=int(snapshot["version"]),
+        )
+    except Exception:
+        warnings.warn(f"could not retire the records for rollout {rollout_id}.", stacklevel=2)
+        return False
+
+
+def capture_build_can_retire(built: dict | None) -> bool:
+    """Whether a successful build consumed a frozen snapshot."""
+    if built is None or built.get(MASK_SAMPLE_KEY):
+        return False
+    return built.get("rebuilt_response") is not None or bool(built.get(_REDUNDANT_CAPTURE_KEY))
diff --git a/nemo_gym/token_id_capture/sink.py b/nemo_gym/token_id_capture/sink.py
index 53a595cdcb..69ddde7c7d 100644
--- a/nemo_gym/token_id_capture/sink.py
+++ b/nemo_gym/token_id_capture/sink.py
@@ -83,6 +83,21 @@ def reset_token_sink(token: Token) -> None:
     _CAPTURE_CONTEXT.reset(token)
 
 
+async def register_call_intent() -> None:
+    """Record that the captured call is about to be dispatched.
+
+    ``begin_call`` is an optional sink extension.
+    It lets a source detect a call whose entry was lost.
+    A failure happens before generation and must fail the model call.
+    """
+    context = _CAPTURE_CONTEXT.get()
+    if context is None or context.token_sink is None:
+        return
+    begin_call = getattr(context.token_sink, "begin_call", None)
+    if begin_call is not None:
+        await begin_call(context.rollout_id, context.model_call_id)
+
+
 async def capture_tokens(response: Any) -> None:
     """Record a ``TokenEntry`` from a complete model response.
 
diff --git a/nemo_gym/token_id_capture/store.py b/nemo_gym/token_id_capture/store.py
index d6d81bcc55..1d30bb3670 100644
--- a/nemo_gym/token_id_capture/store.py
+++ b/nemo_gym/token_id_capture/store.py
@@ -17,7 +17,9 @@
 
 Each rollout uses one ``<rollout_id>.tokens.jsonl`` file.
 Evaluation records use a separate file.
-Each write uses ``fsync``.
+Every entry line is ``fsync``ed before ``put`` returns — that is the durability
+guarantee. The state index is written atomically but fsynced only on lifecycle
+transitions (freeze, mark, drop); it is reconstructible from the JSONL tail.
 A per-rollout file lock serializes writers to the same rollout.
 Different rollouts can write concurrently.
 """
@@ -27,8 +29,10 @@
 import asyncio
 import fcntl
 import hashlib
+import logging
 import os
 import tempfile
+import time
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
@@ -40,6 +44,9 @@
 from nemo_gym.token_id_capture.records import TokenEntry
 
 
+logger = logging.getLogger(__name__)
+
+
 def validate_rollout_id(rollout_id: str) -> str:
     """Reject anything that could escape the store directory or index a bad file."""
     if not rollout_id or any(not (char.isascii() and (char.isalnum() or char in "._-")) for char in rollout_id):
@@ -65,6 +72,10 @@ def incomplete_path_for(self, rollout_id: str) -> Path:
         """Sentinel marking that at least one call of this rollout failed to capture."""
         return self._root / f"{validate_rollout_id(rollout_id)}.tokens.incomplete"
 
+    def intents_path_for(self, rollout_id: str) -> Path:
+        """Return the durable per-call intent path."""
+        return self._root / f"{validate_rollout_id(rollout_id)}.tokens.intents"
+
     def state_path_for(self, rollout_id: str) -> Path:
         return self._root / f"{validate_rollout_id(rollout_id)}.tokens.state.json"
 
@@ -85,6 +96,7 @@ def _read_state(self, rollout_id: str) -> dict[str, Any]:
         if not path.exists():
             return {
                 "frozen": False,
+                "retired": False,
                 "incomplete": False,
                 "snapshot_id": "",
                 "version": 0,
@@ -124,11 +136,27 @@ def _sync_entry_index(self, rollout_id: str, state: dict[str, Any]) -> bool:
         if path.exists():
             with path.open("rb") as handle:
                 handle.seek(indexed_size)
-                for line in handle:
-                    payload = line.strip()
-                    if not payload:
-                        continue
-                    entry = TokenEntry.model_validate(orjson.loads(payload))
+                tail = handle.read()
+            position = 0
+            while position < len(tail):
+                newline = tail.find(b"\n", position)
+                end = len(tail) if newline == -1 else newline
+                payload = tail[position:end].strip()
+                if payload:
+                    try:
+                        parsed = orjson.loads(payload)
+                    except orjson.JSONDecodeError:
+                        remainder = tail[end + 1 :] if newline != -1 else b""
+                        if remainder.strip():
+                            # A malformed line before more content is corrupt.
+                            raise
+                        # A torn final line was never acknowledged.
+                        # Drop it while the caller holds the exclusive lock.
+                        os.truncate(path, indexed_size + position)
+                        file_size = indexed_size + position
+                        logger.warning("Dropped %d torn trailing bytes from %s", len(tail) - position, path)
+                        break
+                    entry = TokenEntry.model_validate(parsed)
                     digest = self._entry_digest(payload)
                     existing = entry_digests.get(entry.model_call_id)
                     if existing is not None and existing != digest:
@@ -141,6 +169,7 @@ def _sync_entry_index(self, rollout_id: str, state: dict[str, Any]) -> bool:
                         )
                     entry_digests[entry.model_call_id] = digest
                     recovered += 1
+                position = end + 1
 
         state["entry_digests"] = entry_digests
         state["indexed_size"] = file_size
@@ -148,20 +177,26 @@ def _sync_entry_index(self, rollout_id: str, state: dict[str, Any]) -> bool:
             state["version"] = int(state.get("version", 0)) + recovered
         return True
 
-    def _write_state(self, rollout_id: str, state: dict[str, Any]) -> None:
+    def _write_state(self, rollout_id: str, state: dict[str, Any], *, durable: bool = True) -> None:
+        # durable=False skips both fsyncs.
+        # The temporary-file replacement remains atomic.
+        # Use it only for state reconstructed from the JSONL tail.
+        # Lifecycle flags are not reconstructible.
         payload = orjson.dumps(state, option=orjson.OPT_SORT_KEYS | orjson.OPT_APPEND_NEWLINE)
         with tempfile.NamedTemporaryFile(dir=self._root, prefix=".tokens-state-", delete=False) as handle:
             temporary_path = Path(handle.name)
             try:
                 handle.write(payload)
                 handle.flush()
-                os.fsync(handle.fileno())
+                if durable:
+                    os.fsync(handle.fileno())
             except BaseException:
                 temporary_path.unlink(missing_ok=True)
                 raise
         try:
             os.replace(temporary_path, self.state_path_for(rollout_id))
-            self._fsync_root()
+            if durable:
+                self._fsync_root()
         finally:
             temporary_path.unlink(missing_ok=True)
 
@@ -175,6 +210,8 @@ def _fsync_root(self) -> None:
     def _mark_incomplete(self, rollout_id: str, model_call_id: str = "") -> None:
         with self._locked(rollout_id):
             state = self._read_state(rollout_id)
+            if state.get("retired", False):
+                raise RuntimeError(f"Token capture for rollout {rollout_id} is retired")
             state["incomplete"] = True
             state["version"] = int(state.get("version", 0)) + 1
             self._write_state(rollout_id, state)
@@ -200,6 +237,8 @@ def append(self, entry: TokenEntry) -> None:
         rollout_id = entry.rollout_id
         with self._locked(rollout_id):
             state = self._read_state(rollout_id)
+            if state.get("retired", False):
+                raise RuntimeError(f"Token capture for rollout {rollout_id} is retired")
             if state.get("frozen", False):
                 raise RuntimeError(f"Token capture for rollout {rollout_id} is already frozen")
             index_changed = self._sync_entry_index(rollout_id, state)
@@ -224,7 +263,9 @@ def append(self, entry: TokenEntry) -> None:
                 state["indexed_size"] = handle.tell()
             entry_digests[entry.model_call_id] = digest
             state["version"] = int(state.get("version", 0)) + 1
-            self._write_state(rollout_id, state)
+            # The entry line's fsync is the durability guarantee.
+            # The index is reconstructible from the JSONL tail.
+            self._write_state(rollout_id, state, durable=False)
 
     # The file store is Gym's default TokenSink and TokenSource.
     # A framework can replace it without changing the capture path.
@@ -238,6 +279,35 @@ async def put(self, entry: TokenEntry) -> None:
         """
         await asyncio.to_thread(self.append, entry)
 
+    def _begin_call(self, rollout_id: str, model_call_id: str) -> None:
+        """Durably record that a captured call is about to be dispatched.
+
+        A lost entry leaves a dangling intent.
+        ``freeze_now`` then masks the rollout.
+        A failure here happens before generation.
+        """
+        with self._locked(rollout_id):
+            state = self._read_state(rollout_id)
+            if state.get("retired", False):
+                raise RuntimeError(f"Token capture for rollout {rollout_id} is retired")
+            if state.get("frozen", False):
+                raise RuntimeError(f"Token capture for rollout {rollout_id} is already frozen")
+            with self.intents_path_for(rollout_id).open("ab") as handle:
+                handle.write(model_call_id.encode("utf-8") + b"\n")
+                handle.flush()
+                os.fsync(handle.fileno())
+
+    async def begin_call(self, rollout_id: str, model_call_id: str) -> None:
+        await asyncio.to_thread(self._begin_call, rollout_id, model_call_id)
+
+    def _dangling_intents(self, rollout_id: str, entries: tuple[TokenEntry, ...]) -> list[str]:
+        path = self.intents_path_for(rollout_id)
+        if not path.exists():
+            return []
+        recorded = {entry.model_call_id for entry in entries}
+        intents = [line.strip().decode("utf-8") for line in path.read_bytes().splitlines() if line.strip()]
+        return [call_id for call_id in intents if call_id not in recorded]
+
     async def freeze(self, rollout_id: str) -> TokenCaptureSnapshot:
         return await asyncio.to_thread(self.freeze_now, rollout_id)
 
@@ -245,6 +315,8 @@ def freeze_now(self, rollout_id: str) -> TokenCaptureSnapshot:
         """Synchronously freeze one rollout and return its stable snapshot."""
         with self._locked(rollout_id):
             state = self._read_state(rollout_id)
+            if state.get("retired", False):
+                raise RuntimeError(f"Token capture for rollout {rollout_id} is retired")
             index_changed = self._sync_entry_index(rollout_id, state)
             if not state.get("frozen", False):
                 state["frozen"] = True
@@ -254,21 +326,17 @@ def freeze_now(self, rollout_id: str) -> TokenCaptureSnapshot:
             elif index_changed:
                 self._write_state(rollout_id, state)
             entries = tuple(self._read_entries_unlocked(rollout_id))
+            # A dispatched call with no entry was lost.
+            # The rollout must be masked.
+            incomplete = bool(state.get("incomplete", False)) or bool(self._dangling_intents(rollout_id, entries))
             return TokenCaptureSnapshot(
                 rollout_id=rollout_id,
                 entries=entries,
-                incomplete=bool(state.get("incomplete", False)),
+                incomplete=incomplete,
                 snapshot_id=str(state["snapshot_id"]),
                 version=int(state["version"]),
             )
 
-    async def tokens_for(self, rollout_id: str) -> list[TokenEntry]:
-        """Read records for compatibility diagnostics.
-
-        Consumers should use ``freeze``.
-        """
-        return await asyncio.to_thread(self.read_entries, rollout_id)
-
     async def drop(self, rollout_id: str, *, snapshot_id: str, version: int) -> bool:
         """Delete snapshot payloads while retaining its tombstone and lock."""
         return await asyncio.to_thread(self._drop, rollout_id, snapshot_id, version)
@@ -284,6 +352,7 @@ def _drop(self, rollout_id: str, snapshot_id: str, version: int) -> bool:
                 return False
             self.path_for(rollout_id).unlink(missing_ok=True)
             self.incomplete_path_for(rollout_id).unlink(missing_ok=True)
+            self.intents_path_for(rollout_id).unlink(missing_ok=True)
             # Keep a frozen tombstone until explicit pre-dispatch cleanup.
             # A late writer from this attempt must still observe the freeze.
             state["indexed_size"] = 0
@@ -306,9 +375,42 @@ def delete(self, rollout_id: str) -> None:
         with self._locked(rollout_id):
             self.path_for(rollout_id).unlink(missing_ok=True)
             self.incomplete_path_for(rollout_id).unlink(missing_ok=True)
+            self.intents_path_for(rollout_id).unlink(missing_ok=True)
             self.state_path_for(rollout_id).unlink(missing_ok=True)
             self._fsync_root()
 
+    def sweep_retired(self, older_than_seconds: float) -> int:
+        """Remove retired tombstones older than the cutoff and return the count removed.
+
+        Callers choose the retention policy.
+        ``drop`` already removed entries and JSONL payloads.
+        This removes state, locks, intents, and incomplete markers.
+        """
+        cutoff = time.time() - older_than_seconds
+        removed = 0
+        for state_path in self._root.glob("*.tokens.state.json"):
+            rollout_id = state_path.name[: -len(".tokens.state.json")]
+            try:
+                validate_rollout_id(rollout_id)
+            except ValueError:
+                continue
+            with self._locked(rollout_id):
+                try:
+                    if state_path.stat().st_mtime > cutoff:
+                        continue
+                except FileNotFoundError:
+                    continue
+                if not self._read_state(rollout_id).get("retired", False):
+                    continue
+                state_path.unlink(missing_ok=True)
+                self.intents_path_for(rollout_id).unlink(missing_ok=True)
+                self.incomplete_path_for(rollout_id).unlink(missing_ok=True)
+                self.lock_path_for(rollout_id).unlink(missing_ok=True)
+                removed += 1
+        if removed:
+            self._fsync_root()
+        return removed
+
     def read_entries(self, rollout_id: str) -> list[TokenEntry]:
         with self._locked(rollout_id, shared=True):
             return self._read_entries_unlocked(rollout_id)
diff --git a/tests/unit_tests/test_base_responses_api_agent.py b/tests/unit_tests/test_base_responses_api_agent.py
index 74a456d6ab..57703ac067 100644
--- a/tests/unit_tests/test_base_responses_api_agent.py
+++ b/tests/unit_tests/test_base_responses_api_agent.py
@@ -68,3 +68,37 @@ async def run(self, body=...):
         assert result.group_level_metrics == []
         assert result.agent_metrics == {}
         assert result.key_metrics == {}
+
+    def _agent(self, global_config: dict, *, token_id_capture: bool = False) -> SimpleResponsesAPIAgent:
+        config = BaseResponsesAPIAgentConfig(
+            host="", port=0, entrypoint="", name="", token_id_capture=token_id_capture
+        )
+
+        class _Agent(SimpleResponsesAPIAgent):
+            async def responses(self, body=...):
+                raise NotImplementedError
+
+            async def run(self, body=...):
+                raise NotImplementedError
+
+        client = MagicMock(spec=ServerClient)
+        client.global_config_dict = global_config
+        return _Agent(config=config, server_client=client)
+
+    def test_eval_capture_prefix_applies_to_every_agent(self) -> None:
+        # Evaluation capture correlates every agent.
+        # It does not depend on the agent's training-token opt-in.
+        body = {"_ng_task_index": 0, "_ng_rollout_index": 0}
+        assert self._agent({}).rollout_id_from_run(body) is None
+        assert self._agent({"observability_enabled": True}).rollout_id_from_run(body) == "0-0"
+
+    def test_token_capture_prefix_is_scoped_to_participating_agents(self) -> None:
+        # Training-token capture requires both run-level enablement and agent opt-in.
+        # Correlated calls preserve ``/ng-rollout/<id>/training-token-capture``.
+        # Native agents carry token ids inline and do not opt in.
+        body = {"_ng_task_index": 0, "_ng_rollout_index": 0}
+        gc = {"token_id_capture": {"enabled": True}}
+        assert self._agent(gc, token_id_capture=False).rollout_id_from_run(body) is None
+        assert self._agent(gc, token_id_capture=True).rollout_id_from_run(body) == "0-0"
+        # Agent opt-in alone does not enable capture.
+        assert self._agent({}, token_id_capture=True).rollout_id_from_run(body) is None
diff --git a/tests/unit_tests/test_base_responses_api_model.py b/tests/unit_tests/test_base_responses_api_model.py
index 0c9c5fb1f0..b39a90506a 100644
--- a/tests/unit_tests/test_base_responses_api_model.py
+++ b/tests/unit_tests/test_base_responses_api_model.py
@@ -1563,3 +1563,76 @@ def test_capture_store_cross_process_append_no_loss(tmp_path):
     rows = CaptureStore(tmp_path).read("0-0")
     assert len(rows) == 400
     assert sorted(r["request"]["i"] for r in rows) == list(range(400))
+
+
+def _run_capture_middleware_on(path: str, *, token_store, sent: list | None = None, forwarded: list | None = None):
+    """Drive _CaptureMiddleware over one request to ``path`` with a token store."""
+    import asyncio
+
+    from nemo_gym.base_responses_api_model import _CaptureMiddleware
+
+    forwarded = forwarded if forwarded is not None else []
+    sent = sent if sent is not None else []
+
+    async def app(scope, receive, send):
+        forwarded.append(scope["path"])
+        await receive()
+        await send({"type": "http.response.start", "status": 200, "headers": [(b"content-type", b"application/json")]})
+        await send({"type": "http.response.body", "body": b"{}", "more_body": False})
+
+    async def receive():
+        return {"type": "http.request", "body": b"{}", "more_body": False}
+
+    async def send(message):
+        sent.append(message)
+
+    asyncio.run(
+        _CaptureMiddleware(
+            app,
+            store=None,
+            model_server_name="srv",
+            token_store=token_store,
+            token_capture_enabled=True,
+        )({"type": "http", "path": path, "raw_path": path.encode(), "headers": []}, receive, send)
+    )
+    return forwarded, sent
+
+
+def test_unobserved_dialect_under_capture_prefix_marks_incomplete(tmp_path):
+    # The middleware cannot capture tokens from /v1/completions.
+    # Its output can still feed later prompts.
+    from nemo_gym.token_id_capture import TokenCaptureStore
+
+    token_store = TokenCaptureStore(tmp_path)
+    forwarded, sent = _run_capture_middleware_on(
+        "/ng-rollout/hole-0/training-token-capture/v1/completions", token_store=token_store
+    )
+
+    assert forwarded == ["/v1/completions"]
+    assert sent[0]["status"] == 200
+    assert token_store.is_incomplete("hole-0")
+
+
+def test_unobserved_dialect_marking_failure_still_forwards(tmp_path):
+    class _BrokenSink:
+        async def mark_incomplete(self, rollout_id, model_call_id=""):
+            raise RuntimeError("sink down")
+
+    forwarded, sent = _run_capture_middleware_on(
+        "/ng-rollout/hole-1/training-token-capture/v1/completions", token_store=_BrokenSink()
+    )
+
+    assert forwarded == ["/v1/completions"]
+    assert sent[0]["status"] == 200
+
+
+def test_observed_dialect_under_capture_prefix_is_not_marked_incomplete(tmp_path):
+    from nemo_gym.token_id_capture import TokenCaptureStore
+
+    token_store = TokenCaptureStore(tmp_path)
+    forwarded, _sent = _run_capture_middleware_on(
+        "/ng-rollout/hole-2/training-token-capture/v1/chat/completions", token_store=token_store
+    )
+
+    assert forwarded == ["/v1/chat/completions"]
+    assert not token_store.is_incomplete("hole-2")
diff --git a/tests/unit_tests/test_rollout_collection.py b/tests/unit_tests/test_rollout_collection.py
index 8ce96be4f7..f171ef26d1 100644
--- a/tests/unit_tests/test_rollout_collection.py
+++ b/tests/unit_tests/test_rollout_collection.py
@@ -13,8 +13,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import json
+import warnings
 from asyncio import Future
 from collections import Counter
+from copy import deepcopy
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock
 
@@ -23,6 +25,7 @@
 import yaml
 
 import nemo_gym.rollout_collection
+import nemo_gym.token_id_capture.delivery
 from nemo_gym.base_resources_server import AggregateMetrics, AggregateMetricsRequest
 from nemo_gym.config_types import ConfigError, ConfigPathNotFoundError
 from nemo_gym.global_config import AGENT_REF_KEY_NAME, ROLLOUT_INDEX_KEY_NAME, TASK_INDEX_KEY_NAME
@@ -45,6 +48,20 @@
     _rollout_request_debug_summary,
     loads_jsonl_line,
 )
+from nemo_gym.token_id_capture import (
+    TokenCaptureSnapshot,
+    TokenCaptureStore,
+    TokenEntry,
+    clear_token_captures_for_rollouts,
+)
+from nemo_gym.token_id_capture.delivery import (
+    MASK_SAMPLE_KEY,
+    TOKEN_CAPTURE_KEY,
+    capture_build_can_retire,
+    finalize_rollout_token_capture,
+    retire_rollout_token_capture,
+    rollout_carries_token_ids,
+)
 
 
 @pytest.fixture
@@ -1051,6 +1068,150 @@ def run_examples(self, examples, *args, **kwargs):
         # The explicit id replaces it.
         assert store.read("0-0") == []
 
+    async def test_run_from_config_does_not_finalize_a_nonparticipating_agent(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        capture_dir = tmp_path / "tokens"
+        monkeypatch.setattr(
+            nemo_gym.rollout_collection,
+            "get_global_config_dict",
+            lambda: {
+                "token_id_capture": {"enabled": True, "dir": str(capture_dir)},
+                "agent": {"responses_api_agents": {"implementation": {"token_id_capture": False}}},
+            },
+        )
+        input_fpath = tmp_path / "input.jsonl"
+        input_fpath.write_bytes(
+            orjson.dumps(
+                {
+                    "responses_create_params": {"input": []},
+                    AGENT_REF_KEY_NAME: {"name": "agent"},
+                }
+            )
+            + b"\n"
+        )
+        config = RolloutCollectionConfig(
+            input_jsonl_fpath=str(input_fpath),
+            output_jsonl_fpath=str(tmp_path / "output.jsonl"),
+            resume_from_cache=False,
+            disable_aggregation=True,
+        )
+
+        class Helper(RolloutCollectionHelper):
+            def run_examples(self, examples, *args, **kwargs):
+                [example] = examples
+                future = Future()
+                future.set_result((example, {"response": {"output": [], "usage": {}}}))
+                return [future]
+
+        [result] = await Helper().run_from_config(config)
+
+        assert MASK_SAMPLE_KEY not in result
+        assert TOKEN_CAPTURE_KEY not in result
+
+    async def test_run_from_config_requires_source_before_dispatch(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(
+            nemo_gym.rollout_collection,
+            "get_global_config_dict",
+            lambda: {
+                "token_id_capture": {
+                    "enabled": True,
+                    "all_agents": True,
+                    "sink": "framework.capture:Sink",
+                    "rebuild_response": True,
+                }
+            },
+        )
+        input_fpath = tmp_path / "input.jsonl"
+        input_fpath.write_bytes(
+            orjson.dumps(
+                {
+                    "responses_create_params": {"input": []},
+                    AGENT_REF_KEY_NAME: {"name": "agent"},
+                }
+            )
+            + b"\n"
+        )
+        config = RolloutCollectionConfig(
+            input_jsonl_fpath=str(input_fpath),
+            output_jsonl_fpath=str(tmp_path / "output.jsonl"),
+            resume_from_cache=False,
+            disable_aggregation=True,
+        )
+
+        class Helper(RolloutCollectionHelper):
+            def run_examples(self, examples, *args, **kwargs):
+                raise AssertionError("Dispatch must not start without a TokenSource.")
+
+        with pytest.raises(ValueError, match="rollout-collector process"):
+            await Helper().run_from_config(config)
+
+    async def test_run_from_config_does_not_close_an_installed_source(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        class Source:
+            closed = False
+
+            async def freeze(self, rollout_id):
+                return TokenCaptureSnapshot(
+                    rollout_id=rollout_id,
+                    entries=(),
+                    incomplete=False,
+                    snapshot_id="snapshot",
+                    version=1,
+                )
+
+            async def drop(self, rollout_id, *, snapshot_id, version):
+                return True
+
+            async def close(self):
+                self.closed = True
+
+        source = Source()
+        monkeypatch.setattr(nemo_gym.rollout_collection, "installed_token_source", lambda: source)
+        monkeypatch.setattr(
+            nemo_gym.rollout_collection,
+            "get_global_config_dict",
+            lambda: {
+                "token_id_capture": {
+                    "enabled": True,
+                    "all_agents": True,
+                    "sink": "framework.capture:Sink",
+                    "rebuild_response": True,
+                }
+            },
+        )
+        input_fpath = tmp_path / "input.jsonl"
+        input_fpath.write_bytes(
+            orjson.dumps(
+                {
+                    "responses_create_params": {"input": []},
+                    AGENT_REF_KEY_NAME: {"name": "agent"},
+                }
+            )
+            + b"\n"
+        )
+        config = RolloutCollectionConfig(
+            input_jsonl_fpath=str(input_fpath),
+            output_jsonl_fpath=str(tmp_path / "output.jsonl"),
+            resume_from_cache=False,
+            disable_aggregation=True,
+        )
+
+        class Helper(RolloutCollectionHelper):
+            def run_examples(self, examples, *args, **kwargs):
+                [example] = examples
+                future = Future()
+                future.set_result((example, {"response": {"output": [], "usage": {}}}))
+                return [future]
+
+        with pytest.warns(UserWarning, match="capture contains no token records"):
+            await Helper().run_from_config(config)
+
+        assert source.closed is False
+
     async def test_run_from_config_sorted(self, tmp_path: Path, empty_global_config: MagicMock) -> None:
         input_jsonl_fpath = tmp_path / "input.jsonl"
         samples = [
@@ -1697,3 +1858,285 @@ async def _noop(self, results, rows, output_fpath):
         # though output_jsonl_fpath is used to derive the metrics path.
         assert not output_fpath.exists()
         assert (tmp_path / "rollouts_aggregate_metrics.json").exists()
+
+
+class TestTokenCaptureRetention:
+    """Test retirement after handoff and stale-record clearing.
+
+    ``TokenCaptureStore.append`` uses append mode.
+    Rollout ids are deterministic.
+    Clearing prevents a rerun from merging different attempts.
+    Retirement prevents unbounded growth after durable handoff.
+    """
+
+    @staticmethod
+    def _entry(rollout_id: str, mcid: str) -> TokenEntry:
+        return TokenEntry(
+            rollout_id=rollout_id,
+            model_call_id=mcid,
+            prompt_token_ids=[1, 2, 3],
+            generation_token_ids=[4, 5],
+            generation_log_probs=[-0.1, -0.2],
+        )
+
+    async def test_clear_removes_stale_records_before_dispatch(self, tmp_path: Path) -> None:
+        store = TokenCaptureStore(tmp_path)
+        store.append(self._entry("0-0", "old"))
+        await store.mark_incomplete("0-0", "old")
+        rows = [{TASK_INDEX_KEY_NAME: 0, ROLLOUT_INDEX_KEY_NAME: 0}]
+
+        clear_token_captures_for_rollouts(rows, [tmp_path])
+
+        assert store.read_entries("0-0") == []
+        assert not store.is_incomplete("0-0")
+
+    def test_clear_is_a_noop_without_capture_dirs(self, tmp_path: Path) -> None:
+        store = TokenCaptureStore(tmp_path)
+        store.append(self._entry("0-0", "keep"))
+        clear_token_captures_for_rollouts([{TASK_INDEX_KEY_NAME: 0, ROLLOUT_INDEX_KEY_NAME: 0}], [])
+        assert len(store.read_entries("0-0")) == 1
+
+    def test_clear_skips_rows_without_a_derivable_rollout_id(self, tmp_path: Path) -> None:
+        store = TokenCaptureStore(tmp_path)
+        store.append(self._entry("0-0", "keep"))
+        clear_token_captures_for_rollouts([{"unrelated": True}], [tmp_path])
+        assert len(store.read_entries("0-0")) == 1
+
+
+class TestFinalizeRolloutTokenCapture:
+    """Test the per-record token-capture finalizer.
+
+    The finalizer accepts a record and a ``TokenSource``.
+    A framework can provide a source without using Gym configuration.
+    """
+
+    @staticmethod
+    def _record(output: list | None = None) -> dict:
+        return {
+            TASK_INDEX_KEY_NAME: 0,
+            ROLLOUT_INDEX_KEY_NAME: 0,
+            "reward": 1.0,
+            "response": {"model": "m", "output": output if output is not None else []},
+        }
+
+    @staticmethod
+    def _capture(store: TokenCaptureStore) -> None:
+        store.append(
+            TokenEntry(
+                rollout_id="0-0",
+                model_call_id="c1",
+                prompt_token_ids=[1, 2, 3],
+                generation_token_ids=[4, 5],
+                generation_log_probs=[-0.1, -0.2],
+                output_items=[{"type": "message", "role": "assistant", "content": []}],
+                token_item_index=0,
+            )
+        )
+
+    async def test_rebuilds_a_rollout_that_has_no_token_ids(self, tmp_path: Path) -> None:
+        store = TokenCaptureStore(tmp_path)
+        self._capture(store)
+        result = self._record()
+
+        built = await finalize_rollout_token_capture(result, store)
+
+        [item] = result["response"]["output"]
+        assert item["generation_token_ids"] == [4, 5]
+        assert result["reward"] == 1.0  # Preserve harness and verifier output.
+        assert result[TOKEN_CAPTURE_KEY]["delivered_fraction"] == 1.0
+        assert built is not None and built["rebuilt_response"] is not None
+        assert len(store.read_entries("0-0")) == 1  # Retain evidence until durable handoff.
+        assert await retire_rollout_token_capture("0-0", store, built) is True
+        assert store.read_entries("0-0") == []
+
+    async def test_retirement_cannot_delete_a_newer_rollout_attempt(self, tmp_path: Path) -> None:
+        store = TokenCaptureStore(tmp_path)
+        self._capture(store)
+        built = await finalize_rollout_token_capture(self._record(), store)
+
+        store.delete("0-0")
+        replacement = TokenEntry(
+            rollout_id="0-0",
+            model_call_id="new",
+            prompt_token_ids=[1],
+            generation_token_ids=[2],
+            generation_log_probs=[-0.1],
+        )
+        store.append(replacement)
+
+        assert await retire_rollout_token_capture("0-0", store, built) is False
+        assert [entry.model_call_id for entry in store.read_entries("0-0")] == ["new"]
+
+    async def test_a_rollout_that_already_has_token_ids_is_left_alone(self, tmp_path: Path) -> None:
+        """Keep the token ids sampled by a native agent.
+
+        A reconstruction may differ from the sampled ids.
+        Overwriting them would silently train on that difference.
+        """
+        store = TokenCaptureStore(tmp_path)
+        self._capture(store)
+        native = [{"type": "message", "role": "assistant", "generation_token_ids": [9, 9], "content": []}]
+        result = self._record(output=native)
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("error")  # Existing ids are not an error.
+            built = await finalize_rollout_token_capture(result, store)
+
+        assert result["response"]["output"] == native
+        assert TOKEN_CAPTURE_KEY not in result
+        assert capture_build_can_retire(built)
+        assert len(store.read_entries("0-0")) == 1
+        assert await retire_rollout_token_capture("0-0", store, built) is True
+        assert store.read_entries("0-0") == []
+
+    async def test_native_and_external_rollouts_are_handled_in_one_batch(self, tmp_path: Path) -> None:
+        """Finalize native and external rollouts through the same call."""
+        store = TokenCaptureStore(tmp_path)
+        self._capture(store)
+        native = self._record(
+            output=[{"type": "message", "role": "assistant", "generation_token_ids": [7], "content": []}]
+        )
+        external = self._record()
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("error")
+            native_build = await finalize_rollout_token_capture(native, store)
+        built = await finalize_rollout_token_capture(external, store)
+
+        assert native["response"]["output"][0]["generation_token_ids"] == [7]
+        assert external["response"]["output"][0]["generation_token_ids"] == [4, 5]
+        assert capture_build_can_retire(native_build)
+        assert built is not None
+
+    async def test_a_second_call_is_a_no_op(self, tmp_path: Path) -> None:
+        """Leave a finalized rollout unchanged on a second call."""
+        store = TokenCaptureStore(tmp_path)
+        self._capture(store)
+        result = self._record()
+
+        await finalize_rollout_token_capture(result, store)
+        rebuilt = deepcopy(result["response"]["output"])
+        second = await finalize_rollout_token_capture(result, store)
+        assert second is not None
+        assert second.get("rebuilt_response") is None
+        assert second.get("_capture_snapshot", {}).get("snapshot_id")
+        assert result["response"]["output"] == rebuilt
+
+    async def test_no_source_means_this_caller_is_not_capturing(self, tmp_path: Path) -> None:
+        result = self._record()
+        with warnings.catch_warnings():
+            warnings.simplefilter("error")
+            assert await finalize_rollout_token_capture(result, None) is None
+        assert result["response"]["output"] == []
+
+    async def test_a_masked_rollout_is_flagged_at_the_top_of_the_record(self, tmp_path: Path) -> None:
+        store = TokenCaptureStore(tmp_path)
+        self._capture(store)
+        # A call that failed to capture leaves a chain that looks contiguous but is missing a turn.
+        await store.mark_incomplete("0-0", "c2")
+        result = self._record()
+
+        with pytest.warns(UserWarning, match="marked for masking"):
+            await finalize_rollout_token_capture(result, store)
+
+        # Keep the masking decision in one top-level field.
+        assert result[MASK_SAMPLE_KEY] is True
+        assert MASK_SAMPLE_KEY not in result[TOKEN_CAPTURE_KEY]
+        assert result[TOKEN_CAPTURE_KEY]["capture_incomplete"] is True
+
+    async def test_a_healthy_rollout_is_not_flagged(self, tmp_path: Path) -> None:
+        store = TokenCaptureStore(tmp_path)
+        self._capture(store)
+        result = self._record()
+
+        await finalize_rollout_token_capture(result, store)
+
+        # Omit the field so presence-based consumers keep healthy samples.
+        assert MASK_SAMPLE_KEY not in result
+
+    async def test_a_failed_build_keeps_its_records_and_reports_why(self, tmp_path: Path) -> None:
+        store = TokenCaptureStore(tmp_path)
+        malformed = TokenEntry(
+            rollout_id="0-0",
+            model_call_id="c1",
+            prompt_token_ids=[1, 2],
+            generation_token_ids=[4, 5],
+            generation_log_probs=[-0.1, -0.2],
+            output_items=[{"type": "message", "role": "assistant", "content": []}],
+            token_item_index=0,
+        )
+        malformed.generation_log_probs = [-0.1]
+        store.append(malformed)
+        result = self._record()
+
+        with pytest.warns(UserWarning, match="marked for masking"):
+            await finalize_rollout_token_capture(result, store)
+
+        assert result[MASK_SAMPLE_KEY] is True
+        assert "ValidationError" in result[TOKEN_CAPTURE_KEY]["error"]
+        # Retain failed-build records as diagnostic evidence.
+        assert store.path_for("0-0").stat().st_size > 0
+
+    async def test_a_rollout_with_no_capture_key_is_masked(self, tmp_path: Path) -> None:
+        result = self._record()
+        del result[TASK_INDEX_KEY_NAME]
+        del result[ROLLOUT_INDEX_KEY_NAME]
+
+        with pytest.warns(UserWarning, match="carries no id"):
+            built = await finalize_rollout_token_capture(result, TokenCaptureStore(tmp_path))
+
+        # Mask the rollout before it reaches the trainer without ids.
+        assert result[MASK_SAMPLE_KEY] is True
+        assert result[TOKEN_CAPTURE_KEY]["error"] == "no capture key"
+        assert built is not None and built["rebuilt_response"] is None
+
+    async def test_nothing_recorded_for_a_rollout_that_needs_ids_is_masked(self, tmp_path: Path) -> None:
+        result = self._record()
+
+        with pytest.warns(UserWarning, match="marked for masking"):
+            built = await finalize_rollout_token_capture(result, TokenCaptureStore(tmp_path))
+
+        assert result[MASK_SAMPLE_KEY] is True
+        assert result[TOKEN_CAPTURE_KEY]["error"] == "capture contains no token records"
+        # Report the rollout as both masked and unbuilt.
+        assert built is not None and built[MASK_SAMPLE_KEY] is True and built["rebuilt_response"] is None
+
+    async def test_a_source_that_raises_loses_one_rollout_not_the_batch(self, tmp_path: Path) -> None:
+        """Keep transport failures scoped to their rollout."""
+
+        class _Failing:
+            async def freeze(self, rollout_id: str):
+                raise ConnectionError("data plane unreachable")
+
+            async def drop(self, rollout_id: str, *, snapshot_id: str, version: int) -> bool:
+                return False
+
+            async def close(self) -> None: ...
+
+        result = self._record()
+
+        with pytest.warns(UserWarning, match="marked for masking"):
+            built = await finalize_rollout_token_capture(result, _Failing())
+
+        assert result[MASK_SAMPLE_KEY] is True
+        assert "ConnectionError" in result[TOKEN_CAPTURE_KEY]["error"]
+        assert built is not None and built["rebuilt_response"] is None
+
+
+class TestRolloutCarriesTokenIds:
+    def test_true_when_any_item_carries_generated_ids(self) -> None:
+        result = {"response": {"output": [{"type": "message"}, {"generation_token_ids": [1]}]}}
+        assert rollout_carries_token_ids(result) is True
+
+    @pytest.mark.parametrize(
+        "response",
+        [
+            {"output": []},
+            {"output": [{"type": "message", "content": []}]},
+            {"output": [{"generation_token_ids": []}]},  # An empty list contains no sampled ids.
+            {},
+            None,
+        ],
+    )
+    def test_false_without_them(self, response) -> None:
+        assert rollout_carries_token_ids({"response": response}) is False
diff --git a/tests/unit_tests/test_token_id_capture.py b/tests/unit_tests/test_token_id_capture.py
index b13f074325..35d628cc9b 100644
--- a/tests/unit_tests/test_token_id_capture.py
+++ b/tests/unit_tests/test_token_id_capture.py
@@ -19,13 +19,13 @@
 Middleware sets a request-scoped token sink.
 The model server records a ``TokenEntry``.
 Consumers read records through ``TokenSource.freeze``.
-There is no HTTP token reader.
 """
 
 import asyncio
 import json
 import logging
 import multiprocessing
+import os
 import subprocess
 import sys
 from time import time
@@ -63,9 +63,11 @@
     current_capture_context,
     extract_token_fields,
     install_token_sink,
+    register_call_intent,
     reset_token_sink,
     set_token_sink,
 )
+from nemo_gym.token_id_capture.config import token_id_capture_enabled_for_agent
 from nemo_gym.token_id_capture.protocols import TokenSource
 from nemo_gym.token_id_capture.store import make_token_store
 
@@ -222,6 +224,51 @@ def test_token_store_recovers_an_unindexed_durable_tail(tmp_path):
     assert set(state["entry_digests"]) == {"c0"}
 
 
+def test_dangling_call_intent_marks_frozen_snapshot_incomplete(tmp_path):
+    store = TokenCaptureStore(tmp_path)
+    asyncio.run(store.begin_call("lost", "c1"))
+
+    snapshot = store.freeze_now("lost")
+
+    assert snapshot.entries == ()
+    assert snapshot.incomplete is True
+
+
+def test_committed_call_satisfies_its_intent(tmp_path):
+    store = TokenCaptureStore(tmp_path)
+    asyncio.run(store.begin_call("complete", "c1"))
+    store.append(
+        TokenEntry(
+            rollout_id="complete",
+            model_call_id="c1",
+            prompt_token_ids=PTOKS,
+            generation_token_ids=GTOKS,
+            generation_log_probs=LPS,
+        )
+    )
+
+    snapshot = store.freeze_now("complete")
+
+    assert [entry.model_call_id for entry in snapshot.entries] == ["c1"]
+    assert snapshot.incomplete is False
+
+
+def test_register_call_intent_uses_optional_sink_extension():
+    calls: list[tuple[str, str]] = []
+
+    class Sink:
+        async def begin_call(self, rollout_id: str, model_call_id: str) -> None:
+            calls.append((rollout_id, model_call_id))
+
+    token = set_token_sink(CaptureContext(rollout_id="r", model_call_id="c", token_sink=Sink()))
+    try:
+        asyncio.run(register_call_intent())
+    finally:
+        reset_token_sink(token)
+
+    assert calls == [("r", "c")]
+
+
 def test_token_store_freeze_is_atomic_and_conditional_drop_is_race_safe(tmp_path):
     store = TokenCaptureStore(tmp_path)
     entry = TokenEntry(
@@ -248,11 +295,9 @@ def test_token_store_freeze_is_atomic_and_conditional_drop_is_race_safe(tmp_path
     assert state["retired"] is True
     assert state["indexed_size"] == 0
     assert state["entry_digests"] == {}
-    retired = asyncio.run(store.freeze("r0"))
-    assert retired.entries == ()
-    assert retired.snapshot_id == updated.snapshot_id
-    assert retired.version == updated.version
-    with pytest.raises(RuntimeError, match="already frozen"):
+    with pytest.raises(RuntimeError, match="retired"):
+        asyncio.run(store.freeze("r0"))
+    with pytest.raises(RuntimeError, match="retired"):
         asyncio.run(store.put(entry.model_copy(update={"model_call_id": "late-after-drop"})))
 
     store.delete("r0")
@@ -261,6 +306,51 @@ def test_token_store_freeze_is_atomic_and_conditional_drop_is_race_safe(tmp_path
     assert store.read_entries("r0") == [replacement]
 
 
+def test_token_store_sweeps_only_old_retired_tombstones(tmp_path):
+    store = TokenCaptureStore(tmp_path)
+    for rollout_id in ("old", "recent", "live"):
+        store.append(
+            TokenEntry(
+                rollout_id=rollout_id,
+                model_call_id=f"{rollout_id}-c1",
+                prompt_token_ids=PTOKS,
+                generation_token_ids=GTOKS,
+                generation_log_probs=LPS,
+            )
+        )
+    for rollout_id in ("old", "recent"):
+        snapshot = store.freeze_now(rollout_id)
+        assert asyncio.run(store.drop(rollout_id, snapshot_id=snapshot.snapshot_id, version=snapshot.version))
+
+    old = time() - 3600
+    os.utime(store.state_path_for("old"), (old, old))
+
+    assert store.sweep_retired(older_than_seconds=600) == 1
+    assert not store.state_path_for("old").exists()
+    assert store.state_path_for("recent").exists()
+    assert store.path_for("live").exists()
+
+
+def test_token_store_recovers_state_lag_from_the_durable_jsonl_tail(tmp_path):
+    store = TokenCaptureStore(tmp_path)
+    first = TokenEntry(
+        rollout_id="lag",
+        model_call_id="c1",
+        prompt_token_ids=PTOKS,
+        generation_token_ids=GTOKS,
+        generation_log_probs=LPS,
+    )
+    store.append(first)
+    state_after_first = store.state_path_for("lag").read_bytes()
+    store.append(first.model_copy(update={"model_call_id": "c2"}))
+
+    store.state_path_for("lag").write_bytes(state_after_first)
+    snapshot = store.freeze_now("lag")
+
+    assert {entry.model_call_id for entry in snapshot.entries} == {"c1", "c2"}
+    assert snapshot.incomplete is False
+
+
 # --- config -------------------------------------------------------------------
 
 
@@ -300,6 +390,35 @@ def test_config_keeps_settings_when_capture_is_off(tmp_path):
     assert cfg.build_sink() is None
 
 
+def test_mask_fraction_limit_defaults_off_and_parses():
+    default = TokenIdCaptureConfig.model_validate(_block(dir="/tmp/token-capture"))
+    configured = TokenIdCaptureConfig.model_validate(_block(dir="/tmp/token-capture", max_mask_fraction=0.5))
+
+    assert default.token_id_capture.max_mask_fraction is None
+    assert configured.token_id_capture.max_mask_fraction == 0.5
+    assert configured.token_id_capture.mask_fraction_min_samples == 50
+
+
+def test_agent_capture_selection_uses_static_agent_config_or_all_agents():
+    config = {
+        "token_id_capture": {"enabled": True, "rebuild_response": False},
+        "captured": {"responses_api_agents": {"implementation": {"token_id_capture": True}}},
+        "ordinary": {"responses_api_agents": {"implementation": {"token_id_capture": False}}},
+    }
+
+    assert token_id_capture_enabled_for_agent(config, "captured")
+    assert not token_id_capture_enabled_for_agent(config, "ordinary")
+    assert not token_id_capture_enabled_for_agent(config, "missing")
+    assert token_id_capture_enabled_for_agent(
+        {**config, "token_id_capture": {**config["token_id_capture"], "all_agents": True}},
+        "ordinary",
+    )
+    assert not token_id_capture_enabled_for_agent(
+        {**config, "token_id_capture": {**config["token_id_capture"], "enabled": False, "all_agents": True}},
+        "captured",
+    )
+
+
 def test_config_warns_rather_than_fails_on_a_sink_beside_a_directory(caplog):
     """Warn when a custom sink replaces the configured directory."""
     with caplog.at_level(logging.WARNING):
@@ -929,7 +1048,7 @@ def test_a_record_is_readable_as_soon_as_put_returns(tmp_path):
         generation_log_probs=[-0.1],
     )
     asyncio.run(store.put(entry))
-    assert [e.model_call_id for e in asyncio.run(store.tokens_for("r0"))] == ["c1"]
+    assert [entry.model_call_id for entry in asyncio.run(store.freeze("r0")).entries] == ["c1"]
 
 
 def test_a_rollout_that_lost_a_call_is_distinguishable_from_a_complete_one(tmp_path):
@@ -1144,7 +1263,7 @@ def test_the_store_is_a_token_source(tmp_path):
             generation_log_probs=[-0.1],
         )
     )
-    assert [e.model_call_id for e in asyncio.run(store.tokens_for("r0"))] == ["c1"]
+    assert [entry.model_call_id for entry in asyncio.run(store.freeze("r0")).entries] == ["c1"]
 
     # A colocated source can detect a capture failure.
     # This prevents training on an incomplete rollout.