nemo_gym.observability.model_calls

Read saved model exchanges, resolve ownership, and attach evidence to rollouts.

Module Contents

Classes

Name	Description
`ModelCallRecord`	Observability record derived from one captured model-server exchange.

Functions

Name	Description
`_as_arguments`	-
`_cache_signal`	Cache hit/miss + cached-token count, from usage cache fields (OpenAI / Anthropic).
`_store_for_rollout`	-
`_token_count`	-
`_tool_calls_and_reasoning`	Structured tool calls (name, arguments, call_id) and reasoning text, across all three shapes.
`_usage_detail_token`	Return the first valid token count across equivalent provider detail shapes.
`aggregate_model_call_metrics`	Aggregate model-call metrics for one rollout id.
`aggregate_model_call_records`	Aggregate token and latency values from model-call records.
`await_model_call_capture_quiescence`	Wait until no claimed capture write for this rollout is outstanding.
`build_model_call_record`	Map one captured exchange and its transport metadata into an observability record.
`clear_model_call_captures_for_rollouts`	Remove stale per-rollout capture files for these records before dispatch.
`extract_token_stats`	Normalize token totals across Responses, Chat Completions, and Anthropic Messages usage.
`merge_model_call_capture_into_record`	Attach captured model-call observability data to a rollout record in place.
`model_call_capture_dirs_from_config`	Return the single run-wide capture directory when capture is enabled.
`observability_enabled_from_config`	Return the run-wide `observability_enabled` flag directly, without going through capture dirs.
`read_available_model_call_records`	Read valid call records and count damaged records.
`read_model_call_records`	Read captured exchanges in durable append order.

Data

CAPTURE_DRAIN_TIMEOUT_S

_CAPTURE_DRAIN_POLL_S

logger

API

class nemo_gym.observability.model_calls.ModelCallRecord()

Bases: BaseModel

Observability record derived from one captured model-server exchange.

cache_creation_tokens

Optional[int] = None

cache_hit

Optional[bool] = None

cached_tokens

Optional[int] = None

call_index

int

client_assistant_message_id

Optional[str]

client_session_id

Optional[str]

completed_at

Optional[float] = None

dialect

Optional[str] = None

error_category

Optional[str] = None

finish_reason

Optional[str] = None

latency_total_ms

Optional[float] = None

latency_ttft_ms

Optional[float] = None

local_response_reason

Optional[str] = None

model

Optional[str] = None

model_call_id

Optional[str] = None

model_ref

Optional[ModelServerRef] = None

reasoning_content

Optional[str] = None

request

Optional[dict[str, Any]] = None

request_raw

Optional[str] = None

response

Optional[dict[str, Any]] = None

response_id

Optional[str] = None

response_raw

Optional[str] = None

response_source

Optional[Literal['upstream', 'local']] = None

response_status

Optional[str] = None

started_at

Optional[float] = None

status_code

Optional[int] = None

tokens_in

Optional[int] = None

tokens_out

Optional[int] = None

tokens_reasoning

Optional[int] = None

tokens_total

Optional[int] = None

tool_calls

list[dict[str, Any]] = Field(default_factory=list)

upstream_attempted

Optional[bool] = None

upstream_status_code

Optional[int] = None

nemo_gym.observability.model_calls._as_arguments(
    arguments: typing.Any
) -> dict[str, typing.Any]

nemo_gym.observability.model_calls._cache_signal(
    usage: typing.Any
) -> tuple[typing.Optional[bool], typing.Optional[int]]

Cache hit/miss + cached-token count, from usage cache fields (OpenAI / Anthropic).

nemo_gym.observability.model_calls._store_for_rollout(
    rollout_id: str,
    capture_dirs: list[pathlib.Path]
) -> typing.Optional[nemo_gym.base_responses_api_model.CaptureStore]

nemo_gym.observability.model_calls._token_count(
    value: typing.Any
) -> typing.Optional[int]

nemo_gym.observability.model_calls._tool_calls_and_reasoning(
    response: dict[str, typing.Any]
) -> tuple[list[dict[str, typing.Any]], typing.Optional[str]]

Structured tool calls (name, arguments, call_id) and reasoning text, across all three shapes.

nemo_gym.observability.model_calls._usage_detail_token(
    usage: typing.Mapping[str, typing.Any],
    detail_groups: tuple[str, ...],
    field_names: tuple[str, ...]
) -> typing.Optional[int]

Return the first valid token count across equivalent provider detail shapes.

