Triggers

Triggers start work without an interactive user.

One event has two possible readers. It can start new work, and it can resume work that already waits for it.

raw source -> EventProducer -> PlatformEvent -> EventRouter
                                            │
                        ┌───────────────────┴───────────────────┐
                        ▼                                       ▼
                  TriggerMatcher                         Inngest, as
                        │                            platform/<event.type>
                        ▼                                       │
                TriggerDispatchService                          ▼
                        │                              a waiting Run resumes
                        ▼
                 StartRunCommand
                        │
                        ▼
                    RunManager

The Trigger layer stops at RunManager. It does not execute Agents, implement policy, or own another scheduler/queue.

The event router

An inbound email reply must do two different things. It may start a Run, when a Trigger matches it. It must also resume the email sequence Run that already waits for that reply. The second reader was missing: an Inngest wait resolves only when an Inngest event arrives, and no producer sends one.

EventRouter is that one fan-out point.

class EventRouter:
    async def dispatch(self, event: PlatformEvent) -> None:
        # 1. resume: every platform event reaches Inngest under one name
        await self.inngest.send(InngestEvent(
            name=f'platform/{event.type}',
            id=f'event:{event.id}',        # the same stable ID the producer minted
            data=event.as_dict(),
        ))
        # 2. start: the existing trigger path, unchanged
        await self.trigger_dispatch.handle(event)

Two properties make this cheap.

• No wait index table. A wait node declares its own correlation, and Inngest matches it. Nothing tracks which Runs wait for what. This page is where that rule is stated; the other pages point here. agent.provider_jobs is not an exception to it: it holds one external job's state, and the wait on it still carries its own correlation.
• The whole envelope is the payload, and a wait node reads two levels of it. The expression tests async.data.organization_id for the tenant and async.data.data.<key> for the correlation. Send the business fields alone and every wait times out reporting that nobody sent an event somebody did send.
• No producer learns about Runs. A producer still emits one PlatformEvent and knows nothing below it.

The send is idempotent on event.id, so a duplicate delivery resumes one Run once, exactly as it starts one Run once.

A platform event that no Run waits for and no Trigger matches is dropped by both readers. That is normal, and it costs one Inngest event.

Components

The trigger row

agent.triggers holds one row per machine entry point. The row is the whole configuration. There is no second table and no per-row scheduler state.

A schedule row carries cron and timezone and no event_type. An event row carries event_type and no cron. A check constraint holds that, because a row that carries both is a row two producers claim.

The event contract

@dataclass(frozen=True)
class PlatformEvent:
    id: str
    organization_id: str
    type: str
    producer: str
    occurred_at: datetime
    subject: EventSubject | None
    data: dict
    metadata: dict

metadata carries transport/causality such as caused_by_run_id and caused_by_trigger_id. Business matching fields live in data.

A PlatformEvent is not another event store. The producer source is authoritative; Runs/spans become the durable execution record after dispatch.

The event type carries its version

A platform event the platform itself produces ends its type with a version: agentic.provider_job.completed.v1. A wait node stores that string in a published definition, and a definition outlives the code that emits the event.

platform produced   <domain>.<subject>.<verb>.v<n>
vendor sourced      <provider>.<verb>            the vendor owns the name

Bump the version when the fields a match may read change name, type or meaning. A new version is a new event type. A Run parked on .v1 never matches .v2. It times out on its own clock instead of reading a payload it cannot parse.

Adding a field is not a bump. A match reference must resolve to a scalar, so a reader never sees the whole payload and a new field breaks nothing.

⚠️ A vendor event type carries no version, because the platform does not own it. The adapter catches a vendor that changes its payload. It is the one place that reads the vendor's vocabulary.

Producers

class EventProducer(Protocol):
    async def produce(self, raw_event) -> PlatformEvent: ...

Every producer does the same four things: verify the source, normalize it, mint a stable event.id, and emit one PlatformEvent. Only the last two columns below actually differ, so read the table as one shape with four id rules.

The id rule is the load-bearing column. It must be stable for one logical occurrence and derived, never random, or a retried delivery or a replayed tick starts a second Run. A producer never queries trigger definitions or creates Runs.

A webhook producer acknowledges first

