nemo_gym.agent_utils.ordered_operations

In-process exactly-once ordered operations for agent activations.

Module Contents

Classes

Name	Description
`ActivationConflictError`	An activation conflicts with its session’s ordering or immutable input.
`OrderedOperationLedger`	Join/replay identical calls, reject conflicts, and fence work before closing.
`_Operation`	-

API

class nemo_gym.agent_utils.ordered_operations.ActivationConflictError()

Bases: ValueError

An activation conflicts with its session’s ordering or immutable input.

class nemo_gym.agent_utils.ordered_operations.OrderedOperationLedger()

Join/replay identical calls, reject conflicts, and fence work before closing.

Owners must serialize seed/close binding separately and retain this ledger for the session lifetime. Disconnecting a waiter leaves execution alive. Failed operations are retained and replay their failure; a new physical episode is needed to retry with different inputs. This is process-local, not crash recovery.

_lock

= asyncio.Lock()

_operations

list[_Operation[Result]] = []

nemo_gym.agent_utils.ordered_operations.OrderedOperationLedger.close(
    timeout: float
) -> None

async

Fence new work, cancel/await the active operation, then allow owner cleanup.

A timeout retains the active task and the fence, so cleanup cannot be declared successful while an activation might still launch remotely.

nemo_gym.agent_utils.ordered_operations.OrderedOperationLedger.execute(
    index: int,
    request: pydantic.BaseModel,
    operation: collections.abc.Callable[[], collections.abc.Awaitable[nemo_gym.agent_utils.ordered_operations.OrderedOperationLedger[Result]]]
) -> nemo_gym.agent_utils.ordered_operations.OrderedOperationLedger[Result]

async

Run the next operation once or join an exact retry, without holding the lock while running.

class nemo_gym.agent_utils.ordered_operations._Operation(
    request: pydantic.BaseModel,
    task: asyncio.Task[nemo_gym.agent_utils.ordered_operations._Operation[Result]]
)

Dataclass

request

BaseModel

task

Task[Result]