Event Bus

Canonical source: docs/03-architecture/event-bus.md — this page is rendered from it on every publish; edit the source.

Event Bus Architecture

Overview

ClawAI uses RabbitMQ as its asynchronous event bus to decouple services and enable event-driven workflows. All inter-service events flow through a single topic exchange with routing-key-based filtering. The system includes dead-letter queues (DLQ) and retry mechanisms to ensure reliable message delivery.

---

Exchange Topology

                         +---------------------------+
                         |   claw.events             |
                         |   (topic exchange,        |
                         |    durable)               |
                         +---------------------------+
                            |   |   |   |   |   |
          +-----------------+   |   |   |   |   +------------------+
          |                     |   |   |   |                      |
          v                     v   |   v   v                      v
  +---------------+  +----------+  |  +---------+  +------------+  +----------+
  | routing.queue |  | chat.q   |  |  | audit.q |  | memory.q   |  | srvlog.q |
  | bind:         |  | bind:    |  |  | bind:   |  | bind:      |  | bind:    |
  | message.      |  | message. |  |  | *.#     |  | message.   |  | log.     |
  | created       |  | routed   |  |  |         |  | completed  |  | server   |
  +-------+-------+  +----+-----+  |  +----+----+  +-----+------+  +----+-----+
          |                |        |       |              |              |
          v                v        |       v              v              v
  routing-service   chat-service    |  audit-service  memory-service  server-logs
                                    |
                                    v
                            +-------+--------+
                            | DLQ exchange   |
                            | claw.events.dlx|
                            | (direct)       |
                            +-------+--------+
                                    |
                                    v
                            +-------+--------+
                            | DLQ queues     |
                            | (per service)  |
                            +----------------+

---

Event Catalog

message.created

Published when a user sends a new chat message.

Field	Type	Description
messageId	UUID	The newly created message ID
threadId	UUID	The thread this message belongs to
userId	UUID	The user who sent the message
content	string	The message text content
forcedProvider	string or null	User-forced provider (MANUAL_MODEL mode)
forcedModel	string or null	User-forced model
fileIds	UUID[]	IDs of attached files
contextPackIds	UUID[]	IDs of attached context packs
threadRoutingMode	RoutingMode	The thread's routing mode setting
threadPreferredProvider	string or null	Thread's preferred provider
threadPreferredModel	string or null	Thread's preferred model

Publisher: chat-service Consumers: routing-service

---

message.routed

Published when the routing engine has determined the optimal provider and model.

Field	Type	Description
messageId	UUID	The message being routed
threadId	UUID	The thread ID
selectedProvider	string	Chosen provider (e.g., "anthropic", "local-ollama")
selectedModel	string	Chosen model (e.g., "claude-sonnet-4")
confidence	float	Routing confidence score (0.0 - 1.0)
reasonTags	string[]	Why this selection was made
privacyClass	PrivacyClass	LOW, MEDIUM, or HIGH
costClass	CostClass	LOW, MEDIUM, or HIGH
fallbackProvider	string or null	Fallback if primary fails
fallbackModel	string or null	Fallback model
heuristicUsed	boolean	Whether Ollama router was bypassed
decisionId	UUID	The recorded routing decision ID

Publisher: routing-service Consumers: chat-service

---

message.completed

Published when an AI response has been generated and stored.

Field	Type	Description
messageId	UUID	The assistant message ID
threadId	UUID	The thread ID
userId	UUID	The user who initiated the request
userContent	string?	The original user message; absent on save turns (ADR-134)
useMemory	boolean?	Thread's memory switch; false = memory-service learns nothing from this turn (SEC-006)
content	string	The AI response
provider	string	Actual provider used
model	string	Actual model used
routingMode	RoutingMode	Routing mode that was active
inputTokens	number	Tokens in the prompt
outputTokens	number	Tokens in the response
latencyMs	number	Total execution time
fallbackUsed	boolean	Whether a fallback provider was used
fallbackProvider	string or null	Which fallback was used, if any
fallbackModel	string or null	Which fallback model, if any

Publisher: chat-service Consumers: memory-service, audit-service, routing-service

---

thread.created

Published when a new chat thread is created.

Field	Type	Description
threadId	UUID	The new thread ID
userId	UUID	The user who created it
title	string	Thread title
routingMode	RoutingMode	Initial routing mode

Publisher: chat-service Consumers: (currently no consumers; reserved for future analytics)

---

user.login

Published when a user successfully authenticates.

Field	Type	Description
userId	UUID	The authenticated user
email	string	User email
role	UserRole	ADMIN, OPERATOR, or VIEWER
sessionId	UUID	The new session ID
ipAddress	string	Client IP (for audit)
userAgent	string	Client user agent

Publisher: auth-service Consumers: audit-service

---

user.logout

Published when a user logs out or their session is invalidated.

Field	Type	Description
userId	UUID	The user who logged out
sessionId	UUID	The terminated session
reason	string	"manual", "expired", or "revoked"

