Events and consumers

Every mutation in Radd appends to one table: the event outbox. This page covers declaring an event type, emitting one, and the consumers that read them back.

Declare an event type

A plugin lists its event types on event_types=(...), as EventTypeSpec values, not as string literals typed at each call site:

@dataclass(frozen=True)
class EventTypeSpec:
    """A registered event type. `trigger=True` makes it an automation trigger with
    the builder metadata the UI needs — the inversion of the hardcoded catalog."""

    event_type: str
    label: str
    group: str = "Other"
    item_scoped: bool = False  # a target item resolves → SLQ + item actions apply
    has_changes: bool = False  # payload carries a field diff (old/new subjects work)
    trigger: bool = True  # appears in the automation trigger catalog
    entity_type: str = ""  # the entity this event is about (for auto-registered CRUD events)
    subjects: tuple[str, ...] = ()
    payload_schema: dict[str, Any] = field(default_factory=dict)

— server/src/radd/kernel/specs.py

event_type is a plain string on the spec, but a plugin never writes that string twice. It names the value once, as a member of a StrEnum, and reuses the member everywhere — the EventTypeSpec, the emit() call, and any consumer that filters on it:

class CsatEvent(StrEnum):
    # Both emitted with entity_type=item (item History feed) and actor_id=None
    # (a requester has no user; automation rules on them still fire — the
    # engine skips SYSTEM-actor events as its loop guard).
    REQUESTED = "csat.requested"
    RESPONDED = "csat.responded"

— server/src/radd/modules/csat/types.py

CLAUDE.md states the rule this follows for the whole codebase: "Anything that names a behavior, state, or type is a StrEnum or a dataclass/Settings field." A typo in a bare string compiles and runs; a typo in an enum member name fails at import. One exception exists, and it is worth naming precisely: a table auto-wired from an EntitySpec (see Architecture: the kernel and plugins) has no plugin code to hold an enum, so the kernel builds its event type names by string formatting — f"{key}.{verb}" for created/updated/deleted. That path is generated once, by the kernel, not authored per call site, which is the distinction the rule is protecting against.

subjects names the entity types the event is ABOUT, as ids the kernel resolves into refs — see Architecture: the kernel and plugins for EntityRefSpec. The loader refuses to boot a plugin whose event names a subject nothing describes.

Emit

A plugin calls events.service.emit inside the same session as the write the event describes, so the two commit or roll back together:

async def emit(
    session: AsyncSession,
    *,
    event_type: StrEnum,
    entity_type: StrEnum,
    entity_id: object,
    actor_id: uuid.UUID | None = None,
    payload: dict[str, Any] | None = None,
    subjects: dict[str, Any] | None = None,
    occurred_at: datetime | None = None,
    silent: bool | None = None,
    automated_cause: bool | None = None,
) -> None:
    """Append to the outbox inside the caller's transaction — commits or rolls back with it.
    ...
    **`subjects` are IDS; the kernel writes the shape (RADD-923.)** Pass
    `subjects={"item": item_id}` and `payload["item"]` becomes the canonical ref
    for that entity, resolved through `registries.entity_refs`. Emitters do not
    build refs, so they cannot build them differently — which is what fourteen
    of them had done before RADD-922 fixed it by hand.
    """

— server/src/radd/modules/events/service.py

A plugin's own worked example — csat, mailing a satisfaction survey when an item resolves — creates its row and emits csat.requested in one call:

async def create_survey(
    session: AsyncSession,
    *,
    item_id: uuid.UUID,
    item_key: str,
) -> CsatSurvey:
    """Mint the item's one-and-only survey row + emit csat.requested in the same
    transaction (the sender commits both with its cursor, before sending)."""
    survey = CsatSurvey(
        item_id=item_id, token=secrets.token_urlsafe(32), sent_at=utcnow()
    )
    session.add(survey)
    await session.flush()
    await events.emit(
        session,
        event_type=CsatEvent.REQUESTED,
        entity_type=ItemEntity.ITEM,
        entity_id=item_id,
        actor_id=None,
        subjects={"item": item_id},
    )
    return survey

