nat.cli.telemetry_hook

CLI-side telemetry plumbing.

This module exists so that telemetry concerns stay out of entrypoint.py and main.py. The integration is a thin pair of helpers:

• record_invocation_start() — called by the Click group callback, stashes a session ID and timestamp on the Click context.

• emit_command_event() — called by the entry-point wrapper after the CLI exits (success, interrupt, or failure), constructs and enqueues a CliCommandEvent.

All public functions swallow exceptions: telemetry must never disrupt the host CLI invocation.

Attributes

logger
_CTX_SESSION_ID
_CTX_START_TIME
_CTX_COMMAND
_CTX_ROOT_COMMAND

Functions

record_invocation_start(→ None)	Stash the session ID, start time, and invoked command on ctx.obj.
emit_command_event(→ None)	Build and enqueue a CliCommandEvent for the just-finished call.
_resolve_subcommand(→ str | None)	Recover the second-level command name, validated against the registered

Module Contents

logger

_CTX_SESSION_ID = 'telemetry_session_id'

_CTX_START_TIME = 'telemetry_start_time'

_CTX_COMMAND = 'telemetry_command'

_CTX_ROOT_COMMAND = 'telemetry_root_command'

record_invocation_start(ctx: click.Context) → None

Stash the session ID, start time, and invoked command on ctx.obj.

Safe to call unconditionally; does nothing meaningful when telemetry is disabled, but keeping the bookkeeping unconditional makes the call sites simpler.

Also stashes the root Click group so that _resolve_subcommand() can validate any second-level token against the set of actually registered subcommand names. This is the privacy boundary that prevents user-supplied positional arguments (file paths, workflow names, queries) from leaking into the telemetry payload.

emit_command_event(

ctx_obj: dict | None,

*,

task_status: nat.utils.telemetry.TaskStatusEnum,

exit_code: int,

error_class: str | None = None,

) → None

Build and enqueue a CliCommandEvent for the just-finished call.

Parameters

ctx_obj:

The mutable dict that was used as Click’s obj. Reads back the identifiers stashed by record_invocation_start(). May be None or empty if Click never reached the group callback (e.g. argument parse error before any subcommand resolved); we still emit an event in that case with command="unknown".

task_status:

Outcome of the invocation.

exit_code:

Process exit code we are about to return.

error_class:

Exception class name on failure (no message). None otherwise.

_resolve_subcommand(

root_command: object | None,

top_level: str,

) → str | None

Recover the second-level command name, validated against the registered Click command tree.

Click does not expose nested invoked_subcommand from the root context, so we scan sys.argv for a candidate token. We only return a token if it matches the name of an actually registered subcommand of the top_level group. This is the privacy boundary: positional arguments (file paths, workflow names, free-form queries) cannot match a registered subcommand name and therefore cannot leak.

To prevent option-value tokens that appear later on the command line from being misclassified as the subcommand, the scan only considers the first non-flag token after top_level. Anything later — even if it happens to spell a subcommand name — is ignored.

Known limitation: an option whose value (e.g. --filter list-components) appears before the subcommand position can still be picked up here, because we cannot infer Click option metadata from raw argv. The exposure is bounded — the value must exactly match a registered subcommand name — and is fully eliminated only by switching to Click’s parsed context tree.

Returns None when:

• The root command isn’t a Click group (callable from non-CLI contexts).

• top_level isn’t a registered subcommand of the root group.

• The resolved top-level command is itself a leaf (not a group), so no second level is possible.

• The first non-flag token after top_level doesn’t match a registered subcommand.

• top_level doesn’t appear in sys.argv at all.