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

ConditionLeafTrace

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

Attributes

ConditionLeafTrace.actual

actual Any | None = None #

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

ConditionLeafTrace.comparator

comparator roboto.query.Comparator #

The leaf’s comparator.

ConditionLeafTrace.expected

expected Any | None = None #

The value the condition compares against.

ConditionLeafTrace.field

field str #

The condition field, as stored on the trigger.

ConditionLeafTrace.passed

passed bool #

Whether this leaf held for the event.

DispatchSlotTrace

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

Attributes

DispatchSlotTrace.occupied

occupied bool #

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

DispatchSlotTrace.result_ref

result_ref str | None = None #

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

DispatchSlotTrace.status

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

The occupying dispatch’s status, when one exists.

DispatchSlotTrace.target_id

target_id str #

The target within the trigger.

TargetAcceptanceTrace

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

Bases: pydantic.BaseModel

One target’s prefilter decision.

Parameters

data Any

Attributes

TargetAcceptanceTrace.accepted

accepted bool #

Whether the target’s accepts prefilter passed.

TargetAcceptanceTrace.reason

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.

TargetAcceptanceTrace.target_id

target_id str #

The target within the trigger.

TargetAcceptanceTrace.target_type

target_type roboto.domain.triggers.targets.TriggerTargetType #

Kind of target.

TriggerDryRunGate

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

Attributes

TriggerDryRunGate.condition_leaves

condition_leaves list[ConditionLeafTrace] | None = None #

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

TriggerDryRunGate.detail

detail str | None = None #

Plain-English explanation of the outcome.

TriggerDryRunGate.dispatches

dispatches list[DispatchSlotTrace] | None = None #

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

TriggerDryRunGate.gate

gate TriggerDryRunGateName #

Which gate this is.

TriggerDryRunGate.idempotency_token

idempotency_token str | None = None #

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

TriggerDryRunGate.status

status TriggerDryRunGateStatus #

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

TriggerDryRunGate.targets

targets list[TargetAcceptanceTrace] | None = None #

Per-target prefilter decisions (target_prefilter gate only).

TriggerDryRunGateName

class roboto.domain.triggers.dry_run.TriggerDryRunGateName#

Bases: roboto.compat.StrEnum

The gates the evaluator runs, in evaluation order.

Attributes

TriggerDryRunGateName.AlreadyFired

AlreadyFired = 'already_fired' #

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

TriggerDryRunGateName.Condition

Condition = 'condition' #

Does the trigger’s condition hold for the event?

TriggerDryRunGateName.Enabled

Enabled = 'enabled' #

Is the trigger enabled?

TriggerDryRunGateName.Subscribed

Subscribed = 'subscribed' #

Is the trigger subscribed to the event’s type?

TriggerDryRunGateName.TargetPrefilter

TargetPrefilter = 'target_prefilter' #

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

TriggerDryRunGateStatus

class roboto.domain.triggers.dry_run.TriggerDryRunGateStatus#

Bases: roboto.compat.StrEnum

Outcome of one gate in a dry run.

Attributes

TriggerDryRunGateStatus.Failed

Failed = 'failed' #

TriggerDryRunGateStatus.NotEvaluated

NotEvaluated = 'not_evaluated' #

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

TriggerDryRunGateStatus.Passed

Passed = 'passed' #

TriggerDryRunRequest

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, 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

Attributes

TriggerDryRunRequest.dataset_id

dataset_id str | None = None #

Synthesize an event about this dataset.

TriggerDryRunRequest.event

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

A platform event to evaluate as-is.

TriggerDryRunRequest.event_id

event_id str | None = None #

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

TriggerDryRunRequest.event_type

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

Which subscribed event type to synthesize for an entity reference.

TriggerDryRunRequest.file_id

file_id str | None = None #

Synthesize an event about this file.

TriggerDryRunRequest.invocation_id

invocation_id str | None = None #

Synthesize an event about this invocation.

TriggerDryRunRequest.scheduled_for

scheduled_for datetime.datetime | None = None #

For a schedule-fired trigger: 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.

TriggerDryRunResponse

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

Attributes

TriggerDryRunResponse.event_type

event_type roboto.domain.platform_events.PlatformEventType #

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

TriggerDryRunResponse.gates

gates list[TriggerDryRunGate] #

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

TriggerDryRunResponse.trigger_id

trigger_id str #

The trigger that was dry-run.

TriggerDryRunResponse.verdict

verdict str #

One plain-English sentence summarizing the outcome.

TriggerDryRunResponse.would_fire

would_fire bool #

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