nemo_gym.token_id_capture.conformance

Verify that an external token-capture transport follows Gym’s protocols.

A framework transport can replace Gym’s file store. For example, NeMo-RL can carry records through a transfer queue. An incorrect implementation can lose records or make a committed parent invisible to another client. run_conformance checks these behaviors directly. Factories must return fresh client instances over the same shared backend. The checks use fresh instances to verify cross-client visibility. Each check uses its own rollout id, so failed checks cannot poison later ones. This module avoids FastAPI, Ray, Torch, and aiohttp imports.

Module Contents

Classes

Name	Description
`ConformanceError`	One named contract check failed.

Functions

Name	Description
`_check_begin_call_custody`	-
`_check_conditional_retirement`	-
`_check_conflicting_reput`	-
`_check_freeze_idempotency`	-
`_check_idempotent_reput`	-
`_check_lineage_visibility`	-
`_check_mark_incomplete`	-
`_check_post_freeze_write`	-
`_check_put_then_freeze`	-
`_identical_retry`	Simulate a retry that resends the same serialized entry.
`_make_entry`	Build a committed entry with the metadata that `capture_tokens` records.
`_require`	-
`run_conformance`	Run the ordered contract checks and return the names that passed.

Data

_REQUEST

API

class nemo_gym.token_id_capture.conformance.ConformanceError(
    check_name: str,
    detail: str
)

Bases: AssertionError

One named contract check failed.

nemo_gym.token_id_capture.conformance._check_begin_call_custody(
    sink: nemo_gym.token_id_capture.protocols.TokenSink,
    src: nemo_gym.token_id_capture.protocols.TokenSource,
    rollout_id: str
) -> None

async

nemo_gym.token_id_capture.conformance._check_conditional_retirement(
    sink: nemo_gym.token_id_capture.protocols.TokenSink,
    src: nemo_gym.token_id_capture.protocols.TokenSource,
    rollout_id: str
) -> None

async

nemo_gym.token_id_capture.conformance._check_conflicting_reput(
    sink: nemo_gym.token_id_capture.protocols.TokenSink,
    src: nemo_gym.token_id_capture.protocols.TokenSource,
    rollout_id: str
) -> None

async

nemo_gym.token_id_capture.conformance._check_freeze_idempotency(
    sink: nemo_gym.token_id_capture.protocols.TokenSink,
    source_factory: typing.Callable[[], nemo_gym.token_id_capture.protocols.TokenSource],
    rollout_id: str
) -> None

async

nemo_gym.token_id_capture.conformance._check_idempotent_reput(
    sink: nemo_gym.token_id_capture.protocols.TokenSink,
    src: nemo_gym.token_id_capture.protocols.TokenSource,
    rollout_id: str
) -> None

async

nemo_gym.token_id_capture.conformance._check_lineage_visibility(
    sink: nemo_gym.token_id_capture.protocols.TokenSink,
    lineage_factory: typing.Callable[[], nemo_gym.token_id_capture.protocols.LineageStore],
    rollout_id: str,
    fresh_client: bool
) -> None

async

nemo_gym.token_id_capture.conformance._check_mark_incomplete(
    sink: nemo_gym.token_id_capture.protocols.TokenSink,
    source_factory: typing.Callable[[], nemo_gym.token_id_capture.protocols.TokenSource],
    rollout_id: str
) -> None

async

nemo_gym.token_id_capture.conformance._check_post_freeze_write(
    sink: nemo_gym.token_id_capture.protocols.TokenSink,
    src: nemo_gym.token_id_capture.protocols.TokenSource,
    rollout_id: str
) -> None

async

nemo_gym.token_id_capture.conformance._check_put_then_freeze(
    sink: nemo_gym.token_id_capture.protocols.TokenSink,
    src: nemo_gym.token_id_capture.protocols.TokenSource,
    rollout_id: str
) -> None

async

nemo_gym.token_id_capture.conformance._identical_retry(
    entry: nemo_gym.token_id_capture.records.TokenEntry
) -> nemo_gym.token_id_capture.records.TokenEntry

Simulate a retry that resends the same serialized entry.

nemo_gym.token_id_capture.conformance._make_entry(
    rollout_id: str,
    model_call_id: str,
    prompt: list[int],
    generation: list[int],
    request_items: list[dict],
    text: str
) -> nemo_gym.token_id_capture.records.TokenEntry

Build a committed entry with the metadata that capture_tokens records.

nemo_gym.token_id_capture.conformance._require(
    condition: bool,
    check_name: str,
    detail: str
) -> None

nemo_gym.token_id_capture.conformance.run_conformance(
    sink_factory: typing.Callable[[], nemo_gym.token_id_capture.protocols.TokenSink],
    source_factory: typing.Callable[[], nemo_gym.token_id_capture.protocols.TokenSource],
    lineage_factory: typing.Callable[[], nemo_gym.token_id_capture.protocols.LineageStore] | None = None,
    rollout_id: str = 'conformance-rollout'
) -> list[str]

async

Run the ordered contract checks and return the names that passed.

Raise ConformanceError on the first failure. Lineage checks are skipped without a lineage_factory. The begin_call check is skipped when the sink lacks the extension.

nemo_gym.token_id_capture.conformance._REQUEST = [{'role': 'user', 'content': 'What is the weather in Paris?'}]