EventRouter.dispatch() sends to Inngest, matches triggers and starts a Run. That is too much work to hold a vendor webhook open for. Nylas and every other provider retries on a slow answer, so an inline dispatch turns one reply into a retry storm.

A webhook producer therefore uses the same boundary the Channel Gateway already uses.

vendor webhook
   ├─ verify the signature and the timestamp window
   ├─ produce the PlatformEvent, so the ID is minted from the vendor delivery ID
   ├─ IdempotencyService.claim(UUID(organization_id), 'webhook.<provider>',
   │                           delivery_id, sha256(raw_body))
   ├─ durable enqueue                -> Inngest
   └─ ACK                            -> under 3 seconds
                │
                ▼
          EventRouter.dispatch()     -> on a worker, not on the web request

The claim completes only after the enqueue succeeds, so a failed enqueue leaves the vendor free to retry. This is one shared rule with two callers, not a second gateway.

The claim answers processing for a delivery already in flight and conflict for a delivery id the vendor re-sent with different content. This boundary answers an ACK in both cases and never raises, exactly as the Gateway does: a 5xx tells the vendor that a delivery it made successfully failed. See idempotency.

A CronProducer and a CrmChangeProducer need none of this. Neither has a caller waiting on an HTTP answer, so both call EventRouter.dispatch() directly.

A provider callback uses this boundary and starts no Run. It resumes one.

The route adds two steps to the four above. It reads the job row first, because the claim is organization scoped and a vendor callback names no organization. It stores the state transition before it enqueues, because the row is the durable truth and the event is only a wake-up signal.

That page owns the full order, and this page does not restate it. See asynchronous provider jobs.

The schedule producer is one platform cron

An Inngest cron trigger is declared at registration, so a schedule an admin writes at 10:00 cannot become one. The platform registers one Inngest cron, trigger.cron, at a one minute cadence. It reads the enabled schedule rows, evaluates each cron expression with croniter in that row's timezone, and emits one PlatformEvent for each row that is due.

This is a producer and not a scheduler. It owns no queue, no durable timer and no per-row cursor.

trigger.cron  (* * * * *)
  -> read enabled rows of kind 'schedule'
  -> croniter(row.cron, row.timezone): is this minute a match?
  -> PlatformEvent(type='schedule.tick', id=f'schedule:{row.id}:{local_minute}')
  -> EventRouter.dispatch()

Four rules make the tick safe.

• The id names the matched minute, and not now(). A tick at 09:00:03 and its retry at 09:00:47 floor to one value, so they start one Run.
• The minute is the local wall clock minute, and it carries no offset. The row is written by a person who meant a wall clock time, so 0 1 * * * fires once a day. On the autumn shift 01:30 happens twice, and both instants floor to 2026-10-25T01:30: the second start key conflicts, RunManager answers duplicate, and the dispatcher records one skip. A UTC id would mint two ids for that one wall clock tick and start two Runs. On the spring shift the named minute does not exist, and no tick fires. That is what every cron does, and the row is not special.
• A missed tick is not replayed. The producer emits the current minute alone. A worker that was down for thirty minutes emits one event and not thirty. A catch-up turns one outage into a run storm that the day cap eats.
• The scan is bounded. The read is one indexed query with a hard row cap, as idempotency.sweeper is. Past the cap the producer logs a warning that names the count.

The CRM change producer polls, and keeps no cursor

A CRM row is written by the legacy product stack and by the platform tools, so no single write path can emit the event. CrmChangeProducer runs on the same trigger.cron minute and reads the rows whose updated_at moved inside a short window. The window overlaps on purpose.

trigger.cron
  -> read public.crm_companies where updated_at > now() - interval '5 minutes'
  -> PlatformEvent(type='crm.company.updated',
                   id=f'crm.company:{row.id}:{row.updated_at}')
  -> EventRouter.dispatch()

The id is the row id and its version, so an overlapping window re-emits an event Inngest already holds and a start key already claims. Both readers drop it. That is why the producer needs no cursor table and no exactly-once read.

The producer emits every changed row, and the trigger row decides which matter. Scoping is a conditions expression an admin writes, and not a filter inside the producer. One producer then serves every organization.