nemo_gym.observability.model_calls.aggregate_model_call_metrics(
    store: nemo_gym.base_responses_api_model.CaptureStore,
    rollout_id: str
) -> dict[str, typing.Any]

Aggregate model-call metrics for one rollout id.

nemo_gym.observability.model_calls.aggregate_model_call_records(
    calls: list[nemo_gym.observability.model_calls.ModelCallRecord]
) -> dict[str, typing.Any]

Aggregate token and latency values from model-call records.

nemo_gym.observability.model_calls.await_model_call_capture_quiescence(
    rollout_id: str,
    capture_dirs: list[pathlib.Path],
    timeout_s: float = CAPTURE_DRAIN_TIMEOUT_S,
    poll_s: float = _CAPTURE_DRAIN_POLL_S
) -> bool

async

Wait until no claimed capture write for this rollout is outstanding.

A cancelled model call finishes its capture write detached (see the shield in _CaptureMiddleware), so the row can land after the rollout record is snapshotted and be missing from ng_model_call_capture while present on disk. Returns False on timeout, which the merge turns into a model_call_capture_pending gap rather than a silent omission.

nemo_gym.observability.model_calls.build_model_call_record(
    exchange: dict[str, typing.Any],
    call_index: int
) -> nemo_gym.observability.model_calls.ModelCallRecord

Map one captured exchange and its transport metadata into an observability record.

nemo_gym.observability.model_calls.clear_model_call_captures_for_rollouts(
    records: list[typing.Any],
    capture_dirs: list[pathlib.Path]
) -> None

Remove stale per-rollout capture files for these records before dispatch.

Capture files are keyed by a deterministic rollout id (task-rollout-attempt), so without this a fresh run or a kill-shaped retry would append onto the previous attempt’s capture for the same id. The caller passes only rows about to be dispatched, after assigning any retry suffix.

nemo_gym.observability.model_calls.extract_token_stats(
    usage: typing.Any
) -> dict[str, typing.Optional[int]]

Normalize token totals across Responses, Chat Completions, and Anthropic Messages usage.

For native Anthropic /v1/messages with prompt caching, input_tokens is only the uncached remainder, so cache-read + cache-creation tokens are folded into tokens_in to reflect the true prompt size (and cache-creation is surfaced separately as cache_creation_tokens). OpenAI / Responses usage already includes cached tokens in input_tokens / prompt_tokens (where cached_tokens is a subset), so it is left untouched — no double counting.

tokens_in is a prompt-size metric, not a cost proxy: providers price cache-read (~0.1x) and cache-creation (~1.25x) differently from base input, so cost-accurate consumers should weight cached_tokens and cache_creation_tokens separately rather than summing tokens_in.

nemo_gym.observability.model_calls.merge_model_call_capture_into_record(
    record: dict[str, typing.Any],
    capture_dirs: list[pathlib.Path],
    include_payloads: bool = False
) -> dict[str, typing.Any]

Attach captured model-call observability data to a rollout record in place.

Keyed by the rollout id derived from the record’s task/rollout/attempt indices, so the attached shape is identical for every agent harness. Adds ng_model_call_capture = {rollout_id, metrics, calls} where calls are derived observability records. Raw request and response payloads remain in the capture store and are omitted from the attachment unless include_payloads is true. Capture/read/join failures are attached as gaps. The harness output and reward are not modified.

nemo_gym.observability.model_calls.model_call_capture_dirs_from_config(
    global_config_dict: typing.Any
) -> list[pathlib.Path]

Return the single run-wide capture directory when capture is enabled.

nemo_gym.observability.model_calls.observability_enabled_from_config(
    global_config_dict: typing.Any
) -> bool

Return the run-wide observability_enabled flag directly, without going through capture dirs.

nemo_gym.observability.model_calls.read_available_model_call_records(
    store: nemo_gym.base_responses_api_model.CaptureStore,
    rollout_id: str
) -> tuple[list[nemo_gym.observability.model_calls.ModelCallRecord], int]

Read valid call records and count damaged records.

nemo_gym.observability.model_calls.read_model_call_records(
    store: nemo_gym.base_responses_api_model.CaptureStore,
    rollout_id: str
) -> list[nemo_gym.observability.model_calls.ModelCallRecord]

Read captured exchanges in durable append order.

nemo_gym.observability.model_calls.CAPTURE_DRAIN_TIMEOUT_S = 10.0

nemo_gym.observability.model_calls._CAPTURE_DRAIN_POLL_S = 0.05

nemo_gym.observability.model_calls.logger = logging.getLogger(__name__)