roboto.domain.triggers#

Triggers: fire on platform events or a schedule, filter with a condition, dispatch targets.

A trigger pairs a firing source — a subscription to one or more platform events (fired at most once per the subscription’s once_per) or a cron schedule — with an optional condition over the firing’s variable namespace and one or more targets to run on a match.

The event catalog describes what each event type exposes — payload model, namespace roots, supported once_per values — and drives both save-time validation and evaluation. Conditions reuse the Condition wire format; target templates reuse the roboto.templating {{...}} placeholder syntax.

roboto.domain.actions.Trigger is the legacy model of the same thing (one action, causes/for_each), deprecated and kept only for existing code; it cannot read a trigger the current model added capabilities to. Trigger here reads them all.

Submodules#

Package Contents#

class roboto.domain.triggers.ConditionLeafTrace(/, **data)#

Bases: pydantic.BaseModel

One leaf of the trigger’s condition, with the actual value it saw.

Parameters:

data (Any)

actual: Any | None = None#

The value the event’s namespace resolved for field; None when the field did not resolve (missing entity, missing key).

comparator: roboto.query.Comparator#

The leaf’s comparator.

expected: Any | None = None#

The value the condition compares against.

field: str#

The condition field, as stored on the trigger.

passed: bool#

Whether this leaf held for the event.

class roboto.domain.triggers.ConditionMatcher(namespace, default_root)#

Evaluates a trigger condition tree against an event namespace.

Reuses the query Condition comparators unchanged; only the value source differs. Each leaf routes to the namespace root its field targets and matches against that root’s record, so roots hydrate lazily on first reference. A root whose record is missing evaluates to a non-match, so NOT_EXISTS and IS_NULL never match against an entity that was deleted before evaluation.

Parameters:
matches(condition)#

Return whether condition holds for the event. A None condition always matches.

Parameters:

condition (Optional[roboto.query.ConditionType])

Return type:

bool

class roboto.domain.triggers.CreateTriggerRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to create a trigger.

The server assigns identity and audit fields, and runs TriggerValidator over the cross-field rules (exposed roots, once_per legality, target placeholders) before persisting.

Triggered work runs as the organization’s actions service user; a trigger cannot run its targets as anyone else.

Parameters:

data (Any)

condition: roboto.query.ConditionType | None = None#

Optional predicate over the firing’s namespace.

enabled: bool = True#

Whether the trigger should be active immediately after creation.

fires_on: roboto.domain.triggers.sources.TriggerSource#

an event subscription or a schedule.

Type:

What makes the trigger fire

name: str = None#

Trigger name. Unique within the caller’s organization.

targets: list[roboto.domain.triggers.targets.TriggerTargetSpec]#

What the trigger dispatches when it fires. At least one.

class roboto.domain.triggers.DispatchSlot#

One target of one trigger at one dedup token: the unit a dispatch claims.

At most one dispatch ever occupies a slot. It is the dispatch table’s primary key and the vocabulary every dispatch port speaks.

idempotency_token: str#
target_id: str#
trigger_id: str#
class roboto.domain.triggers.DispatchSlotTrace(/, **data)#

Bases: pydantic.BaseModel

The state of one target’s dispatch slot at the dry run’s idempotency token.

Parameters:

data (Any)

occupied: bool#

Whether a dispatch row occupies the slot (the trigger already fired here).

result_ref: str | None = None#

What the occupying dispatch produced (invocation id, thread id, Slack ts).

status: roboto.domain.triggers.dispatch.TriggerDispatchStatus | None = None#

The occupying dispatch’s status, when one exists.

target_id: str#

The target within the trigger.

class roboto.domain.triggers.EventNamespace(event, source, *, trigger=None, _records=None)#

Resolves dotted variable paths against a platform event.

The one namespace with three consumers: condition evaluation (whole records via record()), target template substitution (single paths via resolve(), satisfying the VariableResolver protocol), and idempotency projections. Entity-backed roots hydrate through the NamespaceSource at most once each and are cached for the namespace’s lifetime — including None results. The envelope, changed, and tag roots are served from the event itself and never touch the source; trigger is served from the bound trigger (see bound_to()) and is otherwise absent.

Parameters:
bound_to(trigger)#

A view of this namespace for one trigger, in which trigger.* resolves.

One namespace is shared across every trigger evaluated for an event so each entity hydrates once; the trigger being evaluated is per-trigger state, so it lives on a view rather than on the shared object. The view delegates every other root to the same hydration cache – reading dataset.name through it and through the parent costs one fetch in total.