— server/src/radd/modules/csat/service.py

The plugin declares the event once (CsatEvent), lists it on the manifest (event_types=(EventTypeSpec(CsatEvent.REQUESTED, "CSAT survey sent", "Service desk", item_scoped=True), ...), in modules/csat/__init__.py), and emits the same enum member. Nothing else in the codebase spells "csat.requested".

Subscribe and run: the consumer half of the same example

A consumer is an offset-tracked poll loop: it reads events after its stored cursor, does something with each, then advances the cursor. The shared scaffold for the common shape — deliver something outside the instance, at most once — is events.runner.run_head_seeded:

async def run_head_seeded[P](
    consumer_name: str,
    *,
    batch_size: int,
    plan: Callable[[AsyncSession, Event], Awaitable[P | None]],
    deliver: Callable[[list[P]], Awaitable[None]],
) -> int:
    """One consumer iteration (see module docstring). Returns the number of
    events consumed (0 on first-start seeding / an empty stream)."""
    async with SessionLocal() as session:
        if not await service.offset_exists(session, consumer_name):
            head = await service.latest_event_id(session)
            await service.set_offset(session, consumer_name, head)
            await session.commit()
            logger.info(
                "%s: first start — cursor seeded at stream head %s", consumer_name, head
            )
            return 0
        offset = await service.get_offset(session, consumer_name)
        batch = await service.read_after(session, offset, batch_size)
        if not batch:
            return 0
        plans: list[P] = []
        for event in batch:
            if event.silent:
                continue
            try:
                planned = await plan(session, event)
            except Exception:
                logger.exception("%s: planning failed for event %s", consumer_name, event.id)
                continue
            if planned is not None:
                plans.append(planned)
        await service.set_offset(session, consumer_name, batch[-1].id)
        await session.commit()  # planning writes + the cursor land BEFORE delivery
    if plans:  # post-commit, post-cursor: at-most-once delivery
        await deliver(plans)
    return len(batch)

— server/src/radd/modules/events/runner.py

The CSAT sender supplies plan and deliver to this scaffold. plan runs the guards — did an item.updated event actually move the item into a done state, is CSAT turned on for the project, is there an unsurveyed item and a resolvable recipient — and, on a pass, writes the survey row through create_survey above:

async def _plan(session: AsyncSession, event: Event) -> SurveyEmail | None:
    if event.event_type != ItemEvent.UPDATED.value:
        return None
    return await process_event(session, event)


async def run_once() -> int:
    return await runner.run_head_seeded(
        CONSUMER_NAME, batch_size=settings.csat_batch, plan=_plan, deliver=_deliver_all
    )

— server/src/radd/modules/csat/sender.py

run_once is scheduled by a PeriodicLoop, started and stopped from the plugin's on_startup/on_shutdown:

_loop = PeriodicLoop(
    sender.run_once,
    interval=lambda: settings.csat_poll_seconds,
    name="csat-sender",
    enabled=lambda: settings.run_workers,  # web-only process skips (spec 48 worker split)
)

start = _loop.start
stop = _loop.stop

— server/src/radd/modules/csat/dispatcher.py

offset_exists is why a newly enabled CSAT plugin never mails a survey for every item that resolved before it existed: the first tick seeds the cursor at the current stream head and returns without processing anything.

Not every consumer fits this scaffold. search's indexer replays the whole backlog as its index build; notify bootstraps only over watched rows; automations and webhooks have their own shapes, covered below. A plugin picks run_head_seeded when it delivers something outward, at most once, and must never replay history into a channel or a recipient.

The worker split: RADD_RUN_WORKERS

A Radd deployment can run two kinds of process from the same image: a web-only replica and a worker replica. settings.run_workers (env RADD_RUN_WORKERS) decides which one a process is:

    # Worker split (spec 48): false = this process serves web only; the
    # background loops (webhooks/automations/notify/search/sla/googlechat/mail)
    # stay dormant. Realtime + storage init always run.
    run_workers: bool = True

— server/src/radd/config.py

