nemo_gym.models.token_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.
`ConformanceWarning`	One named contract check was skipped, and the gap has a consequence at run time.

Functions

Name	Description
`_check_begin_call_custody`	-
`_check_conditional_retirement`	-
`_check_conflicting_reput`	-
`_check_delete_clears_retirement`	-
`_check_freeze_idempotency`	-
`_check_idempotent_reput`	-
`_check_lineage_visibility`	-
`_check_mark_incomplete`	-
`_check_post_freeze_write`	-
`_check_put_then_freeze`	-
`_check_refuse_call_resolution`	-
`_check_unconditional_retirement`	-
`_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`	-
`_require_fenced`	A write to a retired rollout must fail with `TokenCaptureRetiredError`, as `TokenSource.retire` requires.
`_visible_entries`	Entries a consumer could build from; a retired rollout may refuse to freeze instead.
`run_conformance`	Run the ordered contract checks and return the names that passed.

Data

_REQUEST

API

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

Bases: AssertionError

One named contract check failed.

class nemo_gym.models.token_capture.conformance.ConformanceWarning(
    check_name: str,
    detail: str
)

Bases: UserWarning

One named contract check was skipped, and the gap has a consequence at run time.

run_conformance issues it through warnings.warn after every check that ran passed, so the transport conforms and the caller still sees what the missing extension costs.

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

async

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

async

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

async

nemo_gym.models.token_capture.conformance._check_delete_clears_retirement(
    sink: nemo_gym.models.token_capture.protocols.TokenSink,
    src: nemo_gym.models.token_capture.protocols.TokenSource,
    rollout_id: str
) -> None

async

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

async

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

async

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

async

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

async

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

async

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

async

nemo_gym.models.token_capture.conformance._check_refuse_call_resolution(
    sink: nemo_gym.models.token_capture.protocols.TokenSink,
    src: nemo_gym.models.token_capture.protocols.TokenSource,
    rollout_id: str
) -> None

async

nemo_gym.models.token_capture.conformance._check_unconditional_retirement(
    sink: nemo_gym.models.token_capture.protocols.TokenSink,
    src: nemo_gym.models.token_capture.protocols.TokenSource,
    rollout_id: str
) -> None

async

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

Simulate a retry that resends the same serialized entry.

nemo_gym.models.token_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.models.token_capture.records.TokenEntry

Build a committed entry with the metadata that capture_tokens records.

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

nemo_gym.models.token_capture.conformance._require_fenced(
    name: str,
    call: typing.Awaitable[object],
    what: str
) -> None

async

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.

nemo_gym.models.token_capture.conformance._visible_entries(
    src: nemo_gym.models.token_capture.protocols.TokenSource,
    rollout_id: str
) -> int

async

Entries a consumer could build from; a retired rollout may refuse to freeze instead.

nemo_gym.models.token_capture.conformance.run_conformance(
    sink_factory: typing.Callable[[], nemo_gym.models.token_capture.protocols.TokenSink],
    source_factory: typing.Callable[[], nemo_gym.models.token_capture.protocols.TokenSource],
    lineage_factory: typing.Callable[[], nemo_gym.models.token_capture.protocols.LineageResolver] | 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 and record_engine_refusal checks are skipped when the sink lacks the extension. A sink that implements begin_call without record_engine_refusal passes with a ConformanceWarning for the skipped check: a call the engine refuses then leaves its pre-dispatch intent dangling, and the rollout is masked.

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