nemo_gym.web.session_control

Own web seed requests independently of the HTTP client’s lifetime.

This is a web adapter, not a replacement for Gym’s proposed core lifecycle API. Keep task, browser and evaluator ownership in WebSessionManager.

Module Contents

Classes

Name	Description
`SessionIdentityError`	The request does not authorize access to the named web session.
`WebSessionControl`	Deduplicate seed, authorize cookie-less close, and retain tombstones.
`_SessionRequest`	-

Data

LOG

API

class nemo_gym.web.session_control.SessionIdentityError()

Bases: RuntimeError

The request does not authorize access to the named web session.

class nemo_gym.web.session_control.WebSessionControl(
    manager: nemo_gym.web.session_manager.WebSessionManager,
    lifetime_seconds: float,
    max_records: int = 10000
)

Deduplicate seed, authorize cookie-less close, and retain tombstones.

_reaper

Task[None] | None = None

_records

dict[str, _SessionRequest] = {}

nemo_gym.web.session_control.WebSessionControl._close_record(
    session_id: str,
    record: nemo_gym.web.session_control._SessionRequest
) -> bool

async

nemo_gym.web.session_control.WebSessionControl._consume_exception(
    task: asyncio.Task
) -> None

staticmethod

nemo_gym.web.session_control.WebSessionControl._finish_close(
    session_id: str,
    record: nemo_gym.web.session_control._SessionRequest
) -> bool

async

nemo_gym.web.session_control.WebSessionControl._prune() -> None

nemo_gym.web.session_control.WebSessionControl._reap() -> None

async

nemo_gym.web.session_control.WebSessionControl._reap_once() -> None

async

nemo_gym.web.session_control.WebSessionControl._record(
    identity: nemo_gym.web.api_models.WebSessionIdentity
) -> tuple[str, nemo_gym.web.session_control._SessionRequest]

nemo_gym.web.session_control.WebSessionControl.close(
    identity: nemo_gym.web.api_models.WebSessionIdentity
) -> bool

async

nemo_gym.web.session_control.WebSessionControl.seed(
    body: nemo_gym.web.api_models.WebSeedSessionRequest
) -> nemo_gym.web.api_models.WebSeedSessionResponse

async

nemo_gym.web.session_control.WebSessionControl.start() -> None

nemo_gym.web.session_control.WebSessionControl.stop(
    timeout: float
) -> None

async

class nemo_gym.web.session_control._SessionRequest(
    token_digest: bytes,
    created_at: float,
    fingerprint: str | None = None,
    seed: asyncio.Task[nemo_gym.web.api_models.WebSeedSessionResponse] | None = None,
    close: asyncio.Task[bool] | None = None,
    closing: bool = False,
    closed_at: float | None = None,
    expiry_close_attempts: int = 0
)

Dataclass

Task[bool] | None = None

closed_at

float | None = None

closing

bool = False

created_at

float

expiry_close_attempts

int = 0

fingerprint

str | None = None

seed

Task[WebSeedSessionResponse] | None = None

token_digest

bytes

nemo_gym.web.session_control.LOG = logging.getLogger(__name__)