Every consumer loop gates its own enabled callback on this flag, the way the CSAT sender does above. A web-only replica boots with every consumer's PeriodicLoop.start() returning immediately:

    async def start(self) -> None:
        if not self._enabled():
            return
        self._task = asyncio.create_task(self._run(), name=self._name)

— server/src/radd/worker.py

Realtime and storage initialization skip this gate and run in every process, because they serve the web tier directly rather than draining the outbox. The deployment chart keeps exactly one worker replica, because a consumer's offset has one writer by design:

- **Worker split**: `--set workers.enabled=true` gives web replicas
  `RADD_RUN_WORKERS=false` and adds ONE worker pod running the loops
  (webhooks, automations, notifications/email, search indexing, SLA clock,
  connectors). Keep `workers.replicas: 1` — consumers are single-writer by
  design (per-consumer offsets); scale beyond that means NATS per the plan.

— docs/deploy.md

A second worker replica would race the same consumer offset with no lock between them — two processes reading the same batch and both racing to advance the same cursor row.

Consumer health: lag, heartbeat, and OK / catching-up / STALLED

events.service.consumer_status compares every consumer's stored cursor against the stream head, computed server-side so no client clock skews the result:

async def consumer_status(session: AsyncSession) -> list[dict[str, Any]]:
    """Every consumer's cursor vs the stream head, for monitoring: name, lag,
    and seconds since the cursor last moved (computed server-side against the
    same clock that wrote `updated_at`, so timezones can't skew it)."""
    head = await latest_event_id(session)
    result = await session.execute(
        select(
            ConsumerOffset.name,
            ConsumerOffset.last_event_id,
            func.extract("epoch", func.now() - ConsumerOffset.updated_at),
        ).order_by(ConsumerOffset.name)
    )
    return [
        {
            "name": name,
            "last_event_id": last_event_id,
            "stream_head": head,
            "lag": max(0, head - last_event_id),
            "seconds_since_update": max(0, int(seconds or 0)),
        }
        for name, last_event_id, seconds in result.all()
    ]

— server/src/radd/modules/events/service.py

