Event System

All inter-component communication in Crewlet flows through a persistent event queue (crewlet.queue), backed by Apache Pulsar.

Interfaces

One protocol serves all inter-component communication:

EventQueue — persistent pub/sub with consumer groups. For fire-and-forget messages: task routing to agent inboxes, inbound/outbound notifications.

An in-memory implementation (MemoryEventQueue in crewlet.queue.memory) is used exclusively in tests. Production deployments use Apache Pulsar.

Topic Structure

# EventQueue topics (persistent, at-least-once via Apache Pulsar)
crewlet.agent.{handle}.inbox         # Per-agent inbox — all work arrives here
crewlet.notifications.inbound        # Inbound webhooks from external systems
crewlet.notifications.outbound       # Outbound messages to external systems

Routing

Routing is two-stage: events are first published to internal topics (e.g., crewlet.events.task_assigned), where the Engine’s subscription handlers determine the target agent and re-publish to that agent’s inbox topic. This keeps the event producers decoupled from the routing logic — they emit events without knowing which agent should handle them.

Handlers read the org through a provider on every event (never a captured snapshot), so a hot reload that swaps engine.org — including seat-kind flips — re-routes immediately.

When the resolved target is a human seat, the event is skipped: a human has no inbox and no turn to wake, and the engine never sends as itself. These internal events route to agents only; the human is notified natively by the PM tool / Slack where the work lives (and agents reach humans through their own colleague-surface tools with an @-mention). Inbound external-surface webhooks addressed to a human are likewise recorded as an info-level skip, not an undeliverable warning.

Inbox Batching & Coalescing

An agent turn takes minutes; webhooks arrive in seconds. Without batching, ten comments on one Jira issue that arrived while the agent was mid-turn would drain as ten sequential full turns — 10× LLM cost, each turn seeing one comment in isolation, potentially ten separate replies. Inbox delivery is therefore batched per conversation.

The buffer is the broker backlog — no second in-memory buffering layer. Holding events in process memory after their broker message was acknowledged would silently lose them on a crash; instead, the inbox consume loop changes how the backlog is drained:

subscribe_batch (EventQueue protocol; Pulsar + memory backends) implements steps 1–4: after the first message arrives it drains everything immediately available — plus anything arriving within BatchOptions.linger_seconds of the first message — up to BatchOptions.max_batch, partitions by a caller-supplied key, invokes the handler once per partition (sequentially — per-agent serialization is unchanged), and acknowledges a partition’s messages only after its handler returns. A failing partition negatively-acknowledges exactly its own messages (normal redelivery / DLQ policy per message) without blocking or replaying other conversations from the same drain. pause_delivery during collection NAKs anything fetched-but-undispatched so the next engine subscription gets it promptly.

Ack-budget deferral. Every drained message’s broker ack-timeout clock starts at receive, but a partition handler is typically a full multi-minute turn — dispatching a long tail of partitions sequentially would hold later messages delivered-but-unacked for the sum of preceding turns and blow through the 30-minute ack window mid-drain (redelivered duplicate turns). The Pulsar backend therefore dispatches partitions only while the drain is within a 60-second budget measured from the first receive — time spent lingering in collection counts against it, which is also why the linger config is capped at 60s; once the budget is spent, every remaining partition is requeued — each event republished to the topic (identity intact: same id, timestamp, trace; the event store’s (event_time, event_id) upsert stays idempotent), then the original acked. Requeued events carry zero accrued redeliveries (deferral can never push a healthy conversation toward the DLQ) and re-partition on the next drain, which begins right after the current turn — so throughput matches sequential dispatch while each message’s unacked window only ever covers one turn. Partitions dispatch oldest conversation first (by oldest constituent event timestamp, which survives requeue): a deferred conversation ages and outranks the hot conversation’s fresh arrivals on the next drain, so steady inflow on one issue cannot starve a waiting DM.

Conversation keys (crewlet.notifications.coalesce.conversation_key) are derived by pure logic from webhook metadata, via the same per-source NotificationPrompt classes that own prompt building: Jira keys on the issue (jira:POC-7), Confluence on the page, GitHub on repo#number. Slack keys on the whole channel for top-level DM and group-DM messages (channel_type im/mpim, or a D-prefixed channel id when the event variant omits the field — a human firing four rapid top-level DM messages is one conversation; a DM thread reply keeps its thread key so the merged trigger never carries the wrong reply target) and on channel + thread root elsewhere (slack:C9:1718.001 — a top-level channel message keys on its own ts so its replies join it, while two unrelated asks in a shared channel never merge). Everything else — task_assigned, A2A wakes, notifications without a derivable conversation — keys uniquely on the event id and is never coalesced: single-event partitions follow exactly the pre-batching dispatch path.

