pynenc.trigger.monitoring

Monitoring DTOs for the trigger component.

These dataclasses are the durable monitoring contract that trigger backends expose to any visibility / observability consumer. They follow the same shape as InvocationHistory and other pynenc DTOs: backend-neutral, JSON round-trippable, UTC-aware timestamps, and no dependency on any specific UI.

Module contents:

• class:

  EventRecord: durable record of one emitted event.

• class:

  EventMarker: lightweight projection used by timeline overlays.

• class:

  EventMarkerPage: paginated marker projection with truncation flag.

• class:

  TriggerRunRecord: durable record of one trigger run.

• class:

  TriggerRunParticipant: per-condition row attached to a trigger run preserving which condition each event or source invocation satisfied.

Design notes:

• EventRecord.triggered_invocation_ids is rehydrated on read from the per-event invocation index rather than rewritten on the hot path.

• class:

  TriggerRunParticipant stores only reference ids and lightweight labels. Detail views re-fetch the live TriggerCondition / context through :meth:BaseTrigger.get_condition and the event / state-backend lookups when they need full payloads.

Module Contents

Classes

EventRecord	Durable record of an emitted event.
EventMarker	Lightweight projection of an event used by the timeline overlay.
EventMarkerPage	Paginated event-marker projection with explicit truncation metadata.
TriggerRunParticipant	Per-condition row attached to a :class:TriggerRunRecord.
TriggerRunRecord	Durable record of a single trigger run.

Functions

_now_utc	Return a UTC timezone-aware datetime for default_factory usage.
_parse_timestamp	Parse an ISO timestamp string, defaulting to “now” when missing.
_parse_timestamp_optional	Parse an ISO timestamp string into UTC, returning None when missing.
_iso_or_none	Return value.isoformat() or None.

API

pynenc.trigger.monitoring._now_utc() → datetime.datetime

Return a UTC timezone-aware datetime for default_factory usage.

class pynenc.trigger.monitoring.EventRecord

Durable record of an emitted event.

Created by :meth:pynenc.trigger.base_trigger.BaseTrigger.emit_event regardless of whether any condition matched. The first write captures the payload; a single follow-up write records matched/valid condition IDs when at least one condition accepts the event.

triggered_invocation_ids is populated by the backend on read from a dedicated event->invocation index. The hot path does not rewrite the event JSON to append invocation IDs.

event_id: str

None

event_code: str

None

payload: dict[str, Any]

‘field(…)’

timestamp: datetime.datetime

‘field(…)’

matched_condition_ids: list[str]

‘field(…)’

valid_condition_ids: list[str]

‘field(…)’

triggered_invocation_ids: list[str]

‘field(…)’

emitted_by_invocation_id: str | None

None

emitted_by_task_id: str | None

None

emitted_by_runner_context_id: str | None

None

property matched: bool

True when at least one event condition accepted this event.

property triggered: bool

True when at least one invocation was created from this event.

to_json(app: pynenc.app.Pynenc) → str

Serialize to JSON, routing payload values through the data store.

_to_dict(app: pynenc.app.Pynenc) → dict[str, Any]

Return a JSON-ready dict using app.client_data_store for payload.

classmethod from_json(json_str: str, app: pynenc.app.Pynenc) → pynenc.trigger.monitoring.EventRecord

Rebuild an :class:EventRecord from its JSON representation.

classmethod _from_dict(data: dict[str, Any], app: pynenc.app.Pynenc) → pynenc.trigger.monitoring.EventRecord

class pynenc.trigger.monitoring.EventMarker

Lightweight projection of an event used by the timeline overlay.

Backends populate this directly from indexed columns without deserializing the full event payload. The marker is intentionally narrow so a busy timeline window can load many markers cheaply.

event_id: str

None

event_code: str

None

timestamp: datetime.datetime

None

matched: bool

None

triggered: bool

None

emitted_by_invocation_id: str | None

None

emitted_by_runner_context_id: str | None

None

class pynenc.trigger.monitoring.EventMarkerPage

Paginated event-marker projection with explicit truncation metadata.

truncated is True when the backend stopped at limit and more markers exist for the requested window. Pynmon uses this to surface a visible “showing N of more” hint instead of silently dropping markers.

markers: list[pynenc.trigger.monitoring.EventMarker]

None

total: int

None

truncated: bool

None

class pynenc.trigger.monitoring.TriggerRunParticipant

Per-condition row attached to a :class:TriggerRunRecord.

Preserves the mapping between a valid condition and the event or source invocation that satisfied it. Composite (AND) triggers need this to answer “which condition did each participant satisfy?”.

Exactly one of event_id / source_invocation_id is populated for event/status/result/exception conditions. CronContext and other contexts without a discrete source populate neither.

context_timestamp and context_summary are lightweight labels so the timeline overlay and tooltip rows can anchor and describe a participant without re-fetching the source record. Full condition and context payloads are re-fetched on demand from the trigger registry and the event / state-backend lookups when a detail panel is opened.

context_type: str

None

condition_id: str | None

None

valid_condition_id: str | None

None

event_id: str | None

None

source_invocation_id: str | None

None

context_timestamp: datetime.datetime | None

None

context_summary: str = <Multiline-String>

to_dict() → dict[str, Any]

classmethod from_dict(data: dict[str, Any]) → pynenc.trigger.monitoring.TriggerRunParticipant

class pynenc.trigger.monitoring.TriggerRunRecord

Durable record of a single trigger run.

Captures the participants that satisfied the trigger and the resulting invocation. participants preserves the per-condition mapping that the parallel event_ids / source_invocation_ids lists lose. The parallel lists are kept for backward compatibility and as a flat index for backends that do not need the per-condition mapping.

trigger_run_id: str

None

trigger_id: str

None

task_id_key: str

None

logic_value: str

None

valid_condition_ids: list[str]

‘field(…)’

condition_ids: list[str]

‘field(…)’

event_ids: list[str]

‘field(…)’

source_invocation_ids: list[str]

‘field(…)’

triggered_invocation_id: str | None

None

arguments_preview: dict[str, Any]

‘field(…)’

claimed_at: datetime.datetime | None

None

executed_at: datetime.datetime | None

None

participants: list[pynenc.trigger.monitoring.TriggerRunParticipant]

‘field(…)’

atomic_service_run_id: str | None

None

atomic_service_runner_id: str | None

None

to_json() → str

Serialize to JSON. Arguments preview is stored as plain JSON.

classmethod from_json(json_str: str) → pynenc.trigger.monitoring.TriggerRunRecord

Rebuild a :class:TriggerRunRecord from its JSON representation.

pynenc.trigger.monitoring._parse_timestamp(value: Any) → datetime.datetime

Parse an ISO timestamp string, defaulting to “now” when missing.

pynenc.trigger.monitoring._parse_timestamp_optional(value: Any) → datetime.datetime | None

Parse an ISO timestamp string into UTC, returning None when missing.

pynenc.trigger.monitoring._iso_or_none(value: datetime.datetime | None) → str | None

Return value.isoformat() or None.