nat.atof.events#

ATOF event models for the 2 event kinds per spec v0.1.

Standalone Pydantic models for each event kind. The Event type is a discriminated union keyed on the kind field.

Two event kinds:

ScopeEvent — a scope lifecycle event (start or end, distinguished by scope_category). A start/end pair shares the same uuid (spec §5.3).
MarkEvent — a point-in-time checkpoint, unpaired.

What kind of work an event represents is carried by the category field. Category-specific typed fields are packaged into a single optional category_profile sub-object (spec §4.4) — model_name for llm, tool_call_id for tool, subtype for custom, with additional keys reserved for future categories. category is REQUIRED on ScopeEvent and OPTIONAL on MarkEvent.

See ATOF spec:

§2 (common envelope), §2.1 (attributes)
§3 (event kinds)
§4 (category vocabulary)
§5 (event stream semantics)

Attributes#

`_ATOF_VERSION_PATTERN`
`_CANONICAL_CATEGORIES`
`Event`	Discriminated union of the 2 ATOF event kinds, keyed on `kind` (spec §3).

Classes#

`_EventBase`	Common fields shared by all ATOF event types (spec §2).
`ScopeEvent`	Scope lifecycle event (spec §3.1).
`MarkEvent`	Point-in-time checkpoint (spec §3.2).

Functions#

`_canonicalize_attributes`(→ list[str])	Normalize an `attributes` field to a sorted, deduplicated list of strings.
`_require_subtype_when_custom`(→ None)	Enforce §4.2: when `category == "custom"`, `category_profile.subtype` is REQUIRED.
`_get_event_kind`(→ str)	Extract the discriminator value from a raw dict or model instance.

Module Contents#

_ATOF_VERSION_PATTERN#

_CANONICAL_CATEGORIES: frozenset[str]#

_canonicalize_attributes(v: Any) → list[str]#

Normalize an attributes field to a sorted, deduplicated list of strings.

Accepts either a list of strings or Flags StrEnum members. Unknown flag names are preserved — the spec requires consumers to round-trip them.

_require_subtype_when_custom( category: str | None, category_profile: dict[str, Any] | None, ) → None#: Enforce §4.2: when category == "custom", category_profile.subtype is REQUIRED.

class _EventBase(/, **data: Any)#

Bases: pydantic.BaseModel

Common fields shared by all ATOF event types (spec §2).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

atof_version: str = None#

uuid: str = None#

parent_uuid: str | None = None#

timestamp: str | int = None#

name: str = None#

data: Any | None = None#

data_schema: dict[str, Any] | None = None#

metadata: dict[str, Any] | None = None#

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod _validate_atof_version(v: str) → str#

classmethod _validate_uuid_non_empty(v: str | None) → str | None#

classmethod _validate_timestamp(v: Any) → str | int#

property ts_micros: int#

Timestamp normalized to int epoch microseconds (spec §5.1).

Not emitted on the wire (excluded by io.write_jsonl). For in-memory sorting and consumer-side comparison only.

class ScopeEvent(/, **data: Any)#

Bases: _EventBase

Scope lifecycle event (spec §3.1).

A single scope span produces two ScopeEvent instances sharing the same uuid: one with scope_category: "start" when the scope is pushed onto the active scope stack, and one with scope_category: "end" when the scope is popped.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

kind: Literal['scope'] = 'scope'#

scope_category: Literal['start', 'end'] = None#

attributes: list[str] = None#

category: str = None#

category_profile: dict[str, Any] | None = None#

classmethod _canonicalize_attributes_field(v: Any) → list[str]#

classmethod _validate_category(v: str) → str#

_validate_category_subtype_coherence() → Self#

class MarkEvent(/, **data: Any)#

Bases: _EventBase

Point-in-time checkpoint (spec §3.2).

Unpaired (no start/end semantics). MAY carry category + category_profile to indicate the kind of work the checkpoint relates to; when both are absent, the mark is a generic named timestamp. Does NOT carry scope_category or attributes.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

kind: Literal['mark'] = 'mark'#

category: str | None = None#

category_profile: dict[str, Any] | None = None#

_validate_category_subtype_coherence() → Self#

_reject_scope_only_fields() → Self#

_get_event_kind(v: Any) → str#: Extract the discriminator value from a raw dict or model instance.

Event#: Discriminated union of the 2 ATOF event kinds, keyed on kind (spec §3).