Publisher: auth-service Consumers: audit-service

---

user.temporary_password_issued

Published after a temporary password has been successfully delivered via email, persisted by the auth service, and all active sessions for the target account have been revoked.

• Domain: identity
• Publisher: auth-service
• Consumers: audit-service

Field	Type	Description
userId	UUID	Target account that received the temporary password.
issuedBy	UUID	Administrator or system principal that issued the password.
timestamp	ISO-8601	When the event was emitted.
correlationId	string (optional)	Trace correlation identifier for the issuing flow.

This event intentionally does not carry the temporary password, any password hash, the user's email address, or a session token.

---

connector.created

Published when a new cloud provider connector is configured.

Field	Type	Description
connectorId	UUID	The new connector ID
name	string	Connector display name
provider	ConnectorProvider	Provider type
userId	UUID	User who created it

Publisher: connector-service Consumers: audit-service

---

connector.updated

Published when a connector's configuration is modified.

Field	Type	Description
connectorId	UUID	The updated connector ID
name	string	Connector display name
provider	ConnectorProvider	Provider type
changes	string[]	List of changed fields
userId	UUID	User who made the change

Publisher: connector-service Consumers: audit-service

---

connector.deleted

Published when a connector is removed.

Field	Type	Description
connectorId	UUID	The deleted connector ID
provider	ConnectorProvider	Provider type
userId	UUID	User who deleted it

Publisher: connector-service Consumers: audit-service

---

connector.synced

Published after a model sync operation completes for a connector.

Field	Type	Description
connectorId	UUID	The connector that was synced
provider	ConnectorProvider	Provider type
modelsAdded	number	New models discovered
modelsRemoved	number	Models no longer available
modelsUpdated	number	Models with changed metadata
totalModels	number	Total models after sync

Publisher: connector-service Consumers: audit-service, routing-service

---

connector.health_checked

Published after a connector health check completes.

Field	Type	Description
connectorId	UUID	The checked connector
provider	ConnectorProvider	Provider type
status	string	"healthy", "degraded", or "unhealthy"
latencyMs	number	Health check response time
error	string or null	Error message if unhealthy

Publisher: connector-service Consumers: audit-service, routing-service

---

routing.decision_made

Published after every routing decision for analytics.

Field	Type	Description
decisionId	UUID	The routing decision ID
messageId	UUID	The routed message
routingMode	RoutingMode	Mode used
selectedProvider	string	Chosen provider
selectedModel	string	Chosen model
confidence	float	Confidence score
heuristicUsed	boolean	Whether heuristic fallback was used
latencyMs	number	Decision computation time

Publisher: routing-service Consumers: audit-service

---

routing.models.synced

Published after a router model-knowledge sync run finishes. See routing-model-sync.md.

Field	Type	Description
runStartedAt	string	ISO timestamp the sync run began
runFinishedAt	string	ISO timestamp the sync run finished
durationMs	number	Total run duration
totals	object	{ upstreamCount, upsertedCount, skippedCount }
perProvider	array	Per-source { provider, status, upstream/upserted/skipped }

Publisher: routing-service Consumers: audit-service

---

memory.extracted

Published after memories are extracted from a completed message.

Field	Type	Description
userId	UUID	The user whose memories were extracted
messageId	UUID	Source message
threadId	UUID	Source thread
memoriesCreated	number	Count of new memories stored
memoriesSkipped	number	Count of duplicates skipped
types	string[]	Types of memories extracted (FACT, PREFERENCE, etc.)

Publisher: memory-service Consumers: audit-service

---

file.uploaded

Published when a file is successfully uploaded.

Field	Type	Description
fileId	UUID	The uploaded file ID
userId	UUID	The uploader
filename	string	Original filename
mimeType	string	MIME type
sizeBytes	number	File size

Publisher: file-service Consumers: (currently no consumers; reserved for future processing)

---

file.chunked

Published when a file has been chunked for context injection.

Field	Type	Description
fileId	UUID	The chunked file ID
chunkCount	number	Number of chunks created
totalCharacters	number	Total character count across chunks

Publisher: file-service Consumers: (currently no consumers; reserved for future indexing)

---

log.server

Published by all backend services for centralized log aggregation.

Field	Type	Description
level	string	"debug", "info", "warn", "error"
serviceName	string	Originating service
module	string	NestJS module name
action	string	What was being done
message	string	Log message
requestId	string or null	X-Request-ID correlation
traceId	string or null	Distributed trace ID
userId	UUID or null	Associated user
threadId	UUID or null	Associated thread
error	object or null	Error details (stack, code)
timestamp	ISO 8601	When the log was created

Publisher: every backend service Consumers: server-logs-service

---

image.generation.*

Events related to AI image generation lifecycle.

Event	Description
image.generation.started	Image generation job has been queued
image.generation.completed	Image has been successfully generated
image.generation.failed	Image generation failed

Publisher: image-service Consumers: audit-service

---

DLQ and Retry Strategy

Retry Configuration