The event carries a fixed fact set, and the names are the column names. A condition reads data.id, data.name, data.lifecycle_stage, data.industry, data.country, data.lead_score, data.tags and data.owner_id. A path outside the set resolves absent, answers False, and the row never fires.

• A soft deleted row is not a change this producer emits. deleted_at moves on the delete, so the row would otherwise start a Run against a company the product already removed. The platform mints no crm.company.deleted yet.
• A full scan warns, and it does not raise. This differs from the schedule scan on purpose. A platform holding more than 500 enabled schedule rows is a bound an operator raises once; one bulk import writes more CRM rows in five minutes than any bound holds. Raising would emit nothing for the whole import and retry a doomed read every minute, so the pass stays best effort and names the count.

Matching

class TriggerMatcher:
    async def match(self, event: PlatformEvent) -> list[TriggerDefinition]: ...

Use the same deterministic ConditionEvaluator as Policy. Different callers expose different fields, but the platform does not grow a second expression language.

event.type = crm.company.updated
filter     = event.data.lifecycle_stage == 'qualified'
target     = sonar

Dispatch

class TriggerDispatchService:
    async def handle(self, event: PlatformEvent) -> None:
        for trigger in await matcher.match(event):
            if event.metadata.get('caused_by_trigger_id') == trigger.id:
                skip('self_reentry')
                continue

            await run_manager.start(StartRunCommand(
                definition_id=trigger.target_definition_id,
                input=render_input(trigger, event),
                actor=service_identity_for(trigger, event),
                source=RunSource.trigger(trigger.id),
                idempotency_key=f'trigger:{trigger.id}:event:{event.id}',
            ))

The unique key on agent.runs makes (trigger_id, event_id) atomic: the insert is the claim, so the second delivery reads the first delivery's Run back. There is no Idempotency Service on the start path and no transaction. Do not implement a read-then-insert duplicate check.

event.id must therefore be stable per logical occurrence. A vendor webhook uses the provider delivery ID. A CronProducer must derive the ID from the schedule and the tick timestamp, never from a random value, or a retried tick starts a second Run.

render_input is a seam, and not a template

A trigger row names a builder, and the registry answers it. The default builder is static: it renders input_config as the Run input and reads nothing.

class TriggerInputBuilder(Protocol):
    async def build(self, trigger: TriggerDefinition, event: PlatformEvent) -> dict: ...

A static template cannot start a scheduled saved search.SmartFeedPublisher reads run.input['baseline_run_id'] and refuses a run whose baseline is not the saved search's current last_run_id. That value changes on every run, so a fixed template publishes a diff once and answers stale for ever after. The saved_search builder reads SavedSearchRepository and returns the same input the interactive route freezes today. See signals search.

The registry keeps the layer clean. A product registers its builder, and the dispatcher imports the registry alone. triggers/ imports no product package.

The dispatcher passes a service identity. RunManager mints the service Principal at admission.

service_identity_for() fills three fields off the trigger row: the trigger id, the scopes the admin authored, and authored_by. ActorIdentity refuses a trigger actor that names no author, and PrincipalFactory meets the authored scopes with that admin's rights, so a trigger is never wider than its author. Author the row with tool names alone: run.start is added before that meet, and the author's role decides whether it survives. A trigger row that carries no scopes mints a grant that holds nothing, and every run it starts is denied at admission. See policy and governance.

Protections

Do not build general graph-cycle detection in V1. Direct self-reentry is blocked; indirect loops are bounded by budget, concurrency and kill switch until a real need proves otherwise.

Example: inbound email

Nylas webhook
  -> verify transport
  -> PlatformEvent(email.received, stable vendor id)
  -> EventRouter
       ├─ platform/email.received -> the email sequence Run waiting on this thread resumes
       └─ TriggerMatcher -> TriggerDispatchService
              -> StartRunCommand(idempotency = trigger + event)
              -> RunManager

Three retries of the same Nylas webhook create one Run, and resume one waiting Run once. The email body remains untrusted content inside that Run.

Both readers can act on one event, and that is correct. A reply resumes the sequence that sent the mail, and a separate Trigger may still file the reply against the CRM.

Example: scheduled workflow

CronProducer -> PlatformEvent -> EventRouter -> TriggerMatcher -> RunManager -> Workflow Run

A Workflow gets no private scheduler. A schedule is a Trigger like every other machine entry point.