Parameters:

trigger (roboto.domain.triggers.record.TriggerRecord) – The trigger whose templates and condition are being evaluated.

Return type:

EventNamespace

get(path)#

Return the value at dotted path (e.g. dataset.metadata.vehicle_id), or None.

The first path segment names the root; the rest walk nested mappings. Any missing segment yields None.

Parameters:

path (str)

Return type:

Any

record(root)#

Return the whole record for namespace root, hydrating it at most once.

Parameters:

root (str) – Namespace root to fetch. envelope yields the envelope fields; changed yields the changeset’s put fields; tag yields {"added": [...], "removed": [...]}; any other root delegates to the cached NamespaceSource.

Returns:

The record as a mapping, or None when the source cannot resolve the root.

Return type:

Optional[Mapping[str, Any]]

resolve(name)#

Return the string form of the value at name for template substitution, or None.

Parameters:

name (str)

Return type:

Optional[str]

class roboto.domain.triggers.EventSubscription(/, **data)#

Bases: _SourceBase

Fire when any of the subscribed platform events occurs.

Carries once_per because it only means something here: it names what the event is about that the trigger fires at most once for, and a schedule has no such subject.

Parameters:

data (Any)

events: list[roboto.domain.platform_events.PlatformEventType]#

Event types the trigger subscribes to. At least one; a condition may only reference namespace roots exposed by every subscribed event.

fires_for(event_type)#

Return whether an event of event_type is one this subscription fires for.

Parameters:

event_type (roboto.domain.platform_events.PlatformEventType)

Return type:

bool

once_per: roboto.domain.platform_events.OncePer#

the occurrence, or an entity the event names. Must be legal for every subscribed event type.

Type:

What the trigger fires at most once per

type: Literal[TriggerSourceType]#

Discriminator for TriggerSource.

class roboto.domain.triggers.InvokeActionTarget(/, **data)#

Bases: _TargetSpecBase

Stored configuration for invoking an action when a trigger fires.

Spec only — dispatch behavior lives server-side. String leaves of parameter_values, required_inputs, and additional_inputs may contain {{...}} placeholders resolved against the event namespace at dispatch time.

Parameters:

data (Any)

action: roboto.domain.actions.action_record.ActionReference#

The action to invoke. owner defaults to the trigger’s org when omitted and digest to the action’s latest version.

additional_inputs: list[str] | None = None#

Optional extra file patterns passed as invocation inputs beyond the required ones.

compute_requirement_overrides: roboto.domain.actions.action_record.ComputeRequirements | None = None#

Optional compute requirement overrides for the invocation.

container_parameter_overrides: roboto.domain.actions.action_record.ContainerParameters | None = None#

Optional container parameter overrides for the invocation.

invocation_input: roboto.domain.actions.invocation_record.InvocationInput | None = None#

Optional query-based input selection (files, topics, sessions) resolved when the invocation runs. The way a schedule-fired trigger names its inputs, since a schedule fires with no file subject; may also accompany an event-fired trigger’s patterns.

model_config#

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

parameter_values: dict[str, Any] = None#

Parameter values passed to the action; string leaves may be templated.

referenced_placeholders()#

Return every {{name}} placeholder referenced by this target’s templated fields.

Return type:

set[str]

required_inputs: list[str] = None#

