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, tag and schedule 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.

envelope carries the event’s own fields plus the payload under envelope.data. The payload is fixed when the event is published, while an entity root reads that entity as it stands at evaluation time: on a file.uploaded event, envelope.data.file_version is the version that fired the trigger and file.version is the version the file is on when the condition runs. That payload field is optional on file.uploaded, file.ingested and file.metadata_updated, and resolves to None on events published before the payload carried it; only session.file_added always carries it.

Payload values are rendered as JSON, so a timestamp under envelope.data is an ISO-8601 string while envelope.time and schedule.scheduled_for are datetime objects. Conditions compare the two forms alike; a template substitutes each in its own spelling (2026-08-27T09:00:00Z against 2026-08-27 09:00:00+00:00).

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 event’s own fields, with the payload fixed at publish time under data; changed yields the changeset’s put fields; tag yields {"added": [...], "removed": [...]}; schedule and trigger yield the firing schedule and the bound trigger; 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, additional_inputs, and invocation_input may contain {{...}} placeholders resolved against the event namespace at dispatch time.

An event that names a dataset invokes the action against that dataset, with inputs matched from its files by the file patterns. An event that names no dataset has no files to match: the invocation has no data source and selects its inputs with invocation_input.

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. Only for events that name a dataset; must be empty otherwise.

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, e.g. InvocationInput.from_session_id("{{session.session_id}}"). The way a trigger on an event that names no dataset selects its inputs; may also accompany the file patterns of an event that names one.

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. Only for events that name a dataset; must be empty otherwise.

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, an action target’s file patterns need a dataset to match against and are themselves required when the firing event’s own file is the action’s input, 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.iter_query_templates(invocation_input)#

Yield every RoboQL query in invocation_input, in document order.

Parameters:

invocation_input (collections.abc.Mapping[str, Any])

Return type:

collections.abc.Iterator[str]

roboto.domain.triggers.map_query_templates(invocation_input, transform)#

Return invocation_input with transform applied to every RoboQL query it holds.

Parameters:

  • invocation_input (collections.abc.Mapping[str, Any]) – An InvocationInput in its JSON form, whose top-level values are selectors or lists of them.

  • transform (collections.abc.Callable[[str], str]) – Called with each selector’s query; its result replaces that query.

Returns:

A copy, shallow below the selectors it rewrites.

Return type:

dict[str, Any]

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.substitute_query_template(query, resolve)#

Return query with each placeholder replaced by resolve’s value for it, escaped in place.

Each placeholder is substituted exactly once, and a value is escaped for the literal that encloses it, so a value that itself looks like a template or carries a quote stays data.

Parameters:

  • query (str) – A RoboQL query carrying {{placeholder}} templates.

  • resolve (collections.abc.Callable[[str], str]) – Returns the value for a placeholder name.

Raises:

ValueError – A placeholder sits outside a string literal, or a value cannot be escaped into the literal it lands in.

Return type:

str

roboto.domain.triggers.target_catalog_manifest()#

The kinds of target a trigger can have, as the web UI’s target picker reads them.

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. Each entry’s body is empty: a target carries no save-time constraint of its own, so the picker offers every type for every event.

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