Max retries: 3
Retry delays:
  Attempt 1: 1 second
  Attempt 2: 5 seconds
  Attempt 3: 30 seconds

Strategy: Exponential backoff with fixed delays

How Retries Work

1. Consumer receives a message and attempts processing
2. If processing throws an exception: a. Increment x-retry-count header b. If retry count < 3: republish to the same queue with a delay (using RabbitMQ delayed message plugin or TTL + dead-letter routing) c. If retry count >= 3: route to the DLQ
3. The DLQ (claw.events.dlq.<service-name>) holds failed messages for manual inspection

Dead-Letter Exchange

Exchange: claw.events.dlx (direct, durable)

Queues:
  claw.events.dlq.routing    <- failed routing messages
  claw.events.dlq.chat       <- failed chat messages
  claw.events.dlq.memory     <- failed memory extraction
  claw.events.dlq.audit      <- failed audit logging
  claw.events.dlq.server-logs <- failed log ingestion

DLQ Monitoring

Dead-letter queues should be monitored. Messages in the DLQ indicate:

• A consumer bug (most common)
• A downstream dependency failure (database, Ollama)
• A malformed event payload

Resolution: Fix the underlying issue, then replay messages from the DLQ.

---

Error Handling

Publisher-Side

• All event publishing uses the RabbitMQService wrapper from shared-rabbitmq
• If RabbitMQ is temporarily unreachable, the service logs a warning and continues (the HTTP response to the user is not blocked)
• Connection recovery is handled automatically by the amqplib client with reconnection backoff

Consumer-Side

• Each consumer wraps processing in a try-catch
• Known transient errors (database timeout, Ollama unreachable) trigger a retry via NACK + requeue
• Known permanent errors (invalid payload, missing required fields) trigger immediate DLQ routing (no retry)
• Unknown errors are retried up to the max retry count

Idempotency

Consumers should be idempotent because messages may be delivered more than once (at-least-once delivery). Key patterns:

• Memory extraction checks for duplicate memories before creating
• Audit logging uses messageId as a natural idempotency key
• Routing decisions are keyed by messageId (one decision per message)

---

Ordering Guarantees

Within a Single Queue

RabbitMQ guarantees FIFO ordering within a single queue with a single consumer. Messages published to the same queue are delivered in order.

Across Queues

No ordering guarantees exist across different queues. For example, message.created and message.routed may be processed by their respective consumers in any relative order. The system handles this through:

• Causal ordering via event chaining: message.routed is only published AFTER the routing service has finished processing message.created. The chat service only processes message.routed after it has already stored the user message.
• State checks: Before processing an event, consumers verify that prerequisite state exists (e.g., the message record exists in the database).

Parallelism

If multiple instances of a service consume from the same queue, messages are distributed round-robin. This means messages for different threads may be processed in parallel, but messages for the same thread are NOT guaranteed to be in order across instances. For chat messages, this is acceptable because each message triggers a distinct pipeline.

---

Shared RabbitMQ Package

The shared-rabbitmq package provides:

RabbitMQModule

NestJS dynamic module for RabbitMQ connection setup. Handles:

• Connection string configuration
• Exchange declaration
• Queue declaration and binding
• Connection recovery

RabbitMQService

Service class for publishing events:

• publish(routingKey: string, payload: unknown): Promise<void>
• Automatic serialization to JSON
• Adds standard headers: x-timestamp, x-source-service, x-request-id
• Retry on publish failure (3 attempts)

StructuredLogger

Logger utility that publishes log.server events:

• Wraps NestJS Logger
• Automatically includes service name, module, request context
• Redacts sensitive fields before publishing

---

Capability Framework Events (Stream 10 — desktop-agent flagship)

Added 2026-04-26 as part of the desktop-agent capability framework generalisation. Audit-service auto-subscribes to all 12 events.

Event pattern	Publisher	Consumers	Purpose
agent.capability.proposed	claw-agent-service	audit	Capability invocation drafted; risk + class + target visible
agent.capability.policy_matched	claw-agent-service	audit	Risk service decided ALLOW / DENY / AUTO_APPROVE
agent.capability.auto_approved	claw-agent-service	audit, capability-runner	Skipped human approval per policy
agent.capability.approved	claw-agent-service	audit, capability-runner	Human approved
agent.capability.rejected	claw-agent-service	audit	Human rejected with reason
agent.capability.executing	claw-agent-service	audit	CLI started executing
agent.capability.executed	claw-agent-service	audit	Completed successfully with result
agent.capability.failed	claw-agent-service	audit	Execution errored
agent.capability.cancelled	claw-agent-service	audit	User cancelled mid-execute
agent.capability.expired	claw-agent-service	audit	PENDING beyond expiresAt
agent.capability.rolled_back	claw-agent-service	audit	Rollback executed (full or partial)
agent.capability.denied	claw-agent-service	audit	Risk service blocked at draft time

Payload types: packages/shared-types/src/events/capability-events.types.ts.