GET /monitoring/overview (instance administrators only) serves this list alongside workers_in_process (the process's own run_workers value), so a web-only replica shows correctly as "no movement expected" rather than looking broken. Settings → Monitoring turns lag and the heartbeat into three states:

/** A consumer with backlog whose cursor hasn't moved in this long is stalled. */
const STALL_AFTER_SECONDS = 120;

type WorkerState = "ok" | "catching_up" | "stalled";

/** Zero lag = fine no matter how old the cursor is (idle stream). Backlog is
 * fine while the cursor keeps moving; backlog + a frozen cursor = stalled. */
function workerState(worker: WorkerStatus): WorkerState {
  if (worker.lag === 0) return "ok";
  return worker.seconds_since_update > STALL_AFTER_SECONDS ? "stalled" : "catching_up";
}

— web/src/routes/settings/monitoring.tsx

lag alone cannot answer the question a consumer's health check needs, since a busy but healthy consumer chewing a large backlog also has nonzero lag. The heartbeat resolves it: lag with a cursor that keeps moving is catching_up; lag with a cursor frozen past 120 seconds is stalled.

events.quiet and events.silent

The events table carries a silent column. A bulk import — the Jira importer is the one that exists today — wraps its whole run in events.quiet(), and every event emitted inside is marked silent with no change to the service functions doing the emitting:

@contextmanager
def quiet(enabled: bool = True) -> Iterator[None]:
    """Mark every event emitted in this scope `silent`.

    `enabled=False` is a no-op, so a caller can pass a user-facing toggle straight
    through without branching:

        with events.quiet(plan.quiet_import):
            ...
    """
    token = _quiet.set(enabled)
    try:
        yield
    finally:
        _quiet.reset(token)

— server/src/radd/modules/events/quiet.py

The distinction silent draws is "does this consumer tell someone outside the instance," not "does this consumer care." Consumers that reach outward — notify, webhooks, automations, realtime, and everything built on run_head_seeded (the CSAT sender above included) — skip a silent row:

        for event in batch:
            if event.silent:  # a bulk import must not fan 45k deliveries at subscribers
                continue

— server/src/radd/modules/webhooks/service.py

Consumers that build internal state do not skip it: the search index, so an imported issue is findable, and the audit/history reads (query_events, entity_activity), so an imported issue has history. A skip happens inside the loop, and the cursor still advances past a silent row — filtering silent rows out in the SQL query would let an all-silent batch read as an empty stream and wedge the cursor in place.

automated_cause is the sibling scope, for a different problem: an automation action that writes through the same services as a person would emit events indistinguishable from a person's own, which could re-trigger the rule that caused it. events.automated() marks the scope's events so the automations engine's loop guard recognizes and skips its own effects — see server/src/radd/modules/events/quiet.py for the full mechanism; it is outside this page's scope.

Ordering and failure/retry behaviour

The outbox orders itself: id is a BigInteger Identity primary key, so read_after(session, offset, limit) always returns events in the order they committed, and a cursor is a single integer. Nothing reorders a batch.

Failure behaviour differs by consumer, because "failure" means different things at different layers.

A single event's planning failure never blocks the batch. run_head_seeded and the cascade consumer both log and continue on one event's exception, then still advance the cursor past it:

            try:
                result = await spec.sweep(session, parent_id)
            except Exception:  # noqa: BLE001
                # One module's cleanup must not stop another's, nor wedge the cursor
                # on a row that will never succeed. Loud, and the stream moves on.
                logger.exception("cascade %s failed for %s %s", spec.name, event.event_type, parent_id)
                continue

— server/src/radd/modules/events/cascade.py

There is no automatic retry at this layer. A failed plan is a permanent skip for that event; only a fix and a manual replay (advancing the offset backwards, or rerunning against the row directly) recovers it.

The automations engine applies one event per transaction, so one rule's failure cannot roll back another event already committed in the same batch:

    for event in batch:
        if not should_process(event):
            continue
        try:
            async with session_factory() as session:
                await apply_event(session, event)
                await session.commit()
        except Exception:
            logger.exception("automations: failed processing event %s", event.id)
    async with session_factory() as session:
        await events.set_offset(session, CONSUMER_NAME, batch[-1].id)
        await session.commit()

— server/src/radd/modules/automations/engine.py

Here too, nothing retries a failed event — the cursor advances past the whole batch regardless of which individual events raised.

A delivery that leaves the instance gets its own retry policy, because network failure there is routine rather than a bug. Webhook deliveries retry on a backoff schedule and dead-letter once it is exhausted:

    schedule = settings.webhook_retry_schedule
    if delivery.attempts > len(schedule):
        delivery.status = DeliveryStatus.DEAD.value
    else:
        delivery.next_attempt_at = utcnow() + timedelta(seconds=schedule[delivery.attempts - 1])

— server/src/radd/modules/webhooks/service.py

with the schedule itself a plain tunable:

    # Seconds until the Nth retry after a failed attempt; exhausted -> dead-letter.
    webhook_retry_schedule: tuple[int, ...] = (5, 300, 1800, 7200, 18000, 36000)

— server/src/radd/config.py

Outbound notification email carries a similar, independent ladder per row — email_attempts and email_next_try columns, three backoff rungs, then the mailer stamps the row emailed_at to mean "this channel is finished with this row." The same stamp also stops the digest loop from picking the row up again:

#: How long to wait after the first, second and third consecutive failure. A
#: fourth failure exhausts the ladder and the row is given up on.
EMAIL_RETRY_DELAYS: tuple[timedelta, ...] = (
    timedelta(minutes=1),
    timedelta(minutes=5),
    timedelta(minutes=30),
)

— server/src/radd/modules/notify/retry.py

The pattern across all three: the outbox cursor never retries — a batch is consumed once, forever, and a planning failure is a permanent skip logged for a human to notice. Delivery, where failure is expected, gets its own retry state on its own rows, owned by the module doing the delivering.

---

Mirrored from project.radd-hq.com on 2026-09-12. Documentation is written there; this copy is regenerated by scripts/publish_wiki.py and hand edits do not survive it.