Observability

Every skip has a distinct reason:

disabled | duplicate | self_reentry | debounced | concurrency | budget

Three of the six come back from RunManager.start(), and the rest are decided here. The dispatcher owns the mapping, and it is the whole of it.

StartOutcome has ten members, and the dispatcher maps every one. A member this table did not name would reach the counter as an unknown answer.

⚠️ started covers three different ends, and only one of them executes.RunManager.start() answers started for a run it dispatched, for a run it parked on an admission approval, and for a run a policy rule denied. The dispatcher therefore reads result.run.status and result.error_code, and never the outcome alone.

A parked machine run is the case that dies quietly, because it has no conversation. See policy and governance.

Where a skip is recorded

A skip writes no Run, so it writes no span, and it is not a policy decision. Two records carry it, and neither is a new table.

• The trigger row. last_outcome, last_outcome_at and last_run_id hold the answer of the last dispatch. Trigger admin shows it, so the admin who wrote the row reads why nothing happened.
• A structured log line, carrying the organization, the trigger, the event and the reason. A fault also reports to Sentry. A skip does not: a budget skip is an ordinary answer, and paging on it trains an operator to ignore the channel.

budget comes from the accrual gate before a Run exists. duplicate cannot be decided here, because the rule above forbids a read-then-insert check: the start key is the only duplicate guard, so its answer is the only one the dispatcher has. disabled can be decided here for a definition that was already disabled at match time, and it still arrives from start() when an admin disables it in the window between the match and the start.

self_reentry, debounced and concurrency are decided here and reach start() never.

A silent no-op is indistinguishable from broken automation, so skip outcomes must be counted/recorded.

Code shape

triggers/
  models.py             PlatformEvent
  producers/
    base.py             EventProducer
    nylas.py
    webhook.py
    cron.py
    crm.py
  router.py             EventRouter
  matcher.py            TriggerMatcher
  dispatcher.py         TriggerDispatchService
  input_builders.py     TriggerInputBuilder, and the static builder

The registry holds the builders. A product owns its own builder and registers it, so triggers/ imports no product package.

Neither the router, the matcher nor the dispatcher imports Agno.

Rules

• A new source means a new producer, not new dispatch infrastructure.
• A webhook producer verifies, claims, enqueues and acknowledges. It never dispatches inline.
• Every PlatformEvent reaches both readers through EventRouter. A producer never calls one of them directly.
• A resume needs no index. The waiting node owns its own correlation.
• Matching is deterministic and model-free.
• Reuse ConditionEvaluator.
• A Trigger starts an Agent or Workflow through RunManager only.
• Email is produced by Nylas directly; it does not route through Channel Gateway.
• An interactive channel is not an event producer. An unknown sender receives a link invitation from the Channel Gateway, and starts no Run.
• Duplicate delivery is harmless.
• Direct self-reentry is denied by default.
• Policy and Inngest retain their existing responsibilities.

Minimum contract tests

• Same (trigger_id, event_id) creates one Run under concurrency.
• One PlatformEvent reaches the trigger matcher and Inngest, and a duplicate delivery does neither twice.
• An email reply resumes the waiting sequence Run, and may also match a Trigger.
• A trigger cannot directly fire itself from its own caused event by default.
• Nylas retries do not duplicate work.
• A webhook producer answers in under 3 seconds, whatever the router does afterwards.
• A failed enqueue leaves the webhook claim unfinished, so the vendor retry still works.
• Trigger filters and Policy conditions share evaluator semantics.
• Disabled trigger never reaches RunManager.
• Budget denial is produced by the accrual gate in RunManager.start(), not duplicated in trigger code.
• A budget denied trigger creates no Run row, and it is counted as a budget skip.
• Two ticks of one minute, forty seconds apart, start one Run.
• A producer that was down for thirty minutes emits one event, and not thirty.
• A schedule in a local zone keeps one id across a daylight saving shift.
• Two enabled triggers matching one event start two Runs, on two start keys.
• A run the admission parked answers started with status waiting, and the dispatcher records it as parked and not as started.
• A second scheduled saved search run carries the baseline the first run wrote, so the Smart Feed publishes a diff instead of answering stale.
• A trigger row that names an unknown input builder is refused at save.