Busy agents queue; parked agents requeue. A delivery that finds its agent mid-turn does not fail: TurnEngine.run_turn WAITS for the running turn to finish (the handler already holds the delivery for a full turn, so the ack window — 30 minutes — is sized for a wait plus a worst-case turn). A delivery that finds the agent parked on a detached sandbox job (AWAITING_SANDBOX, potentially hours) is requeued + acked instead — the coordinator keeps the topic paused, so the copies buffer on the broker and flow when the job completes. Before the engine has a turn engine at all (booted with zero LLM providers), the handler pauses the topic and requeues likewise; the late turn-engine build resumes every inbox. Busy-agent handling therefore never consumes-and-drops and never pushes healthy events toward the dead-letter topic.

Unsubscribe. EventQueue.unsubscribe(topic, group) tears down the durable group consumer(s) for the pair and deletes the broker-side subscription (retained messages for the group are dropped). The engine calls it when a role is decommissioned live, so a removed seat neither keeps a consumer bound to a terminated instance nor accumulates undeliverable events forever. Inbox subscription is idempotent per agent handle — boot and the late turn-engine path both walk the pool, and only the first subscribe per agent creates a consumer.

The digest trigger. A multi-event partition is merged by coalesce_notifications into ONE ExternalNotification: a chronological digest of the earlier messages, then the latest constituent’s full enriched body — so the per-source scaffolding (triage rules, ## Get Full Context) renders exactly once and points at the most recent state. Two noise filters apply in the digest: per-source supersede rules (NotificationPrompt.digest_entry_body — Jira issue_updated bodies, stale full descriptions whose current state the Jira prompt never renders anyway, collapse to their event lead) and a source-agnostic same-sender duplicate dedupe — a constituent whose effective body is byte-identical to a later message from the same sender collapses to a marker (GitHub lifecycle webhooks re-emit the full PR description per event; the later copy still renders, so nothing is lost, and two different people saying “+1” both survive). Comments and messages always keep their text. The merged event carries every constituent in messages (sender, salient body, metadata, per-message recon flag — full fidelity for the learning workers, which observe each distinct sender), a conservative event-level recon merge, the max-depth constituent’s delegation bookkeeping (batching cannot launder the depth cap), and the latest constituent’s trace context. Same-id duplicate deliveries (an at-least-once edge the requeue machinery itself can produce) are dropped at the handler before any merging. If a partition cannot be merged (a malformed constituent), the engine degrades to per-event dispatch — the tail is requeued as independent inbox messages FIRST, then the first event runs in the current ack scope — so a requeue failure aborts before any turn ran and a completed turn is never replayed by a later event’s failure; partially-requeued copies collapse via the same-id dedupe on redelivery. A NotificationsCoalesced telemetry event records each merge for the dashboard / event store.

Two knobs (Tier B, hot-reloadable — see the configuration reference):

Field	Default	Meaning
`notification_coalesce_window_seconds`	`0`	Linger after the first pending event before dispatching. `0` adds no latency and still coalesces the busy case — backlog that accumulated during a turn is drained together regardless. A positive window (5–15 s) additionally absorbs bursts while the agent is idle (a human typing several messages, a Jira comment+status+assign webhook cluster).
`notification_coalesce_max_batch`	`20`	Cap per digest; a larger backlog arrives as successive capped batches rather than one unbounded megaprompt.

With the window at 0, an idle-agent burst worst-cases at two turns (the first message wakes the agent immediately; everything arriving during that turn coalesces into one follow-up turn per conversation) — never N.

Relation to the rate limiter. notification_rate_limit (NotificationService, default off) drops notifications above N/agent/second — it remains purely a safety valve against pathological webhook storms and notification loops. Burst handling is coalescing’s job: a coalesced comment is context preserved, a dropped one is context lost.

DACI decisions are conducted in Slack threads — the driver opens a thread in the team channel with its own Slack MCP tools and all contributions, proposals, and approvals are thread replies; there is no engine-side decision machinery. See Decision Framework for details.

Event Types

# Lifecycle
OrgStarted, OrgStopped
AgentSpawned, AgentTerminated, AgentReassigned
RoleUpdated              # role definition changed during config reload

# Task (routed to specific agent inboxes)
TaskCreated, TaskAssigned, TaskStarted
TaskCompleted, TaskFailed, TaskDelegated

# Communication
MessageSent              # agent sent a message to a channel

# Knowledge
DocumentCreated, DocumentUpdated

# Notification
ExternalNotification     # inbound from Jira, Slack, GitHub, email
NotificationSkipped      # dropped notification with reason (traceability)
NotificationsCoalesced   # N same-conversation inbox events merged into one
                         # digest trigger (see Inbox Batching above)

# System
AgentTurnCompleted       # full LLM reasoning cycle with tokens/tools
AgentTurnProgress        # incremental per-round updates (not persisted);
                         # carries turn_id/phase/iteration so live
                         # consumers can place in-flight rounds inside
                         # the turn/phase grouping
BudgetExhausted
TurnGuardBreach          # runtime invariant fired (stall / max_iter / depth_cap /
                         # unhandled_exception / scheduled_timeout).
                         # Drives the dashboard `afk` state.
LLMUnavailable           # FallbackLLMProvider chain exhausted.
                         # Drives the dashboard `afk` state.

Event Schema

Every event carries a common set of fields: a unique ID (UUID), a type string, a UTC timestamp, an optional source identifier, and a free-form payload dict. Specialized event types (e.g., TaskAssigned) add their own fields with defaults.

Events also carry OpenTelemetry trace context and self-describing properties:

class Event(BaseModel):
    id: UUID
    type: str
    timestamp: datetime
    source: str = ""
    payload: dict[str, Any] = {}

    # OpenTelemetry trace context — auto-captured from the active span
    trace_id: str       # W3C 32-char hex, groups causally related events
    span_id: str        # W3C 16-char hex, identifies this event in the trace
    parent_span_id: str # links to the event/span that caused this one

    @property
    def summary(self) -> str:
        """Human-readable 'who did what' — overridden by each subclass."""

    @property
    def actor(self) -> str:
        """Human-readable actor name (role > source > agent_id > 'system')."""

Each Event subclass defines its own summary property using its domain fields. New event types automatically get a reasonable default.

Changes are additive-only — new fields get defaults, existing fields are never removed. Pulsar retains each subscription’s undelivered backlog until it’s consumed (so a restart resumes cleanly); durable, replayable event history is the TimescaleDB event store, not the queue. Time-based retention of already-acknowledged messages is an optional Pulsar namespace retention policy.

Distributed Tracing

Events carry OpenTelemetry-compatible trace context (W3C Trace Context format). Trace IDs propagate automatically through the system — no manual threading:

How it works:

Webhook handlers create an OTel span → all events created inside inherit trace_id
When events cross async boundaries (EventQueue → handler), the receiving component restores the OTel context from the event’s trace_id/span_id
The Event model’s trace_id field defaults to current_trace_id() which reads the active OTel span

When notifications are dropped (own message, not following thread, rate limit), a NotificationSkipped event is emitted with the skip reason — visible in the trace so you can see why a webhook didn’t reach an agent.

The dashboard groups events by trace_id into collapsible trace trees. See Deployment — Tracing for OTLP export configuration.

Publish Listeners

The EventQueue supports publish listeners — async callbacks invoked inline during every publish() call. Listeners receive the topic and event, and run in the same coroutine as the publish. Exceptions in listeners are logged but do not prevent the publish or affect queue delivery.

This is used by the event store writer to persist events directly at publish time without routing through a Pulsar subscription. See Deployment — TimescaleDB Event Store for details.

Broadcast Streams (`subscribe_stream`)

Beyond competing-consumer subscribe() and inline add_publish_listener, the EventQueue exposes subscribe_stream(topic_pattern, handler) for live-stream consumers (dashboards, real-time log views). Every subscriber receives every matching event — no consumer-group division.

The Pulsar backend implements this with a per-caller regex topic-pattern subscription, started at the latest message (so it streams new events — backfill is served separately by the REST event store) and torn down when the caller unsubscribes. Because Pulsar discovers pattern-matching topics on a periodic cycle, a brand-new agent’s first events may lag the stream briefly; already-active topics match immediately. The memory backend implements it with a topic-filtered publish listener.

The dashboard’s /ws/stream endpoint uses this primitive: each connected tab is one ephemeral consumer, and the in-process StreamService both updates the live-state projection and fans every event out to every connected WebSocket. See API Endpoints — Live Stream.

topic_pattern accepts subject wildcards: * matches one segment, > matches one-or-more trailing segments.

Communication

Two communication systems:

External Channels (Slack, Email)

Org-wide announcements, department coordination, and team discussions happen through external tools (Slack channels, email) via the Notification Service. Agents use MCP tools to post and receive messages from Slack, and the notification service routes inbound webhooks to agent inboxes.

Org-wide — announcements (via Slack #announcements channel)
Department — leads-only coordination (via Slack department channel)
Team — team coordination, DACI decisions (via Slack team channel)

Ephemeral A2A Bus (`crewlet.a2a`)

Low-latency, in-memory channels for private 1:1 or small-group conversations between agents. The Engine manages lifecycle:

Aspect	External Channels (Slack)	A2A Bus
Lifetime	Permanent (Slack workspace)	Ephemeral (conversation)
Backend	Slack API + Notification Service	`asyncio.Queue` (in-process)
Persistence	Yes (Slack history)	No
Use case	Broadcasting, team coordination	Private agent-to-agent chat

Generated from crewlet/crewlet v0.1.0 at b40ea18.