roboto.domain.triggers.dry_run

Wire models for the trigger dry-run endpoint (POST /v1/triggers/id/{id}/dry-run).

A dry run answers “would this trigger fire for this event, and if not, why not” by replaying the evaluator’s gates in order — subscribed, enabled, condition, target prefilter, already fired — against current platform state, without claiming a dispatch slot or running any target. The response is the structured trace the web UI’s “Test this trigger” stepper and the CLI’s dry-run command render.

Module Contents

class roboto.domain.triggers.dry_run.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.dry_run.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.dry_run.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.dry_run.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.dry_run.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.dry_run.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.dry_run.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.dry_run.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.