Events

Snippbot uses an internal async publish/subscribe event bus (EventBus) that decouples subsystems. Events flow from producers (executor, scheduler, channels) to consumers (hooks engine, monitoring, SSE streams).

Event Bus vs Intent Dispatcher

The Event Bus is orthogonal to the Unified Intent Dispatcher — they solve different problems and often coexist for the same producer.

Aspect	Event Bus	Intent Dispatcher
Purpose	Broadcast internal state changes	Route outbound intents to executors
Producers	Task executor, scheduler engine, channel adapter, security system	Channel webhooks, scheduler delivery, hooks, autonomy followups, insight generator, device events, API/CLI calls
Consumers	Hook engine, SSE streams, monitor, UI live updates	Executors (TurnOrchestrator, channel_adapter.dispatch, DeviceToolRouter, InsightQueue, schedule_followup, HookEngine, push)
Subscription	bus.subscribe("task.*", handler)	None — single direct call: dispatcher.dispatch(envelope)
Policy gate	None	ProactivityManager (Tier 2 — quiet hours, daily caps, urgency)
LLM involvement	Never	Tier 3 classifier when hints are ambiguous
Persistence	Transient (consumer-level only)	JSONL decision log at ~/.snippbot/dispatch/
Source	event_bus.py	dispatch/

A single scheduler tick typically does both: it fires job_run.completed on the Event Bus (so the monitor and hooks can react to the state change) and emits an IntentEnvelope(source=SCHEDULER, kind=FOLLOWUP) to the dispatcher (so the result is routed to the right room or channel with the right policy gating). The two systems are complementary, not redundant.

Architecture

Producers                    EventBus                   Consumers
─────────                    ────────                   ─────────
Task Executor     ──────►   publish(event)   ──────►   Hook Engine
Scheduler Engine  ──────►                   ──────►   SSE Streams
Channel Adapter   ──────►                   ──────►   Monitor / Analytics
Sub-agent Manager ──────►                   ──────►   UI Live Updates
Security System   ──────►                   ──────►   Webhook Deliveries

The event bus is an in-process async queue. Events are not persisted by default, but the hooks engine and monitor subsystem record relevant events to their respective databases.

Subscribing to events

In Python code (for built-in subsystems):

from snippbot_core.event_bus import get_event_bus

bus = get_event_bus()

def my_handler(event: dict) -> None:
    print(f"Event: {event['type']}, data: {event['data']}")

bus.subscribe("task.completed", my_handler)
bus.subscribe("*", my_handler)  # Subscribe to ALL events

Unsubscribe:

bus.unsubscribe("task.completed", my_handler)

Event format

All events share this structure:

{
  "type": "task.completed",
  "data": {...},
  "timestamp": "2026-03-02T14:00:00Z",
  "source": "executor"
}

Event catalog

Task events

Event	Fired when	Key data fields
task.queued	Task added to queue	task_id, project_id
task.started	Task begins execution	task_id, agent_id
task.completed	Task succeeds	task_id, result, duration_seconds
task.failed	Task fails	task_id, error, failure_class
task.retrying	Task retrying after failure	task_id, attempt
task.skipped	Task skipped	task_id

Approval events

Event	Fired when	Key data fields
approval.requested	Agent requires human approval	task_id, action, reason
approval.granted	Human approves	task_id, approver
approval.denied	Human rejects	task_id, approver, reason

Project events

Event	Fired when	Key data fields
project.created	New project created	project_id, name, goal
project.updated	Status or config changes	project_id, old_status, new_status
project.completed	All tasks done	project_id, duration_seconds
project.failed	Project fails	project_id, error

Scheduler events

Event	Fired when	Key data fields
scheduler.started	Scheduler engine starts	tick_interval_seconds
scheduler.stopped	Scheduler engine stops	—
scheduler.tick	Scheduler checks jobs	jobs_checked
scheduler.error	Scheduler encounters error	error
job_run.created	Job run is created	job_id, run_id
job_run.started	Job run begins execution	job_id, run_id
job_run.completed	Job run finishes successfully	job_id, run_id, output, duration_seconds
job_run.failed	Job run fails	job_id, run_id, error, attempt
job_run.cancelled	Job run is cancelled	job_id, run_id
job_run.skipped	Run skipped (condition unmet)	job_id, run_id
job_run.retrying	Job run retrying after failure	job_id, run_id, attempt
job_run.delivered	Output delivered to channel	job_id, run_id, channel

Workflow events

Event	Fired when	Key data fields
workflow.created	New workflow	workflow_id, name
workflow.run.started	Run begins	workflow_id, run_id
workflow.run.completed	All steps done	run_id, duration_seconds
workflow.run.failed	Run fails	run_id, error
workflow.run.cancelled	Run cancelled	run_id
workflow.step.started	Step begins	run_id, step_id
workflow.step.completed	Step finishes	run_id, step_id, output
workflow.step.failed	Step fails	run_id, step_id, error
workflow.step.skipped	Step skipped	run_id, step_id

Channel events

Event	Fired when	Key data fields
channel.message.received	Inbound message	platform, channel_id, user_id, text
channel.message.sent	Outbound message	platform, channel_id, text
channel.message.failed	Outbound delivery failed	platform, channel_id, error
channel.adapter.started	Adapter comes online	platforms
channel.adapter.stopped	Adapter goes offline	reason

Sub-agent events

Event	Fired when	Key data fields
subagent.spawned	Sub-agent created	subagent_id, parent_id, role
subagent.started	Sub-agent begins work	subagent_id
subagent.completed	Sub-agent finishes	subagent_id, result
subagent.failed	Sub-agent fails	subagent_id, error
subagent.cancelled	Sub-agent cancelled	subagent_id
subagent.approval_requested	Approval gate triggered	subagent_id, action
subagent.approval_granted	Approval granted	subagent_id
subagent.approval_denied	Approval denied	subagent_id, reason

Hook events

Event	Fired when	Key data fields
hook.triggered	Hook fires	hook_id, event_type
hook.completed	Hook execution succeeds	hook_id, duration_ms
hook.failed	Hook execution fails	hook_id, error
hook.timeout	Hook exceeds time limit	hook_id, timeout_ms
hook.skipped	Hook condition not met	hook_id, reason

System events

Event	Fired when	Key data fields
system.startup	Daemon starts	version, config
system.shutdown	Daemon stops	reason
system.error	Unhandled exception	error, traceback

Wildcard subscriptions

Subscribe with "*" to receive all events. The hook engine uses this to check each event against registered hooks:

bus.subscribe("*", hook_engine.on_event)

Prefix matching is also supported:

bus.subscribe("task.*", handler)      # All task events
bus.subscribe("scheduler.*", handler) # All scheduler events

Event persistence

Events are transient by default (in-memory only). Persistence happens at the consumer level:

• Hook engine: Stores delivery history in hooks.db
• Monitor: Records events to activity log in main.db
• Memory system: Persists agent conversation episodes to memory_{id}.db

Note

The event bus is not a message queue — if no subscriber is listening when an event fires, it’s dropped. Hooks and monitoring register on startup to ensure events are captured.