File patterns (e.g. **/*.bag) that gate dispatch on the firing event’s files, for file- and dataset-scoped events. May be empty for events with no file subject (e.g. invocation events).

subject_types: ClassVar[frozenset[roboto.uri.RobotoUriType] | None]#

The action runs against the event’s dataset, and a file event names its dataset too.

target_id: str#

Stable id of this target within its trigger. Part of the dispatch idempotency key: the target’s once-per history is kept under this id, so renaming it (or replacing it with an identical target under a new id) starts a fresh history and the target fires again for subjects the old id already fired for.

timeout: int | None = None#

Optional invocation timeout override, in minutes.

type: Literal[TriggerTargetType]#

Discriminator for TriggerTargetSpec.

upload_destination: roboto.domain.actions.invocation_record.InvocationUploadDestination | None = None#

Optional destination for the invocation’s output files.

roboto.domain.triggers.MAX_TRIGGER_NAME_LENGTH = 256#

Maximum trigger name length.

class roboto.domain.triggers.NamespaceSource#

Bases: Protocol

Lazily hydrates entity-backed namespace roots for an event.

record_for_root(root, event)#

Load the entity for namespace root as a JSON-able mapping.

Parameters:

  • root (str) – The namespace root to hydrate ("dataset", "file", "invocation", "event", …). Never a delta or reserved root: the namespace serves envelope, trigger, schedule, changed and tag itself.

  • event (roboto.domain.platform_events.PlatformEvent) – The event whose subject identifies the entity to load.

Returns:

The entity as a mapping, or None when event cannot resolve it (no such root for this event type, or the entity was deleted between emit and evaluation).

Return type:

Optional[Mapping[str, Any]]

class roboto.domain.triggers.PlatformEventSample(/, **data)#

Bases: pydantic.BaseModel

One event type, dereferenced: the envelope, every root it exposes, and the template paths that resolve against them.

Parameters:

data (Any)

event: dict[str, Any]#

id, type, time, org, and payload.

Type:

The event envelope as it would arrive

event_type: roboto.domain.platform_events.PlatformEventType#

The event type this sample illustrates.

namespace: dict[str, dict[str, Any]]#

Every namespace root a condition or template may reference for this event type, keyed by root name, with the full record under each. Always includes envelope and trigger; the rest are the catalog’s exposed roots for the type.

paths: list[str]#

Every root.path a {{ }} placeholder or condition field may name, in the order the roots and their fields appear in namespace.

class roboto.domain.triggers.PlatformEventSamplesResponse(/, **data)#

Bases: pydantic.BaseModel

Wire shape of GET /v1/triggers/events/samples: one sample per platform event type.

Parameters:

data (Any)

samples: dict[roboto.domain.platform_events.PlatformEventType, PlatformEventSample]#
class roboto.domain.triggers.SampleNamespaceSource#

A NamespaceSource that hydrates every root from the sample scenario.

Serves each root the way a real firing does — including action as the invocation’s provenance and upload as the bare transaction id — so a sample carries exactly the roots a trigger would see.

record_for_root(root, event)#
Parameters:
Return type:

Optional[collections.abc.Mapping[str, Any]]

class roboto.domain.triggers.Schedule(/, **data)#

Bases: _SourceBase

Fire on a cron schedule, in UTC.

Each scheduled minute fires the trigger once (once_per='occurrence' over the schedule tick). A missed minute is skipped rather than replayed later, so a delayed schedule never floods targets with backdated firings. Cron expressions are evaluated in UTC; a schedule cannot name a time zone.

Parameters:

data (Any)

cron: str#

A five-field cron expression (minute hour day-of-month month day-of-week), evaluated in UTC — e.g. 0 9 * * 1 for 09:00 UTC every Monday.

fires_for(event_type)#

Return whether event_type is the schedule occurrence this source fires for.

Parameters:

event_type (roboto.domain.platform_events.PlatformEventType)

Return type:

bool

property once_per: roboto.domain.platform_events.OncePer#

A schedule always fires once per scheduled minute.

Return type:

roboto.domain.platform_events.OncePer

type: Literal[TriggerSourceType]#

Discriminator for TriggerSource.

class roboto.domain.triggers.SendSlackMessageTarget(/, **data)#

Bases: _TargetSpecBase

Stored configuration for posting a Slack message when a trigger fires.

Spec only — dispatch behavior lives server-side. The channel must be on the org’s Slack outbound allowlist at dispatch time.

Parameters:

data (Any)

channel_id: str#

Slack channel to post to.

model_config#

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

referenced_placeholders()#

Return every {{name}} placeholder referenced by this target’s templated text.

Return type:

set[str]

target_id: str#

Stable id of this target within its trigger; part of the dispatch idempotency key.

text: str#

Message body; may contain {{...}} placeholders resolved against the event namespace.

type: Literal[TriggerTargetType]#

Discriminator for TriggerTargetSpec.

class roboto.domain.triggers.StartAgentTarget(/, **data)#

Bases: _TargetSpecBase

Stored configuration for starting an agent thread when a trigger fires.

Spec only — dispatch behavior lives server-side. Each value in values is a template resolved against the event namespace, then handed to the agent’s own variable resolution as a plain value.

Parameters:

data (Any)

agent_id: str#

The agent definition to launch.

analysis_scope: roboto.ai.core.AnalysisScope | None = None#

Optional analysis scope for the resulting thread.

model_config#

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

referenced_placeholders()#

Return every {{name}} placeholder referenced by this target’s templated values.

Return type:

set[str]

target_id: str#

Stable id of this target within its trigger; part of the dispatch idempotency key.

type: Literal[TriggerTargetType]#

Discriminator for TriggerTargetSpec.

values: dict[str, str] = None#

Agent variable name to a template string resolved against the event namespace.

visibility: roboto.ai.agent_thread.record.ThreadVisibility#

Visibility of the resulting agent thread.

roboto.domain.triggers.TRIGGER_NAME_PATTERN = '[\\w\\-]+'#

word characters and hyphens.

Type:

Legal trigger names

class roboto.domain.triggers.TargetAcceptanceTrace(/, **data)#

Bases: pydantic.BaseModel

One target’s prefilter decision.

Parameters:

data (Any)

accepted: bool#

Whether the target’s accepts prefilter passed.

reason: str | None = None#

Why the target declined, where determinable (e.g. which required-input pattern had no matching file). None when accepted or when no finer reason is known.

target_id: str#

The target within the trigger.

target_type: roboto.domain.triggers.targets.TriggerTargetType#

Kind of target.

class roboto.domain.triggers.Trigger(record, roboto_client=None)#

A trigger: a firing source (event subscriptions or a schedule), an optional condition, and targets.

Triggers created through the older causes/for_each API (roboto.domain.actions.Trigger) are listed and loaded here too, projected onto this shape and marked engine == "v1"; they accept a narrower set of edits (see update()).

Examples

Start an agent whenever a dataset gains a ready tag:

>>> from roboto.domain.platform_events import OncePer, PlatformEventType
>>> from roboto.domain.triggers import StartAgentTarget, Trigger
>>> trigger = Trigger.create(
...     name="analyze-ready-datasets",
...     targets=[StartAgentTarget(target_id="analyze", agent_id="ag_abc123")],
...     events=[PlatformEventType.DatasetTagAdded],
...     once_per=OncePer.Occurrence,
... )

Post to Slack every Monday at 09:00 UTC:

>>> from roboto.domain.triggers import SendSlackMessageTarget
>>> weekly = Trigger.create(
...     name="weekly-status",
...     targets=[SendSlackMessageTarget(target_id="post", channel_id="C0123", text="Weekly check-in")],
...     schedule="0 9 * * 1",
... )
Parameters:
property condition: roboto.query.ConditionType | None#
Return type:

Optional[roboto.query.ConditionType]

classmethod create(name, targets, *, events=None, once_per=None, schedule=None, fires_on=None, condition=None, enabled=True, caller_org_id=None, roboto_client=None)#

Create a trigger in the caller’s org.

Parameters:

  • name (str) – Trigger name, unique within the org.

  • targets (list[roboto.domain.triggers.targets.TriggerTargetSpec]) – What to dispatch on a match. At least one.

  • events (Optional[list[roboto.domain.platform_events.PlatformEventType]]) – Platform event types to subscribe to (with once_per). At least one.

  • once_per (Optional[roboto.domain.platform_events.OncePer]) – What the trigger fires at most once per; must be legal for every subscribed event.

  • schedule (Optional[str]) – A cron expression (UTC) to fire on instead of events.

  • fires_on (Optional[roboto.domain.triggers.sources.TriggerSource]) – The firing source itself, as an alternative to the events/once_per or schedule shorthands.

  • condition (Optional[roboto.query.ConditionType]) – Optional predicate over the firing’s namespace.

  • enabled (bool) – Whether the trigger is active immediately.

  • caller_org_id (Optional[str]) – Org to create the trigger in. Defaults to the caller’s org.

  • roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client instance. Uses the default if not provided.

Returns:

The created trigger.

Raises:
Return type:

Trigger

delete()#

Delete this trigger. Idempotent.

Return type:

None

disable()#

Disable this trigger.

Return type:

Trigger

dispatches(limit=100)#

Yield this trigger’s dispatch history, newest first.

A dispatch is one attempt to run one target for one matched event; only matches are recorded, so an empty history means the trigger never fired.

Parameters:

limit (int) – Page size for the underlying requests.

Return type:

collections.abc.Generator[roboto.domain.triggers.dispatch.TriggerDispatchRecord, None, None]

dry_run(event=None, dataset_id=None, file_id=None, invocation_id=None, session_id=None, event_id=None, event_type=None, scheduled_for=None)#

Ask “would this trigger fire?” without dispatching anything.

Provide either a full event or exactly one entity reference, from which the server synthesizes an event (event_type optionally picks which kind). A schedule-fired trigger takes no reference: pass scheduled_for to pick the minute, or nothing for the schedule’s next occurrence. The response is the evaluator’s gate-by-gate trace: subscribed, enabled, condition (with per-leaf actual values), target prefilter, already fired.

Examples

>>> trace = trigger.dry_run(dataset_id="ds_abc123")
>>> print(trace.verdict)
Parameters:
Return type:

roboto.domain.triggers.dry_run.TriggerDryRunResponse

enable()#

Enable this trigger.

Return type:

Trigger

property enabled: bool#
Return type:

bool

property engine: str#

"v2", or "v1" for a trigger created through the older causes/for_each API.

Type:

Which trigger shape this trigger was created in

Return type:

str

property events: List[roboto.domain.platform_events.PlatformEventType] | None#

The subscribed event types, or None for a schedule-fired trigger.

Return type:

Optional[List[roboto.domain.platform_events.PlatformEventType]]

property fires_on: roboto.domain.triggers.sources.TriggerSource#

an event subscription or a schedule.

Type:

What makes the trigger fire

Return type:

roboto.domain.triggers.sources.TriggerSource

classmethod from_id(trigger_id, roboto_client=None)#

Load the trigger with the given id, whichever API created it.

Parameters:
Return type:

Trigger

classmethod from_name(name, owner_org_id=None, roboto_client=None)#

Load the trigger with the given name, whichever API created it. Names are unique within an org.

Parameters:
Return type:

Trigger

classmethod list(owner_org_id=None, roboto_client=None)#

Yield every trigger in the org, event-fired and scheduled, newest first, whichever API created it.

Parameters:
Return type:

collections.abc.Generator[Trigger, None, None]

property name: str#
Return type:

str

property once_per: roboto.domain.platform_events.OncePer#

What the trigger fires at most once per; always occurrence (one firing per minute) for a schedule.

Return type:

roboto.domain.platform_events.OncePer

property org_id: str#
Return type:

str

classmethod platform_event_samples(roboto_client=None)#

Fetch a realistic sample of every event type, dereferenced.

Each sample carries the event envelope, every namespace root the type exposes with a full record under it, and the root.path list a condition field or {{ }} placeholder may name. Use it to see what a template will resolve to before writing one.

Examples

>>> samples = Trigger.platform_event_samples()
>>> samples[PlatformEventType.FileUploaded].paths[:3]
['envelope.id', 'envelope.type', 'envelope.time']
Parameters:

roboto_client (Optional[roboto.http.RobotoClient])

Return type:

dict[roboto.domain.platform_events.PlatformEventType, roboto.domain.triggers.samples.PlatformEventSample]

property record: roboto.domain.triggers.record.TriggerRecord#
Return type:

roboto.domain.triggers.record.TriggerRecord

property schedule: str | None#

The cron expression (UTC) the trigger fires on, or None for an event-fired trigger.

Return type:

Optional[str]

set_enabled(enabled)#

Enable or disable this trigger.

Parameters:

enabled (bool)

Return type:

Trigger

property targets: List[roboto.domain.triggers.targets.TriggerTargetSpec]#
Return type:

List[roboto.domain.triggers.targets.TriggerTargetSpec]

to_dict()#

Return this trigger’s record as a JSON-able dict.

Return type:

dict[str, Any]

property trigger_id: str#
Return type:

str

update(*, fires_on=NotSet, events=NotSet, once_per=NotSet, schedule=NotSet, condition=NotSet, targets=NotSet, enabled=NotSet)#

Apply a partial update to this trigger and refresh this instance.

Only provided fields change; condition=None clears the condition. The firing source is replaced whole: pass fires_on, or the events / once_per / schedule shorthands, which are merged over the current source before being sent. A trigger with engine == "v1" accepts only edits its older shape can express — a single invoke-action target, event types that map onto it, once_per of file or dataset — and rejects the rest with a message naming what it cannot store.

Parameters:
Return type:

Trigger

class roboto.domain.triggers.TriggerDispatchRecord(/, **data)#

Bases: pydantic.BaseModel

Wire-transmissible representation of one trigger dispatch.

A dispatch is one attempt to run one target of one trigger for one matched platform event, deduped at the trigger’s OncePer. The triple (trigger_id, idempotency_token, target_id) is the DispatchSlot; at most one dispatch ever occupies a slot, which is what makes redelivered events safe.

Parameters:

data (Any)

claimed_at: datetime.datetime#

When the slot was (most recently) claimed.

dataset_id: str | None = None#

The dataset the subject belongs to (the subject itself for a dataset event), which is what a dataset’s page lists dispatches by. None when the subject has no dataset, such as a session or an invocation.

event_id: str#

Id of the platform event occurrence that most recently claimed this slot.

event_type: roboto.domain.platform_events.PlatformEventType#

Type of the platform event that matched.

finalized_at: datetime.datetime | None = None#

When the dispatch reached a terminal status; None while claimed.

idempotency_token: str#

Dedup token derived from the matched event at the trigger’s once_per ({event.type}|{once_per}:{projection}).

org_id: str#

Organization that owns the trigger.

result_ref: str | None = None#

an invocation id, an agent thread id, or a Slack message timestamp, per target_type.

Type:

What a dispatched target produced

property slot: DispatchSlot#

The slot this dispatch occupies.

Return type:

DispatchSlot

status: TriggerDispatchStatus#

Where this dispatch is in its lifecycle.

status_detail: str | None = None#

Human-readable detail for status, e.g. the error a failed target raised.

subject: str#

The entity the matched platform event was about, as a roboto:// URI; the same value as the event’s own subject.

property subject_uri: roboto.uri.RobotoUri#

subject as a parsed RobotoUri.

Return type:

roboto.uri.RobotoUri

target_id: str#

Which of the trigger’s targets this dispatch ran.

target_type: roboto.domain.triggers.targets.TriggerTargetType#

Kind of target dispatched.

trigger_id: str#

Trigger this dispatch belongs to.

class roboto.domain.triggers.TriggerDispatchStatus#

Bases: roboto.compat.StrEnum

Lifecycle state of one dispatch: one attempt to run one target for one matched event.

Claimed = 'claimed'#

The dispatch slot is claimed and the target is about to run. A claim that neither finalizes nor is reclaimed within the redelivery grace period is presumed dead and may be claimed again, so delivery is at-least-once: a target may run twice for one event.

Dispatched = 'dispatched'#

The target ran; TriggerDispatchRecord.result_ref points at what it produced.

Failed = 'failed'#

The target raised; a redelivered event may claim the slot again.

Unknown = 'unknown'#

The claim outlived every redelivery of its event without finalizing, so the outcome cannot be determined. Set by an operational sweep, never reclaimable.

class roboto.domain.triggers.TriggerDryRunGate(/, **data)#

Bases: pydantic.BaseModel

One gate’s result in a dry-run trace.

condition_leaves is populated only on the condition gate, targets only on target_prefilter, and idempotency_token/dispatches only on already_fired.

Parameters:

data (Any)

condition_leaves: list[ConditionLeafTrace] | None = None#

Per-leaf results with actual values (condition gate only).

detail: str | None = None#

Plain-English explanation of the outcome.

dispatches: list[DispatchSlotTrace] | None = None#

Per-target dispatch-slot state at the token (already_fired gate only).

gate: TriggerDryRunGateName#

Which gate this is.

idempotency_token: str | None = None#

The dedup token the event projects onto (already_fired gate only).

status: TriggerDryRunGateStatus#

Whether the gate passed, failed, or was short-circuited.

targets: list[TargetAcceptanceTrace] | None = None#

Per-target prefilter decisions (target_prefilter gate only).

class roboto.domain.triggers.TriggerDryRunGateName#

Bases: roboto.compat.StrEnum

The gates the evaluator runs, in evaluation order.

AlreadyFired = 'already_fired'#

Is a dispatch slot still claimable at the trigger’s once_per?

Condition = 'condition'#

Does the trigger’s condition hold for the event?

Enabled = 'enabled'#

Is the trigger enabled?

Subscribed = 'subscribed'#

Is the trigger subscribed to the event’s type?

TargetPrefilter = 'target_prefilter'#

Does at least one target accept the event (pathspec and precondition gates)?

class roboto.domain.triggers.TriggerDryRunGateStatus#

Bases: roboto.compat.StrEnum

Outcome of one gate in a dry run.

Failed = 'failed'#
NotEvaluated = 'not_evaluated'#

An earlier gate failed, so this one was short-circuited.

Passed = 'passed'#
class roboto.domain.triggers.TriggerDryRunRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for a trigger dry run.

Provide event (a fully-formed platform event to evaluate) or a single reference (dataset_id, file_id, invocation_id, session_id, event_id, or scheduled_for), from which the server synthesizes an event. With an entity reference, event_type optionally picks which of the trigger’s subscribed event types to synthesize; the default is the trigger’s first subscribed event type compatible with the reference. A schedule-fired trigger needs no reference: the server synthesizes its next scheduled minute.

Parameters:

data (Any)

dataset_id: str | None = None#

Synthesize an event about this dataset.

event: roboto.domain.platform_events.PlatformEvent | None = None#

A platform event to evaluate as-is.

event_id: str | None = None#

Synthesize a platform event about this event (the annotation on your data).

event_type: roboto.domain.platform_events.PlatformEventType | None = None#

Which subscribed event type to synthesize for an entity reference.

file_id: str | None = None#

Synthesize an event about this file.

invocation_id: str | None = None#

Synthesize an event about this invocation.

scheduled_for: datetime.datetime | None = None#

synthesize the occurrence for this scheduled minute (UTC). With no reference at all, a schedule-fired trigger is dry-run for its next scheduled minute.

Type:

For a schedule-fired trigger

session_id: str | None = None#

Synthesize an event about this session.

class roboto.domain.triggers.TriggerDryRunResponse(/, **data)#

Bases: pydantic.BaseModel

The structured trace a trigger dry run produces.

Gates appear in evaluation order. The first failed gate is why the trigger would not fire; every gate after it is not_evaluated.

Parameters:

data (Any)

event_type: roboto.domain.platform_events.PlatformEventType#

Type of the (given or synthesized) event that was evaluated.

gates: list[TriggerDryRunGate]#

The gate-by-gate trace, in evaluation order.

trigger_id: str#

The trigger that was dry-run.

verdict: str#

One plain-English sentence summarizing the outcome.

would_fire: bool#

Whether the trigger would dispatch at least one target for this event.

class roboto.domain.triggers.TriggerRecord(/, **data)#

Bases: pydantic.BaseModel

Wire-transmissible representation of a trigger.

A firing source (a platform event subscription or a schedule), an optional condition over the firing’s namespace, and the targets to dispatch on a match. Cross-field rules — exposed roots, once_per legality, template placeholders — are enforced when the trigger is saved, by TriggerValidator.

Triggered work runs as the organization’s actions service user; a trigger cannot run its targets as anyone else.

Parameters:

data (Any)

condition: roboto.query.ConditionType | None = None#

Optional predicate over the firing’s namespace; the trigger fires only when it holds. Same Condition wire format the query system uses.

created: datetime.datetime#

Timestamp when the trigger was created.

created_by: str#

User ID who created the trigger.

enabled: bool = True#

Whether the trigger is active.

engine: str = 'v1'#

"v2" for the events/once_per/targets engine this module models (every trigger, once a deployment has cut over), "v1" for a trigger the older causes/for_each engine still evaluates, projected onto this shape. Server-assigned and read-only; no request model carries it. Defaults to "v1" for a response that omits it, which is what a Roboto deployment older than the field sends.

Type:

Which engine evaluates this trigger

fires_on: roboto.domain.triggers.sources.TriggerSource#

an EventSubscription (which also carries once_per) or a Schedule.

Type:

What makes the trigger fire

modified: datetime.datetime#

Timestamp when the trigger was last modified.

modified_by: str#

User ID who last modified the trigger.

name: str#

Human-readable name. Unique within org_id.

org_id: str#

Organization that owns the trigger and whose events it sees.

targets: list[roboto.domain.triggers.targets.TriggerTargetSpec]#

What the trigger dispatches when it fires, in order. At least one; each target_id is unique within the trigger.

trigger_id: str#

Unique identifier for the trigger.

roboto.domain.triggers.TriggerSource#

A trigger’s firing source, discriminated on type.

class roboto.domain.triggers.TriggerSourceType#

Bases: roboto.compat.StrEnum

Discriminator for TriggerSource.

Events = 'events'#

The trigger fires when a subscribed platform event occurs.

Schedule = 'schedule'#

The trigger fires on a cron schedule.

roboto.domain.triggers.TriggerTargetSpec#

A trigger target spec, discriminated on type.

class roboto.domain.triggers.TriggerTargetType#

Bases: roboto.compat.StrEnum

The kind of thing a trigger does when it fires.

InvokeAction = 'invoke_action'#

Invoke a containerized action.

SendSlackMessage = 'send_slack_message'#

Post a message to a Slack channel on the org’s allowlist.

StartAgent = 'start_agent'#

Start an AI agent thread from a saved agent definition.

class roboto.domain.triggers.TriggerValidator(catalog=DEFAULT_PLATFORM_EVENT_CATALOG)#

Save-time validation for triggers.

Enforces the cross-field rules that need the event catalog: a condition may only reference namespace roots exposed by every event the source fires for, once_per must be legal for every subscribed event, every target must act on the kind of entity the subscription is about, and every target template placeholder must resolve to a shared exposed root (or a reserved envelope./trigger. context). A schedule fires for exactly one occurrence type, so the same rules apply to it with that occurrence’s roots.

Run before persistence so a trigger that could never fire — or could never resolve its templates — fails loudly with an actionable message instead of silently misbehaving at evaluation time.

Parameters:

catalog (roboto.domain.platform_events.PlatformEventCatalog)

validate(*, fires_on, condition, targets)#

Validate the cross-field rules for one trigger’s parts.

Parameters:

  • fires_on (roboto.domain.triggers.sources.TriggerSource) – The trigger’s firing source.

  • condition (Optional[roboto.query.ConditionType]) – Optional predicate over the firing’s namespace.

  • targets (collections.abc.Sequence[roboto.domain.triggers.targets.TriggerTargetSpec]) – The trigger’s target specs.

Raises:

ValueError – The first rule that fails, with a message naming the offending field, root, or event.

Return type:

None

validate_record(record)#

Validate a full trigger record.

Raises:

ValueError – Any rule in validate() fails.

Parameters:

record (roboto.domain.triggers.record.TriggerRecord)

Return type:

None

class roboto.domain.triggers.UpdateTriggerRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to update a trigger.

Only fields explicitly provided are changed; the NotSetType sentinel distinguishes “field omitted” from “field set to None”. The updated record must still satisfy every TriggerValidator rule.

Parameters:

data (Any)

condition: roboto.query.ConditionType | roboto.sentinels.NotSetType | None#

New condition; explicit None clears it.

enabled: bool | roboto.sentinels.NotSetType#

New enabled status.

fires_on: roboto.domain.triggers.sources.TriggerSource | roboto.sentinels.NotSetType#

an event subscription’s events and once_per change together, and a trigger may switch between events and a schedule.

Type:

New firing source. Replaces the whole source

model_config#

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

targets: list[roboto.domain.triggers.targets.TriggerTargetSpec] | roboto.sentinels.NotSetType#

New target list. A target whose target_id is referenced by existing dispatches must keep that id.

roboto.domain.triggers.platform_event_sample(event_type, catalog=DEFAULT_PLATFORM_EVENT_CATALOG)#

Build the sample for one event type.

The event, changed and tag roots are built by the same EventNamespace that serves them when a trigger fires; the entity roots are exactly the ones catalog exposes for the type.

Parameters:
Return type:

PlatformEventSample

roboto.domain.triggers.platform_event_samples(catalog=DEFAULT_PLATFORM_EVENT_CATALOG)#

A sample for every event type in catalog, in catalog order.

Parameters:

catalog (roboto.domain.platform_events.PlatformEventCatalog)

Return type:

dict[roboto.domain.platform_events.PlatformEventType, PlatformEventSample]

roboto.domain.triggers.sample_schedule_trigger()#

The schedule-fired trigger the schedule.fired sample is evaluated for, so the sample’s schedule root carries the cron a template may name.

Return type:

roboto.domain.triggers.record.TriggerRecord

roboto.domain.triggers.sample_trigger()#

The trigger the sample is evaluated for; what {{trigger.*}} sees.

Return type:

roboto.domain.triggers.record.TriggerRecord

roboto.domain.triggers.target_catalog_manifest()#

Which kinds of entity each target can act on, as the web UI’s target picker reads it.

target_catalog.json in this package is this function’s output, written by scripts/gen_trigger_manifests.py and drift-checked by a test on each side.

Return type:

dict[str, Any]

roboto.domain.triggers.template_paths(record, prefix)#

Every dotted path a template may reference under prefix, in document order.

Walks nested mappings only. A list is a leaf, because path resolution descends through mappings and stops at anything else: dataset.tags is addressable, dataset.tags.0 is not.

Parameters:

  • record (collections.abc.Mapping[str, Any])

  • prefix (str)

Return type:

list[str]

On this page