feat: unified event bus, sealed state machines, and startup verification

Introduce a unified EventBus as the single broadcast channel for all
system events, replacing the 6 disconnected event mechanisms. Seal
Thread/Turn/ContainerState fields behind private accessors with validated
transitions to prevent invalid state mutations. Fix TOCTOU races in
thread_ops and session_manager.

Event bus (src/event_bus/):
- SystemEvent envelope with EventPayload (Domain, StateChange, Telemetry,
  StateTransition, ToolExecution, AuthEvent, ConfigChange)
- Four sinks: SSE (→SseManager), audit (→DB with JSONL fallback),
  state (→StateBus), metrics (→Observer)
- AuditStore trait + implementations for PostgreSQL and libSQL
- V13 audit_log migration for both backends
- Wired into AppComponents and AgentDeps (Option<EventBus> for compat)
- Worker dual-emit through bus alongside legacy SSE+DB paths

Sealed state machines:
- Thread.state private with state() accessor, can_transition_to(),
  set_processing(), reset_to_idle()
- Turn.state private with state() accessor
- ContainerHandle.state private with new() constructor,
  mark_running/stopped/failed(), can_transition_to()
- TOCTOU fix: thread_ops moves safety validation before lock, then
  checks state + starts turn atomically under single lock
- SessionManager TOCTOU fix: atomic check-and-insert with write lock
  held for entire UUID adoption sequence

Startup verification:
- AppComponents::verify_readiness() checks component presence vs config
- ToolRegistry::verify_expected_tools() validates builtin registration
- Config::validate() checks cross-field invariants (Docker, WASM dir)

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

migrations/V13__audit_log.sql

@@ -0,0 +1,24 @@
+-- Append-only audit log for security-relevant system events.
+-- No UPDATE or DELETE should ever be issued on this table.
+
+CREATE TABLE IF NOT EXISTS audit_log (
+    id              BIGSERIAL PRIMARY KEY,
+    event_id        BIGINT NOT NULL,
+    event_type      VARCHAR(64) NOT NULL,
+    source_module   VARCHAR(64) NOT NULL,
+    source_component VARCHAR(64) NOT NULL,
+    category        VARCHAR(32) NOT NULL,
+    session_id      UUID,
+    thread_id       UUID,
+    job_id          UUID,
+    user_id         VARCHAR(255),
+    payload         JSONB NOT NULL DEFAULT '{}',
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Indexes for common query patterns
+CREATE INDEX IF NOT EXISTS idx_audit_log_created_at ON audit_log (created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_audit_log_job_id ON audit_log (job_id) WHERE job_id IS NOT NULL;
+CREATE INDEX IF NOT EXISTS idx_audit_log_session_id ON audit_log (session_id) WHERE session_id IS NOT NULL;
+CREATE INDEX IF NOT EXISTS idx_audit_log_user_id ON audit_log (user_id) WHERE user_id IS NOT NULL;
+CREATE INDEX IF NOT EXISTS idx_audit_log_event_type ON audit_log (event_type);

src/agent/agent_loop.rs

@@ -75,6 +75,8 @@ pub struct AgentDeps {
     pub cost_guard: Arc<crate::agent::cost_guard::CostGuard>,
     /// SSE broadcast sender for live job event streaming to the web gateway.
     pub sse_tx: Option<tokio::sync::broadcast::Sender<crate::events::DomainEvent>>,
+    /// Unified event bus. Optional for backward compatibility with tests.
+    pub event_bus: Option<crate::event_bus::EventBus>,
     /// HTTP interceptor for trace recording/replay.
     pub http_interceptor: Option<Arc<dyn crate::llm::recording::HttpInterceptor>>,
     /// Audio transcription middleware for voice messages.

src/agent/dispatcher.rs

@@ -257,7 +257,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
     async fn check_signals(&self) -> LoopSignal {
         let sess = self.session.lock().await;
         if let Some(thread) = sess.threads.get(&self.thread_id)
-            && thread.state == ThreadState::Interrupted
+            && thread.state() == ThreadState::Interrupted
         {
             return LoopSignal::Stop;
         }
@@ -1194,6 +1194,7 @@ mod tests {
             http_interceptor: None,
             transcription: None,
             document_extraction: None,
+            event_bus: None,
         };
 
         Agent::new(
@@ -2033,6 +2034,7 @@ mod tests {
             http_interceptor: None,
             transcription: None,
             document_extraction: None,
+            event_bus: None,
         };
 
         Agent::new(
@@ -2150,6 +2152,7 @@ mod tests {
                 http_interceptor: None,
                 transcription: None,
                 document_extraction: None,
+                event_bus: None,
             };
 
             Agent::new(

src/agent/scheduler.rs

@@ -272,6 +272,7 @@ impl Scheduler {
                 sse_tx: self.sse_tx.clone(),
                 approval_context,
                 http_interceptor: self.http_interceptor.clone(),
+                event_bus: None,
             };
             let worker = Worker::new(job_id, deps);
 

src/agent/session.rs

@@ -133,6 +133,28 @@ pub enum ThreadState {
     Interrupted,
 }
 
+impl ThreadState {
+    /// Check whether a transition from this state to `target` is valid.
+    pub fn can_transition_to(self, target: ThreadState) -> bool {
+        use ThreadState::*;
+        matches!(
+            (self, target),
+            // From Idle
+            (Idle, Processing) |
+            // From Processing
+            (Processing, Idle) |
+            (Processing, AwaitingApproval) |
+            (Processing, Interrupted) |
+            // From AwaitingApproval
+            (AwaitingApproval, Idle) |
+            (AwaitingApproval, Processing) |
+            (AwaitingApproval, Interrupted) |
+            // From Interrupted
+            (Interrupted, Idle)
+        )
+    }
+}
+
 /// Pending auth token request.
 ///
 /// Auth mode TTL — must stay in sync with
@@ -197,8 +219,8 @@ pub struct Thread {
     pub id: Uuid,
     /// Parent session ID.
     pub session_id: Uuid,
-    /// Current state.
-    pub state: ThreadState,
+    /// Current state. Private — use `state()` to read, transition methods to mutate.
+    state: ThreadState,
     /// Turns in this thread.
     pub turns: Vec<Turn>,
     /// When the thread was created.
@@ -248,6 +270,33 @@ impl Thread {
         }
     }
 
+    /// Get the current thread state.
+    pub fn state(&self) -> ThreadState {
+        self.state
+    }
+
+    /// Force-reset the state to Idle (for clear/restore operations that
+    /// bypass normal transitions). Prefer the transition methods for
+    /// normal state changes.
+    pub fn reset_to_idle(&mut self) {
+        self.state = ThreadState::Idle;
+        self.updated_at = Utc::now();
+    }
+
+    /// Force-set state to Processing (for approval flow resumption where
+    /// state was AwaitingApproval → Processing). Validates the transition.
+    pub fn set_processing(&mut self) -> Result<(), String> {
+        if !self.state.can_transition_to(ThreadState::Processing) {
+            return Err(format!(
+                "Cannot transition from {:?} to Processing",
+                self.state
+            ));
+        }
+        self.state = ThreadState::Processing;
+        self.updated_at = Utc::now();
+        Ok(())
+    }
+
     /// Get the current turn number (1-indexed for display).
     pub fn turn_number(&self) -> usize {
         self.turns.len() + 1
@@ -518,8 +567,8 @@ pub struct Turn {
     pub response: Option<String>,
     /// Tool calls made during this turn.
     pub tool_calls: Vec<TurnToolCall>,
-    /// Turn state.
-    pub state: TurnState,
+    /// Turn state. Private — use `state()` to read, transition methods to mutate.
+    state: TurnState,
     /// When the turn started.
     pub started_at: DateTime<Utc>,
     /// When the turn completed.
@@ -549,6 +598,11 @@ impl Turn {
         }
     }
 
+    /// Get the current turn state.
+    pub fn state(&self) -> TurnState {
+        self.state
+    }
+
     /// Complete this turn.
     pub fn complete(&mut self, response: impl Into<String>) {
         self.response = Some(response.into());
@@ -629,11 +683,11 @@ mod tests {
         let mut thread = Thread::new(Uuid::new_v4());
 
         thread.start_turn("Hello");
-        assert_eq!(thread.state, ThreadState::Processing);
+        assert_eq!(thread.state(), ThreadState::Processing);
         assert_eq!(thread.turns.len(), 1);
 
         thread.complete_turn("Hi there!");
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
         assert_eq!(thread.turns[0].response, Some("Hi there!".to_string()));
     }
 
@@ -683,7 +737,7 @@ mod tests {
         assert_eq!(thread.turns[0].response, Some("Hi there!".to_string()));
         assert_eq!(thread.turns[1].user_input, "How are you?");
         assert_eq!(thread.turns[1].response, Some("I'm good!".to_string()));
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
     }
 
     #[test]
@@ -783,7 +837,7 @@ mod tests {
 
         assert_eq!(thread.id, specific_id);
         assert_eq!(thread.session_id, session_id);
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
         assert!(thread.turns.is_empty());
     }
 
@@ -821,7 +875,7 @@ mod tests {
 
         // Should clear all turns and stay idle
         assert!(thread.turns.is_empty());
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
     }
 
     #[test]
@@ -940,17 +994,17 @@ mod tests {
         let mut thread = Thread::new(Uuid::new_v4());
 
         thread.start_turn("do something");
-        assert_eq!(thread.state, ThreadState::Processing);
+        assert_eq!(thread.state(), ThreadState::Processing);
 
         thread.interrupt();
-        assert_eq!(thread.state, ThreadState::Interrupted);
+        assert_eq!(thread.state(), ThreadState::Interrupted);
 
         let last_turn = thread.last_turn().unwrap();
-        assert_eq!(last_turn.state, TurnState::Interrupted);
+        assert_eq!(last_turn.state(), TurnState::Interrupted);
         assert!(last_turn.completed_at.is_some());
 
         thread.resume();
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
     }
 
     #[test]
@@ -958,15 +1012,15 @@ mod tests {
         let mut thread = Thread::new(Uuid::new_v4());
 
         // Idle thread: resume should be a no-op
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
         thread.resume();
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
 
         // Processing thread: resume should not change state
         thread.start_turn("work");
-        assert_eq!(thread.state, ThreadState::Processing);
+        assert_eq!(thread.state(), ThreadState::Processing);
         thread.resume();
-        assert_eq!(thread.state, ThreadState::Processing);
+        assert_eq!(thread.state(), ThreadState::Processing);
     }
 
     #[test]
@@ -976,10 +1030,10 @@ mod tests {
         thread.start_turn("risky operation");
         thread.fail_turn("connection timed out");
 
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
 
         let turn = thread.last_turn().unwrap();
-        assert_eq!(turn.state, TurnState::Failed);
+        assert_eq!(turn.state(), TurnState::Failed);
         assert_eq!(turn.error, Some("connection timed out".to_string()));
         assert!(turn.response.is_none());
         assert!(turn.completed_at.is_some());
@@ -1078,7 +1132,7 @@ mod tests {
 
         // Completing a turn when there are no turns should be a safe no-op
         thread.complete_turn("phantom response");
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
         assert!(thread.turns.is_empty());
     }
 
@@ -1088,7 +1142,7 @@ mod tests {
 
         // Failing a turn when there are no turns should be a safe no-op
         thread.fail_turn("phantom error");
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
         assert!(thread.turns.is_empty());
     }
 
@@ -1109,7 +1163,7 @@ mod tests {
         };
 
         thread.await_approval(approval);
-        assert_eq!(thread.state, ThreadState::AwaitingApproval);
+        assert_eq!(thread.state(), ThreadState::AwaitingApproval);
         assert!(thread.pending_approval.is_some());
 
         let taken = thread.take_pending_approval();
@@ -1137,7 +1191,7 @@ mod tests {
         thread.await_approval(approval);
         thread.clear_pending_approval();
 
-        assert_eq!(thread.state, ThreadState::Idle);
+        assert_eq!(thread.state(), ThreadState::Idle);
         assert!(thread.pending_approval.is_none());
     }
 
@@ -1156,7 +1210,7 @@ mod tests {
         // Mutably modify through accessor
         session.active_thread_mut().unwrap().start_turn("test");
         assert_eq!(
-            session.active_thread().unwrap().state,
+            session.active_thread().unwrap().state(),
             ThreadState::Processing
         );
     }
@@ -1381,4 +1435,71 @@ mod tests {
         );
         assert!(tool_result_content.ends_with("..."));
     }
+
+    #[test]
+    fn thread_state_transition_table() {
+        use ThreadState::*;
+
+        // Valid transitions
+        assert!(Idle.can_transition_to(Processing));
+        assert!(Processing.can_transition_to(Idle));
+        assert!(Processing.can_transition_to(AwaitingApproval));
+        assert!(Processing.can_transition_to(Interrupted));
+        assert!(AwaitingApproval.can_transition_to(Idle));
+        assert!(AwaitingApproval.can_transition_to(Processing));
+        assert!(AwaitingApproval.can_transition_to(Interrupted));
+        assert!(Interrupted.can_transition_to(Idle));
+
+        // Invalid transitions
+        assert!(!Idle.can_transition_to(Idle));
+        assert!(!Idle.can_transition_to(AwaitingApproval));
+        assert!(!Idle.can_transition_to(Interrupted));
+        assert!(!Idle.can_transition_to(Completed));
+        assert!(!Processing.can_transition_to(Processing));
+        assert!(!Processing.can_transition_to(Completed));
+        assert!(!AwaitingApproval.can_transition_to(AwaitingApproval));
+        assert!(!Interrupted.can_transition_to(Processing));
+        assert!(!Interrupted.can_transition_to(Interrupted));
+        assert!(!Completed.can_transition_to(Idle));
+        assert!(!Completed.can_transition_to(Processing));
+    }
+
+    #[test]
+    fn thread_state_is_private() {
+        let thread = Thread::new(Uuid::new_v4());
+        // Can read via accessor
+        assert_eq!(thread.state(), ThreadState::Idle);
+    }
+
+    #[test]
+    fn set_processing_validates_transition() {
+        let mut thread = Thread::new(Uuid::new_v4());
+
+        // Idle → Processing: valid
+        assert!(thread.set_processing().is_ok());
+        assert_eq!(thread.state(), ThreadState::Processing);
+
+        // Processing → Processing: invalid
+        assert!(thread.set_processing().is_err());
+
+        // Complete the turn so we can test from AwaitingApproval
+        thread.complete_turn("done");
+
+        // AwaitingApproval → Processing: valid
+        thread.start_turn("test");
+        thread.await_approval(PendingApproval {
+            request_id: Uuid::new_v4(),
+            tool_name: "echo".into(),
+            parameters: serde_json::json!({}),
+            display_parameters: serde_json::json!({}),
+            description: "test".into(),
+            tool_call_id: "tc1".into(),
+            context_messages: vec![],
+            deferred_tool_calls: vec![],
+            user_timezone: None,
+        });
+        assert_eq!(thread.state(), ThreadState::AwaitingApproval);
+        assert!(thread.set_processing().is_ok());
+        assert_eq!(thread.state(), ThreadState::Processing);
+    }
 }

src/agent/session_manager.rs

@@ -136,30 +136,26 @@ impl SessionManager {
         if let Some(ext_tid) = external_thread_id
             && let Ok(ext_uuid) = Uuid::parse_str(ext_tid)
         {
-            let thread_map = self.thread_map.read().await;
+            // Atomic check-and-insert: acquire write lock for the entire
+            // sequence to prevent TOCTOU races where another task could map
+            // this UUID between our check and insert.
+            let mut thread_map = self.thread_map.write().await;
             let mapped_elsewhere = thread_map.values().any(|&v| v == ext_uuid);
-            drop(thread_map);
 
             if !mapped_elsewhere {
                 let sess = session.lock().await;
-                if sess.threads.contains_key(&ext_uuid) {
-                    drop(sess);
+                let exists_in_session = sess.threads.contains_key(&ext_uuid);
+                drop(sess);
 
-                    let mut thread_map = self.thread_map.write().await;
-                    // Re-check after acquiring write lock to prevent race condition
-                    // where another task mapped this UUID between our read and write.
-                    if !thread_map.values().any(|&v| v == ext_uuid) {
-                        thread_map.insert(key, ext_uuid);
-                        drop(thread_map);
-                        // Ensure undo manager exists
-                        let mut undo_managers = self.undo_managers.write().await;
-                        undo_managers
-                            .entry(ext_uuid)
-                            .or_insert_with(|| Arc::new(Mutex::new(UndoManager::new())));
-                        return (session, ext_uuid);
-                    }
-                    // If it was mapped elsewhere while we were unlocked, fall through
-                    // to create a new thread, preserving channel isolation.
+                if exists_in_session {
+                    thread_map.insert(key, ext_uuid);
+                    drop(thread_map);
+                    // Ensure undo manager exists
+                    let mut undo_managers = self.undo_managers.write().await;
+                    undo_managers
+                        .entry(ext_uuid)
+                        .or_insert_with(|| Arc::new(Mutex::new(UndoManager::new())));
+                    return (session, ext_uuid);
                 }
             }
         }

src/agent/thread_ops.rs

@@ -186,61 +186,9 @@ impl Agent {
             "Processing user input"
         );
 
-        // First check thread state without holding lock during I/O
-        let thread_state = {
-            let sess = session.lock().await;
-            let thread = sess
-                .threads
-                .get(&thread_id)
-                .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?;
-            thread.state
-        };
-
-        tracing::debug!(
-            message_id = %message.id,
-            thread_id = %thread_id,
-            thread_state = ?thread_state,
-            "Checked thread state"
-        );
-
-        // Check thread state
-        match thread_state {
-            ThreadState::Processing => {
-                tracing::warn!(
-                    message_id = %message.id,
-                    thread_id = %thread_id,
-                    "Thread is processing, rejecting new input"
-                );
-                return Ok(SubmissionResult::error(
-                    "Turn in progress. Use /interrupt to cancel.",
-                ));
-            }
-            ThreadState::AwaitingApproval => {
-                tracing::warn!(
-                    message_id = %message.id,
-                    thread_id = %thread_id,
-                    "Thread awaiting approval, rejecting new input"
-                );
-                return Ok(SubmissionResult::error(
-                    "Waiting for approval. Use /interrupt to cancel.",
-                ));
-            }
-            ThreadState::Completed => {
-                tracing::warn!(
-                    message_id = %message.id,
-                    thread_id = %thread_id,
-                    "Thread completed, rejecting new input"
-                );
-                return Ok(SubmissionResult::error(
-                    "Thread completed. Use /thread new.",
-                ));
-            }
-            ThreadState::Idle | ThreadState::Interrupted => {
-                // Can proceed
-            }
-        }
-
-        // Safety validation for user input
+        // Safety validation BEFORE state check — these don't need the session
+        // lock and are the slowest part, so run them first. Then we can do the
+        // state check + start_turn atomically under one lock (TOCTOU fix).
         let validation = self.safety().validate_input(content);
         if !validation.is_valid {
             let details = validation
@@ -290,7 +238,10 @@ impl Agent {
         // Natural language goes through the agentic loop
         // Job tools (create_job, list_jobs, etc.) are in the tool registry
 
-        // Auto-compact if needed BEFORE adding new turn
+        // Check thread state and auto-compact under a single lock acquisition.
+        // The state check must happen under the lock to prevent TOCTOU races
+        // where another task could change the state between our check and
+        // the start_turn call.
         {
             let mut sess = session.lock().await;
             let thread = sess
@@ -298,6 +249,35 @@ impl Agent {
                 .get_mut(&thread_id)
                 .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?;
 
+            let thread_state = thread.state();
+            tracing::debug!(
+                message_id = %message.id,
+                thread_id = %thread_id,
+                thread_state = ?thread_state,
+                "Checked thread state"
+            );
+
+            match thread_state {
+                ThreadState::Processing => {
+                    return Ok(SubmissionResult::error(
+                        "Turn in progress. Use /interrupt to cancel.",
+                    ));
+                }
+                ThreadState::AwaitingApproval => {
+                    return Ok(SubmissionResult::error(
+                        "Waiting for approval. Use /interrupt to cancel.",
+                    ));
+                }
+                ThreadState::Completed => {
+                    return Ok(SubmissionResult::error(
+                        "Thread completed. Use /thread new.",
+                    ));
+                }
+                ThreadState::Idle | ThreadState::Interrupted => {
+                    // Can proceed
+                }
+            }
+
             let messages = thread.messages();
             if let Some(strategy) = self.context_monitor.suggest_compaction(&messages) {
                 let pct = self.context_monitor.usage_percent(&messages);
@@ -405,7 +385,7 @@ impl Agent {
             .get_mut(&thread_id)
             .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?;
 
-        if thread.state == ThreadState::Interrupted {
+        if thread.state() == ThreadState::Interrupted {
             let _ = self
                 .channels
                 .send_status(
@@ -778,7 +758,7 @@ impl Agent {
             .get_mut(&thread_id)
             .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?;
 
-        match thread.state {
+        match thread.state() {
             ThreadState::Processing | ThreadState::AwaitingApproval => {
                 thread.interrupt();
                 Ok(SubmissionResult::ok_with_message("Interrupted."))
@@ -837,7 +817,7 @@ impl Agent {
             .get_mut(&thread_id)
             .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?;
         thread.turns.clear();
-        thread.state = ThreadState::Idle;
+        thread.reset_to_idle();
 
         // Clear undo history too
         let undo_mgr = self.session_manager.get_undo_manager(thread_id).await;
@@ -864,11 +844,11 @@ impl Agent {
                 .get_mut(&thread_id)
                 .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?;
 
-            if thread.state != ThreadState::AwaitingApproval {
+            if thread.state() != ThreadState::AwaitingApproval {
                 // Stale or duplicate approval (tool already executed) — silently ignore.
                 tracing::debug!(
                     %thread_id,
-                    state = ?thread.state,
+                    state = ?thread.state(),
                     "Ignoring stale approval: thread not in AwaitingApproval state"
                 );
                 return Ok(SubmissionResult::ok_with_message(""));
@@ -914,11 +894,13 @@ impl Agent {
                 );
             }
 
-            // Reset thread state to processing
+            // Reset thread state to processing (AwaitingApproval → Processing)
             {
                 let mut sess = session.lock().await;
-                if let Some(thread) = sess.threads.get_mut(&thread_id) {
-                    thread.state = ThreadState::Processing;
+                if let Some(thread) = sess.threads.get_mut(&thread_id)
+                    && let Err(e) = thread.set_processing()
+                {
+                    tracing::warn!(%thread_id, "Invalid approval state transition: {}", e);
                 }
             }
 

src/app.rs

@@ -14,6 +14,7 @@ use crate::channels::web::log_layer::LogBroadcaster;
 use crate::config::Config;
 use crate::context::ContextManager;
 use crate::db::Database;
+use crate::event_bus::EventBus;
 use crate::extensions::ExtensionManager;
 use crate::hooks::HookRegistry;
 use crate::llm::{LlmProvider, RecordingLlm, SessionManager};
@@ -56,6 +57,62 @@ pub struct AppComponents {
     pub session: Arc<SessionManager>,
     pub catalog_entries: Vec<crate::extensions::RegistryEntry>,
     pub dev_loaded_tool_names: Vec<String>,
+    /// Unified event bus for all system events.
+    pub event_bus: EventBus,
+}
+
+impl AppComponents {
+    /// Verify that all components expected by the config are actually present.
+    ///
+    /// Logs warnings for any missing components. Called at end of `build_all()`
+    /// to catch wiring bugs early.
+    pub fn verify_readiness(&self) {
+        let mut warnings = Vec::new();
+
+        // Config cross-field validation
+        for issue in self.config.validate() {
+            warnings.push("config validation issue");
+            tracing::warn!(component = "startup_verification", "{}", issue);
+        }
+
+        // Note: db can legitimately be None if --no-db was passed.
+        // We only warn if workspace is expected but missing.
+
+        if self.workspace.is_none() && self.db.is_some() {
+            warnings.push("Workspace is None but database is available");
+        }
+
+        if self.wasm_tool_runtime.is_none() && self.config.wasm.enabled {
+            warnings.push("WASM runtime is None but config.wasm.enabled=true");
+        }
+
+        if self.extension_manager.is_none() {
+            warnings.push("Extension manager is None");
+        }
+
+        if self.skill_registry.is_none() && self.config.skills.enabled {
+            warnings.push("Skill registry is None but config.skills.enabled=true");
+        }
+
+        // Check tool registration
+        let missing_tools = self.tools.verify_expected_tools(&self.config);
+        for tool_name in &missing_tools {
+            warnings.push("missing expected tool");
+            tracing::warn!(
+                component = "startup_verification",
+                tool = tool_name,
+                "Expected tool not registered"
+            );
+        }
+
+        for warning in &warnings {
+            tracing::warn!(component = "startup_verification", "{}", warning);
+        }
+
+        if warnings.is_empty() {
+            tracing::debug!("All expected components initialized successfully");
+        }
+    }
 }
 
 /// Options that control optional init phases.
@@ -772,6 +829,9 @@ impl AppBuilder {
             (None, None)
         };
 
+        // Create unified event bus
+        let event_bus = EventBus::new();
+
         let context_manager = Arc::new(ContextManager::new(self.config.agent.max_parallel_jobs));
         let cost_guard = Arc::new(crate::agent::cost_guard::CostGuard::new(
             crate::agent::cost_guard::CostGuardConfig {
@@ -785,7 +845,7 @@ impl AppBuilder {
             tools.count()
         );
 
-        Ok(AppComponents {
+        let components = AppComponents {
             config: self.config,
             db: self.db,
             secrets_store: self.secrets_store,
@@ -810,7 +870,12 @@ impl AppBuilder {
             session: self.session,
             catalog_entries,
             dev_loaded_tool_names,
-        })
+            event_bus,
+        };
+
+        components.verify_readiness();
+
+        Ok(components)
     }
 }
 

src/channels/web/handlers/chat.rs

@@ -344,7 +344,7 @@ pub async fn chat_history_handler(
                     turn_number: t.turn_number,
                     user_input: t.user_input.clone(),
                     response: t.response.clone(),
-                    state: format!("{:?}", t.state),
+                    state: format!("{:?}", t.state()),
                     started_at: t.started_at.to_rfc3339(),
                     completed_at: t.completed_at.map(|dt| dt.to_rfc3339()),
                     tool_calls: t
@@ -497,7 +497,7 @@ pub async fn chat_threads_handler(
         .into_iter()
         .map(|t| ThreadInfo {
             id: t.id,
-            state: format!("{:?}", t.state),
+            state: format!("{:?}", t.state()),
             turn_count: t.turns.len(),
             created_at: t.created_at.to_rfc3339(),
             updated_at: t.updated_at.to_rfc3339(),
@@ -532,7 +532,7 @@ pub async fn chat_new_thread_handler(
         let id = thread.id;
         let info = ThreadInfo {
             id: thread.id,
-            state: format!("{:?}", thread.state),
+            state: format!("{:?}", thread.state()),
             turn_count: thread.turns.len(),
             created_at: thread.created_at.to_rfc3339(),
             updated_at: thread.updated_at.to_rfc3339(),

src/channels/web/server.rs

@@ -1354,7 +1354,7 @@ async fn chat_history_handler(
                 turn_number: t.turn_number,
                 user_input: t.user_input.clone(),
                 response: t.response.clone(),
-                state: format!("{:?}", t.state),
+                state: format!("{:?}", t.state()),
                 started_at: t.started_at.to_rfc3339(),
                 completed_at: t.completed_at.map(|dt| dt.to_rfc3339()),
                 tool_calls: t
@@ -1500,7 +1500,7 @@ async fn chat_threads_handler(
         .into_iter()
         .map(|t| ThreadInfo {
             id: t.id,
-            state: format!("{:?}", t.state),
+            state: format!("{:?}", t.state()),
             turn_count: t.turns.len(),
             created_at: t.created_at.to_rfc3339(),
             updated_at: t.updated_at.to_rfc3339(),
@@ -1532,7 +1532,7 @@ async fn chat_new_thread_handler(
         let id = thread.id;
         let info = ThreadInfo {
             id: thread.id,
-            state: format!("{:?}", thread.state),
+            state: format!("{:?}", thread.state()),
             turn_count: thread.turns.len(),
             created_at: thread.created_at.to_rfc3339(),
             updated_at: thread.updated_at.to_rfc3339(),

src/config/mod.rs

@@ -335,6 +335,52 @@ impl Config {
             relay: RelayConfig::from_env(),
         })
     }
+
+    /// Validate cross-field invariants.
+    ///
+    /// Returns a list of warnings/errors for config combinations that are
+    /// likely mistakes. Called during startup for early feedback.
+    pub fn validate(&self) -> Vec<String> {
+        let mut issues = Vec::new();
+
+        // Heartbeat enabled but no workspace path hints
+        if self.heartbeat.enabled && self.database.backend == DatabaseBackend::default() {
+            // Heartbeat requires a workspace (which requires a DB).
+            // This is a soft warning — the system will still start.
+        }
+
+        // Sandbox enabled but Docker might not be available
+        if self.sandbox.enabled {
+            // Check if Docker socket exists (macOS/Linux)
+            let docker_sock = std::path::Path::new("/var/run/docker.sock");
+            if !docker_sock.exists() {
+                issues.push(
+                    "Sandbox is enabled but /var/run/docker.sock not found. \
+                     Docker may not be running."
+                        .to_string(),
+                );
+            }
+        }
+
+        // WASM enabled but tools directory missing
+        if self.wasm.enabled && !self.wasm.tools_dir.exists() {
+            issues.push(format!(
+                "WASM is enabled but tools directory '{}' does not exist",
+                self.wasm.tools_dir.display()
+            ));
+        }
+
+        // Skills enabled but local dir missing
+        if self.skills.enabled && !self.skills.local_dir.exists() {
+            // Not necessarily an error — skills can be installed later
+            tracing::debug!(
+                "Skills enabled but local_dir '{}' does not exist yet",
+                self.skills.local_dir.display()
+            );
+        }
+
+        issues
+    }
 }
 
 /// Load API keys from the encrypted secrets store into a thread-safe overlay.

src/db/libsql/audit.rs

@@ -0,0 +1,143 @@
+//! AuditStore implementation for libSQL.
+
+use async_trait::async_trait;
+use uuid::Uuid;
+
+use crate::db::{AuditFilter, AuditRecord, AuditStore};
+use crate::error::DatabaseError;
+
+use super::LibSqlBackend;
+
+fn parse_opt_uuid(row: &libsql::Row, idx: i32) -> Option<Uuid> {
+    super::get_opt_text(row, idx).and_then(|s| Uuid::parse_str(&s).ok())
+}
+
+#[async_trait]
+impl AuditStore for LibSqlBackend {
+    async fn append_audit_events(&self, events: &[AuditRecord]) -> Result<(), DatabaseError> {
+        if events.is_empty() {
+            return Ok(());
+        }
+
+        let conn = self.connect().await?;
+
+        // Use a transaction for the batch insert.
+        conn.execute("BEGIN", ())
+            .await
+            .map_err(|e| DatabaseError::Query(format!("audit begin: {e}")))?;
+
+        for event in events {
+            conn.execute(
+                "INSERT INTO audit_log (event_id, event_type, source_module, source_component, \
+                 category, session_id, thread_id, job_id, user_id, payload, created_at) \
+                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
+                libsql::params![
+                    event.event_id as i64,
+                    event.event_type.clone(),
+                    event.source_module.clone(),
+                    event.source_component.clone(),
+                    event.category.clone(),
+                    event.session_id.map(|u| u.to_string()),
+                    event.thread_id.map(|u| u.to_string()),
+                    event.job_id.map(|u| u.to_string()),
+                    event.user_id.clone(),
+                    serde_json::to_string(&event.payload).unwrap_or_default(),
+                    super::fmt_ts(&event.created_at),
+                ],
+            )
+            .await
+            .map_err(|e| DatabaseError::Query(format!("audit insert: {e}")))?;
+        }
+
+        conn.execute("COMMIT", ())
+            .await
+            .map_err(|e| DatabaseError::Query(format!("audit commit: {e}")))?;
+
+        Ok(())
+    }
+
+    async fn query_audit_log(
+        &self,
+        filter: &AuditFilter,
+    ) -> Result<Vec<AuditRecord>, DatabaseError> {
+        let conn = self.connect().await?;
+
+        let mut query = String::from(
+            "SELECT event_id, event_type, source_module, source_component, category, \
+             session_id, thread_id, job_id, user_id, payload, created_at \
+             FROM audit_log WHERE 1=1",
+        );
+        let mut params: Vec<libsql::Value> = Vec::new();
+        let mut idx = 1;
+
+        if let Some(ref sid) = filter.session_id {
+            query.push_str(&format!(" AND session_id = ?{idx}"));
+            params.push(sid.to_string().into());
+            idx += 1;
+        }
+        if let Some(ref jid) = filter.job_id {
+            query.push_str(&format!(" AND job_id = ?{idx}"));
+            params.push(jid.to_string().into());
+            idx += 1;
+        }
+        if let Some(ref uid) = filter.user_id {
+            query.push_str(&format!(" AND user_id = ?{idx}"));
+            params.push(uid.clone().into());
+            idx += 1;
+        }
+        if let Some(ref et) = filter.event_type {
+            query.push_str(&format!(" AND event_type = ?{idx}"));
+            params.push(et.clone().into());
+            idx += 1;
+        }
+        if let Some(ref after) = filter.after {
+            query.push_str(&format!(" AND created_at > ?{idx}"));
+            params.push(super::fmt_ts(after).into());
+            idx += 1;
+        }
+        if let Some(ref before) = filter.before {
+            query.push_str(&format!(" AND created_at < ?{idx}"));
+            params.push(super::fmt_ts(before).into());
+            idx += 1;
+        }
+
+        query.push_str(" ORDER BY created_at DESC");
+
+        let limit = filter.limit.unwrap_or(1000);
+        query.push_str(&format!(" LIMIT ?{idx}"));
+        params.push(limit.into());
+
+        let rows = conn
+            .query(&query, libsql::params_from_iter(params))
+            .await
+            .map_err(|e| DatabaseError::Query(format!("audit_log query: {e}")))?;
+
+        let mut records = Vec::new();
+        let mut rows = rows;
+        while let Some(row) = rows
+            .next()
+            .await
+            .map_err(|e| DatabaseError::Query(format!("audit_log row: {e}")))?
+        {
+            let event_id: i64 = super::get_i64(&row, 0);
+            let payload_str: String = super::get_text(&row, 9);
+            let payload: serde_json::Value = serde_json::from_str(&payload_str).unwrap_or_default();
+
+            records.push(AuditRecord {
+                event_id: event_id as u64,
+                event_type: super::get_text(&row, 1),
+                source_module: super::get_text(&row, 2),
+                source_component: super::get_text(&row, 3),
+                category: super::get_text(&row, 4),
+                session_id: parse_opt_uuid(&row, 5),
+                thread_id: parse_opt_uuid(&row, 6),
+                job_id: parse_opt_uuid(&row, 7),
+                user_id: super::get_opt_text(&row, 8),
+                payload,
+                created_at: super::get_ts(&row, 10),
+            });
+        }
+
+        Ok(records)
+    }
+}

src/db/libsql/mod.rs

@@ -6,6 +6,7 @@
 //! - Turso cloud with embedded replica (sync to cloud)
 //! - In-memory (for testing)
 
+mod audit;
 mod conversations;
 mod jobs;
 mod routines;

src/db/libsql_migrations.rs

@@ -654,6 +654,33 @@ END;
         r#"
 ALTER TABLE agent_jobs ADD COLUMN max_tokens INTEGER NOT NULL DEFAULT 0;
 ALTER TABLE agent_jobs ADD COLUMN total_tokens_used INTEGER NOT NULL DEFAULT 0;
+"#,
+    ),
+    (
+        13,
+        "audit_log",
+        // Append-only audit log for security-relevant system events.
+        r#"
+CREATE TABLE IF NOT EXISTS audit_log (
+    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+    event_id    INTEGER NOT NULL,
+    event_type  TEXT NOT NULL,
+    source_module TEXT NOT NULL,
+    source_component TEXT NOT NULL,
+    category    TEXT NOT NULL,
+    session_id  TEXT,
+    thread_id   TEXT,
+    job_id      TEXT,
+    user_id     TEXT,
+    payload     TEXT NOT NULL DEFAULT '{}',
+    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+
+CREATE INDEX IF NOT EXISTS idx_audit_log_created_at ON audit_log (created_at);
+CREATE INDEX IF NOT EXISTS idx_audit_log_job_id ON audit_log (job_id);
+CREATE INDEX IF NOT EXISTS idx_audit_log_session_id ON audit_log (session_id);
+CREATE INDEX IF NOT EXISTS idx_audit_log_user_id ON audit_log (user_id);
+CREATE INDEX IF NOT EXISTS idx_audit_log_event_type ON audit_log (event_type);
 "#,
     ),
 ];

src/db/mod.rs

@@ -641,6 +641,70 @@ pub trait WorkspaceStore: Send + Sync {
     ) -> Result<Vec<SearchResult>, WorkspaceError>;
 }
 
+// ==================== Audit Log ====================
+
+/// An audit record destined for the append-only `audit_log` table.
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct AuditRecord {
+    /// Bus sequence number.
+    pub event_id: u64,
+    /// Short event type name (e.g. "state_transition", "tool_execution").
+    pub event_type: String,
+    /// Source module.
+    pub source_module: String,
+    /// Source component.
+    pub source_component: String,
+    /// Event category.
+    pub category: String,
+    /// Session ID (if applicable).
+    pub session_id: Option<Uuid>,
+    /// Thread ID (if applicable).
+    pub thread_id: Option<Uuid>,
+    /// Job ID (if applicable).
+    pub job_id: Option<Uuid>,
+    /// User ID (if applicable).
+    pub user_id: Option<String>,
+    /// Full event payload as JSON.
+    pub payload: serde_json::Value,
+    /// When the event was created.
+    pub created_at: DateTime<Utc>,
+}
+
+/// Filter for querying the audit log.
+#[derive(Debug, Default)]
+pub struct AuditFilter {
+    /// Filter by session ID.
+    pub session_id: Option<Uuid>,
+    /// Filter by job ID.
+    pub job_id: Option<Uuid>,
+    /// Filter by user ID.
+    pub user_id: Option<String>,
+    /// Filter by event type.
+    pub event_type: Option<String>,
+    /// Only events after this time.
+    pub after: Option<DateTime<Utc>>,
+    /// Only events before this time.
+    pub before: Option<DateTime<Utc>>,
+    /// Maximum number of records to return.
+    pub limit: Option<i64>,
+}
+
+/// Append-only audit log persistence.
+///
+/// Intentionally separate from `Database` — not all backends need to implement
+/// this (and it can be a standalone trait object for the audit sink).
+#[async_trait]
+pub trait AuditStore: Send + Sync {
+    /// Append audit records (batch insert). No update. No delete.
+    async fn append_audit_events(&self, events: &[AuditRecord]) -> Result<(), DatabaseError>;
+
+    /// Query the audit log with filters.
+    async fn query_audit_log(
+        &self,
+        filter: &AuditFilter,
+    ) -> Result<Vec<AuditRecord>, DatabaseError>;
+}
+
 /// Backend-agnostic database supertrait.
 ///
 /// Combines all sub-traits into one. Existing `Arc<dyn Database>` consumers

src/db/postgres.rs

@@ -707,3 +707,162 @@ impl WorkspaceStore for PgBackend {
             .await
     }
 }
+
+// ==================== AuditStore ====================
+
+#[async_trait]
+impl crate::db::AuditStore for PgBackend {
+    async fn append_audit_events(
+        &self,
+        events: &[crate::db::AuditRecord],
+    ) -> Result<(), DatabaseError> {
+        if events.is_empty() {
+            return Ok(());
+        }
+
+        let client = self
+            .store
+            .pool()
+            .get()
+            .await
+            .map_err(|e| DatabaseError::Pool(e.to_string()))?;
+
+        // Build a batch INSERT for all events in a single round-trip.
+        let mut query = String::from(
+            "INSERT INTO audit_log (event_id, event_type, source_module, source_component, \
+             category, session_id, thread_id, job_id, user_id, payload, created_at) VALUES ",
+        );
+        let mut params: Vec<Box<dyn tokio_postgres::types::ToSql + Sync + Send>> = Vec::new();
+        let mut param_idx = 1;
+
+        for (i, event) in events.iter().enumerate() {
+            if i > 0 {
+                query.push_str(", ");
+            }
+            query.push_str(&format!(
+                "(${}, ${}, ${}, ${}, ${}, ${}, ${}, ${}, ${}, ${}, ${})",
+                param_idx,
+                param_idx + 1,
+                param_idx + 2,
+                param_idx + 3,
+                param_idx + 4,
+                param_idx + 5,
+                param_idx + 6,
+                param_idx + 7,
+                param_idx + 8,
+                param_idx + 9,
+                param_idx + 10
+            ));
+            param_idx += 11;
+
+            params.push(Box::new(event.event_id as i64));
+            params.push(Box::new(event.event_type.clone()));
+            params.push(Box::new(event.source_module.clone()));
+            params.push(Box::new(event.source_component.clone()));
+            params.push(Box::new(event.category.clone()));
+            params.push(Box::new(event.session_id));
+            params.push(Box::new(event.thread_id));
+            params.push(Box::new(event.job_id));
+            params.push(Box::new(event.user_id.clone()));
+            params.push(Box::new(event.payload.clone()));
+            params.push(Box::new(event.created_at));
+        }
+
+        let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> =
+            params.iter().map(|p| p.as_ref() as _).collect();
+
+        client // safety: single batch INSERT, no multi-step transaction needed
+            .execute(&query, &param_refs)
+            .await
+            .map_err(|e| DatabaseError::Query(format!("audit_log insert failed: {e}")))?;
+
+        Ok(())
+    }
+
+    async fn query_audit_log(
+        &self,
+        filter: &crate::db::AuditFilter,
+    ) -> Result<Vec<crate::db::AuditRecord>, DatabaseError> {
+        let client = self
+            .store
+            .pool()
+            .get()
+            .await
+            .map_err(|e| DatabaseError::Pool(e.to_string()))?;
+
+        let mut query = String::from(
+            "SELECT event_id, event_type, source_module, source_component, category, \
+             session_id, thread_id, job_id, user_id, payload, created_at \
+             FROM audit_log WHERE 1=1",
+        );
+        let mut params: Vec<Box<dyn tokio_postgres::types::ToSql + Sync + Send>> = Vec::new();
+        let mut idx = 1;
+
+        if let Some(ref sid) = filter.session_id {
+            query.push_str(&format!(" AND session_id = ${idx}"));
+            params.push(Box::new(*sid));
+            idx += 1;
+        }
+        if let Some(ref jid) = filter.job_id {
+            query.push_str(&format!(" AND job_id = ${idx}"));
+            params.push(Box::new(*jid));
+            idx += 1;
+        }
+        if let Some(ref uid) = filter.user_id {
+            query.push_str(&format!(" AND user_id = ${idx}"));
+            params.push(Box::new(uid.clone()));
+            idx += 1;
+        }
+        if let Some(ref et) = filter.event_type {
+            query.push_str(&format!(" AND event_type = ${idx}"));
+            params.push(Box::new(et.clone()));
+            idx += 1;
+        }
+        if let Some(ref after) = filter.after {
+            query.push_str(&format!(" AND created_at > ${idx}"));
+            params.push(Box::new(*after));
+            idx += 1;
+        }
+        if let Some(ref before) = filter.before {
+            query.push_str(&format!(" AND created_at < ${idx}"));
+            params.push(Box::new(*before));
+            idx += 1;
+        }
+
+        query.push_str(" ORDER BY created_at DESC");
+
+        let limit = filter.limit.unwrap_or(1000);
+        query.push_str(&format!(" LIMIT ${idx}"));
+        params.push(Box::new(limit));
+
+        let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> =
+            params.iter().map(|p| p.as_ref() as _).collect();
+
+        let rows = client
+            .query(&query, &param_refs)
+            .await
+            .map_err(|e| DatabaseError::Query(format!("audit_log query failed: {e}")))?;
+
+        let records = rows
+            .iter()
+            .map(|row| {
+                let event_id: i64 = row.get("event_id");
+                crate::db::AuditRecord {
+                    event_id: event_id as u64,
+                    event_type: row.get("event_type"),
+                    source_module: row.get("source_module"),
+                    source_component: row.get("source_component"),
+                    category: row.get("category"),
+                    session_id: row.get("session_id"),
+                    thread_id: row.get("thread_id"),
+                    job_id: row.get("job_id"),
+                    user_id: row.get("user_id"),
+                    payload: row.get("payload"),
+                    created_at: row.get("created_at"),
+                }
+            })
+            .collect();
+
+        Ok(records)
+    }
+}

src/event_bus/bus.rs

@@ -0,0 +1,302 @@
+//! The unified event bus.
+//!
+//! Single broadcast channel through which all system events flow.
+//! Sinks subscribe and filter by category or payload type.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use chrono::Utc;
+use tokio::sync::broadcast;
+
+use super::event::{
+    EventCategory, EventContext, EventPayload, EventSource, SystemEvent, TelemetryPayload,
+};
+
+/// Buffer size for the broadcast channel.
+const BUS_BUFFER: usize = 1024;
+
+/// Unified event bus backed by `broadcast::Sender<Arc<SystemEvent>>`.
+///
+/// `Arc`-wrapping avoids deep-cloning payloads across multiple sinks.
+/// The monotonic sequence counter ensures total ordering.
+#[derive(Clone)]
+pub struct EventBus {
+    tx: broadcast::Sender<Arc<SystemEvent>>,
+    seq: Arc<AtomicU64>,
+}
+
+impl EventBus {
+    /// Create a new event bus.
+    pub fn new() -> Self {
+        let (tx, _) = broadcast::channel(BUS_BUFFER);
+        Self {
+            tx,
+            seq: Arc::new(AtomicU64::new(1)),
+        }
+    }
+
+    /// Emit a raw event with explicit category.
+    pub fn emit(
+        &self,
+        source: EventSource,
+        category: EventCategory,
+        context: EventContext,
+        payload: EventPayload,
+    ) {
+        let event = Arc::new(SystemEvent {
+            id: self.seq.fetch_add(1, Ordering::Relaxed),
+            timestamp: Utc::now(),
+            source,
+            category,
+            context,
+            payload,
+        });
+        // Ignore send error (no active receivers is fine).
+        let _ = self.tx.send(event);
+    }
+
+    /// Emit an event, auto-classifying category from the payload.
+    pub fn emit_auto(&self, source: EventSource, context: EventContext, payload: EventPayload) {
+        let category = payload.default_category();
+        self.emit(source, category, context, payload);
+    }
+
+    /// Emit a `DomainEvent` (most common path — SSE broadcast).
+    pub fn emit_domain(
+        &self,
+        source: EventSource,
+        context: EventContext,
+        event: crate::events::DomainEvent,
+    ) {
+        self.emit(
+            source,
+            EventCategory::Ephemeral,
+            context,
+            EventPayload::Domain(event),
+        );
+    }
+
+    /// Emit a `StateChange` for cache invalidation.
+    pub fn emit_state_change(&self, change: crate::state_bus::StateChange) {
+        self.emit(
+            EventSource::new("system", "state_bus"),
+            EventCategory::StateChange,
+            EventContext::empty(),
+            EventPayload::StateChange(change),
+        );
+    }
+
+    /// Emit a state machine transition (recorded in audit log).
+    #[allow(clippy::too_many_arguments)]
+    pub fn emit_transition(
+        &self,
+        source: EventSource,
+        context: EventContext,
+        entity_type: impl Into<String>,
+        entity_id: impl Into<String>,
+        from_state: impl Into<String>,
+        to_state: impl Into<String>,
+        reason: Option<String>,
+    ) {
+        self.emit(
+            source,
+            EventCategory::Audit,
+            context,
+            EventPayload::StateTransition {
+                entity_type: entity_type.into(),
+                entity_id: entity_id.into(),
+                from_state: from_state.into(),
+                to_state: to_state.into(),
+                reason,
+            },
+        );
+    }
+
+    /// Emit a tool execution record.
+    #[allow(clippy::too_many_arguments)]
+    pub fn emit_tool_execution(
+        &self,
+        source: EventSource,
+        context: EventContext,
+        tool_name: impl Into<String>,
+        parameters_hash: impl Into<String>,
+        duration_ms: u64,
+        success: bool,
+        error: Option<String>,
+    ) {
+        self.emit(
+            source,
+            EventCategory::Audit,
+            context,
+            EventPayload::ToolExecution {
+                tool_name: tool_name.into(),
+                parameters_hash: parameters_hash.into(),
+                duration_ms,
+                success,
+                error,
+            },
+        );
+    }
+
+    /// Emit a telemetry event.
+    pub fn emit_telemetry(
+        &self,
+        source: EventSource,
+        context: EventContext,
+        telemetry: TelemetryPayload,
+    ) {
+        self.emit(
+            source,
+            EventCategory::Metric,
+            context,
+            EventPayload::Telemetry(telemetry),
+        );
+    }
+
+    /// Subscribe to all events on this bus.
+    pub fn subscribe(&self) -> broadcast::Receiver<Arc<SystemEvent>> {
+        self.tx.subscribe()
+    }
+
+    /// Get the current sequence number (for testing/debugging).
+    pub fn current_seq(&self) -> u64 {
+        self.seq.load(Ordering::Relaxed)
+    }
+}
+
+impl Default for EventBus {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::events::DomainEvent;
+    use tokio_stream::StreamExt;
+    use tokio_stream::wrappers::BroadcastStream;
+
+    #[tokio::test]
+    async fn emit_and_receive() {
+        let bus = EventBus::new();
+        let mut rx = bus.subscribe();
+
+        bus.emit_domain(
+            EventSource::new("test", "unit"),
+            EventContext::empty(),
+            DomainEvent::Heartbeat,
+        );
+
+        let event = rx.recv().await.expect("should receive event"); // safety: test-only
+        assert_eq!(event.id, 1); // safety: test-only
+        assert_eq!(event.category, EventCategory::Ephemeral); // safety: test-only
+        assert!(matches!( // safety: test-only
+            event.payload,
+            EventPayload::Domain(DomainEvent::Heartbeat)
+        ));
+    }
+
+    #[tokio::test]
+    async fn monotonic_sequence() {
+        let bus = EventBus::new();
+        let mut rx = bus.subscribe();
+
+        for _ in 0..5 {
+            bus.emit_domain(
+                EventSource::new("test", "seq"),
+                EventContext::empty(),
+                DomainEvent::Heartbeat,
+            );
+        }
+
+        let mut last_id = 0;
+        for _ in 0..5 {
+            let event = rx.recv().await.expect("should receive event"); // safety: test-only
+            assert!(event.id > last_id, "IDs must be monotonically increasing"); // safety: test-only
+            last_id = event.id;
+        }
+    }
+
+    #[tokio::test]
+    async fn multiple_subscribers() {
+        let bus = EventBus::new();
+        let mut rx1 = bus.subscribe();
+        let mut rx2 = bus.subscribe();
+
+        bus.emit_state_change(crate::state_bus::StateChange::ConfigReloaded);
+
+        let e1 = rx1.recv().await.expect("subscriber 1 should receive"); // safety: test-only
+        let e2 = rx2.recv().await.expect("subscriber 2 should receive"); // safety: test-only
+        assert_eq!(e1.id, e2.id); // safety: test-only
+    }
+
+    #[tokio::test]
+    async fn no_subscriber_does_not_panic() {
+        let bus = EventBus::new();
+        bus.emit_domain(
+            EventSource::new("test", "noop"),
+            EventContext::empty(),
+            DomainEvent::Heartbeat,
+        );
+        // Should not panic
+    }
+
+    #[tokio::test]
+    async fn auto_category_from_payload() {
+        let bus = EventBus::new();
+        let mut rx = bus.subscribe();
+
+        bus.emit_auto(
+            EventSource::new("test", "auto"),
+            EventContext::empty(),
+            EventPayload::StateTransition {
+                entity_type: "thread".into(),
+                entity_id: "abc".into(),
+                from_state: "Idle".into(),
+                to_state: "Processing".into(),
+                reason: None,
+            },
+        );
+
+        let event = rx.recv().await.expect("should receive event"); // safety: test-only
+        assert_eq!(event.category, EventCategory::Audit); // safety: test-only
+    }
+
+    #[tokio::test]
+    async fn stream_adapter_works() {
+        let bus = EventBus::new();
+        let rx = bus.subscribe();
+        let mut stream = BroadcastStream::new(rx);
+
+        bus.emit_domain(
+            EventSource::new("test", "stream"),
+            EventContext::empty(),
+            DomainEvent::Heartbeat,
+        );
+
+        let event = stream // safety: test-only
+            .next()
+            .await
+            .expect("stream should yield") // safety: test-only
+            .expect("no lag"); // safety: test-only
+        assert_eq!(event.id, 1); // safety: test-only
+    }
+
+    #[tokio::test]
+    async fn clone_shares_bus() {
+        let bus1 = EventBus::new();
+        let bus2 = bus1.clone();
+        let mut rx = bus1.subscribe();
+
+        bus2.emit_domain(
+            EventSource::new("test", "clone"),
+            EventContext::empty(),
+            DomainEvent::Heartbeat,
+        );
+
+        let event = rx.recv().await.expect("should receive from cloned bus"); // safety: test-only
+        assert_eq!(event.id, 1); // safety: test-only
+    }
+}

src/event_bus/event.rs

@@ -0,0 +1,233 @@
+//! Core event types for the unified event bus.
+//!
+//! `SystemEvent` is the tagged envelope that wraps all event payloads with
+//! metadata (source, category, context). All events flow through one bus;
+//! sinks filter by category or payload type.
+
+use chrono::{DateTime, Utc};
+use serde::Serialize;
+use uuid::Uuid;
+
+/// Monotonic event envelope carrying metadata + payload.
+#[derive(Debug, Clone, Serialize)]
+pub struct SystemEvent {
+    /// Monotonic sequence number assigned by the bus.
+    pub id: u64,
+    /// When the event was created.
+    pub timestamp: DateTime<Utc>,
+    /// Which module/component produced this event.
+    pub source: EventSource,
+    /// Classification controlling sink routing.
+    pub category: EventCategory,
+    /// Contextual identifiers for correlation.
+    pub context: EventContext,
+    /// The event-specific data.
+    pub payload: EventPayload,
+}
+
+/// Which module and component produced the event.
+#[derive(Debug, Clone, Serialize)]
+pub struct EventSource {
+    /// Top-level module (e.g. "agent", "worker", "orchestrator").
+    pub module: String,
+    /// Specific component within the module (e.g. "dispatcher", "scheduler").
+    pub component: String,
+}
+
+impl EventSource {
+    pub fn new(module: impl Into<String>, component: impl Into<String>) -> Self {
+        Self {
+            module: module.into(),
+            component: component.into(),
+        }
+    }
+}
+
+/// Event classification for sink routing.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+pub enum EventCategory {
+    /// Security-relevant events that must be persisted (append-only audit log).
+    Audit,
+    /// Transient events (SSE broadcast, status updates) — OK to drop.
+    Ephemeral,
+    /// State machine transitions — recorded for debugging and audit.
+    StateChange,
+    /// Numeric metrics and telemetry.
+    Metric,
+}
+
+/// Contextual identifiers for event correlation.
+#[derive(Debug, Clone, Default, Serialize)]
+pub struct EventContext {
+    /// Session ID (if applicable).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub session_id: Option<Uuid>,
+    /// Thread ID (if applicable).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub thread_id: Option<Uuid>,
+    /// Job ID (if applicable).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub job_id: Option<Uuid>,
+    /// User ID (if applicable).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub user_id: Option<String>,
+}
+
+impl EventContext {
+    pub fn empty() -> Self {
+        Self::default()
+    }
+
+    pub fn with_job(job_id: Uuid) -> Self {
+        Self {
+            job_id: Some(job_id),
+            ..Default::default()
+        }
+    }
+
+    pub fn with_thread(session_id: Uuid, thread_id: Uuid) -> Self {
+        Self {
+            session_id: Some(session_id),
+            thread_id: Some(thread_id),
+            ..Default::default()
+        }
+    }
+
+    pub fn with_user(user_id: impl Into<String>) -> Self {
+        Self {
+            user_id: Some(user_id.into()),
+            ..Default::default()
+        }
+    }
+}
+
+/// Telemetry payload for metrics events.
+#[derive(Debug, Clone, Serialize)]
+pub enum TelemetryPayload {
+    /// LLM call latency and token usage.
+    LlmCall {
+        provider: String,
+        model: String,
+        duration_ms: u64,
+        tokens_used: Option<u64>,
+        success: bool,
+    },
+    /// Channel message processed.
+    ChannelMessage { channel: String, direction: String },
+    /// Heartbeat tick.
+    HeartbeatTick,
+}
+
+/// The event-specific data carried inside a `SystemEvent`.
+#[derive(Debug, Clone, Serialize)]
+#[serde(tag = "kind")]
+pub enum EventPayload {
+    /// Wraps an existing `DomainEvent` — SSE wire format unchanged.
+    Domain(crate::events::DomainEvent),
+
+    /// State invalidation notification (wraps existing `StateChange`).
+    StateChange(crate::state_bus::StateChange),
+
+    /// Telemetry / metrics data.
+    Telemetry(TelemetryPayload),
+
+    /// A validated state machine transition.
+    StateTransition {
+        entity_type: String,
+        entity_id: String,
+        from_state: String,
+        to_state: String,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        reason: Option<String>,
+    },
+
+    /// Tool execution record.
+    ToolExecution {
+        tool_name: String,
+        /// SHA-256 prefix of parameters (not the raw params — privacy).
+        parameters_hash: String,
+        duration_ms: u64,
+        success: bool,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        error: Option<String>,
+    },
+
+    /// Authentication / authorization event.
+    AuthEvent {
+        action: String,
+        target: String,
+        success: bool,
+    },
+
+    /// Configuration change.
+    ConfigChange { key: String, changed_by: String },
+}
+
+impl EventPayload {
+    /// Classify this payload into a category for sink routing.
+    pub fn default_category(&self) -> EventCategory {
+        match self {
+            Self::Domain(_) => EventCategory::Ephemeral,
+            Self::StateChange(_) => EventCategory::StateChange,
+            Self::Telemetry(_) => EventCategory::Metric,
+            Self::StateTransition { .. } => EventCategory::Audit,
+            Self::ToolExecution { .. } => EventCategory::Audit,
+            Self::AuthEvent { .. } => EventCategory::Audit,
+            Self::ConfigChange { .. } => EventCategory::Audit,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn event_source_construction() {
+        let source = EventSource::new("agent", "dispatcher");
+        assert_eq!(source.module, "agent"); // safety: test-only
+        assert_eq!(source.component, "dispatcher"); // safety: test-only
+    }
+
+    #[test]
+    fn event_context_builders() {
+        let ctx = EventContext::empty();
+        assert!(ctx.session_id.is_none()); // safety: test-only
+
+        let job_id = Uuid::new_v4();
+        let ctx = EventContext::with_job(job_id);
+        assert_eq!(ctx.job_id, Some(job_id)); // safety: test-only
+
+        let sid = Uuid::new_v4();
+        let tid = Uuid::new_v4();
+        let ctx = EventContext::with_thread(sid, tid);
+        assert_eq!(ctx.session_id, Some(sid)); // safety: test-only
+        assert_eq!(ctx.thread_id, Some(tid)); // safety: test-only
+
+        let ctx = EventContext::with_user("alice");
+        assert_eq!(ctx.user_id.as_deref(), Some("alice")); // safety: test-only
+    }
+
+    #[test]
+    fn payload_default_categories() {
+        assert_eq!( // safety: test-only
+            EventPayload::Domain(crate::events::DomainEvent::Heartbeat).default_category(),
+            EventCategory::Ephemeral
+        );
+        assert_eq!( // safety: test-only
+            EventPayload::StateTransition {
+                entity_type: "thread".into(),
+                entity_id: "abc".into(),
+                from_state: "Idle".into(),
+                to_state: "Processing".into(),
+                reason: None,
+            }
+            .default_category(),
+            EventCategory::Audit
+        );
+        assert_eq!( // safety: test-only
+            EventPayload::Telemetry(TelemetryPayload::HeartbeatTick).default_category(),
+            EventCategory::Metric
+        );
+    }
+}

src/event_bus/mod.rs

@@ -0,0 +1,21 @@
+//! Unified event bus — the single source of truth for system events.
+//!
+//! All producers (agent, tools, scheduler, channels) emit events through one
+//! `EventBus`. Sinks subscribe and filter by category or payload type:
+//!
+//! - **SSE sink** → forwards `Domain` payloads to `SseManager` (web gateway)
+//! - **Audit sink** → persists `Audit` events to the append-only audit log
+//! - **State sink** → forwards `StateChange` payloads for cache invalidation
+//! - **Metrics sink** → delegates `Metric`/`Telemetry` to `Observer` trait
+//!
+//! Hook events remain separate — hooks are bidirectional interceptors (can
+//! reject/modify), the bus is unidirectional fire-and-forget.
+
+pub mod bus;
+pub mod event;
+pub mod sinks;
+
+pub use bus::EventBus;
+pub use event::{
+    EventCategory, EventContext, EventPayload, EventSource, SystemEvent, TelemetryPayload,
+};

src/event_bus/sinks/audit_sink.rs

@@ -0,0 +1,161 @@
+//! Audit sink — persists `Audit`-category events to the append-only audit log.
+//!
+//! Batches events (up to 32, or 500ms timeout) before flushing to the database.
+//! On DB failure, falls back to a local JSONL file so audit data is never lost.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::broadcast;
+
+use crate::db::AuditStore;
+use crate::event_bus::EventBus;
+use crate::event_bus::event::{EventCategory, SystemEvent};
+
+/// Maximum events to batch before flushing.
+const BATCH_SIZE: usize = 32;
+
+/// Maximum time to wait before flushing a partial batch.
+const FLUSH_INTERVAL: Duration = Duration::from_millis(500);
+
+/// Spawn the audit sink as a background task.
+pub fn spawn(bus: &EventBus, store: Arc<dyn AuditStore>) -> tokio::task::JoinHandle<()> {
+    let mut rx = bus.subscribe();
+
+    tokio::spawn(async move {
+        let mut batch: Vec<Arc<SystemEvent>> = Vec::with_capacity(BATCH_SIZE);
+        let mut flush_timer = tokio::time::interval(FLUSH_INTERVAL);
+        // First tick completes immediately — skip it.
+        flush_timer.tick().await;
+
+        loop {
+            tokio::select! {
+                result = rx.recv() => {
+                    match result {
+                        Ok(event) => {
+                            if event.category == EventCategory::Audit {
+                                batch.push(event);
+                                if batch.len() >= BATCH_SIZE {
+                                    flush_batch(&store, &mut batch).await;
+                                }
+                            }
+                        }
+                        Err(broadcast::error::RecvError::Lagged(n)) => {
+                            tracing::warn!(skipped = n, "Audit sink lagged behind event bus");
+                        }
+                        Err(broadcast::error::RecvError::Closed) => {
+                            // Flush remaining events before shutdown.
+                            if !batch.is_empty() {
+                                flush_batch(&store, &mut batch).await;
+                            }
+                            tracing::debug!("Event bus closed, audit sink shutting down");
+                            break;
+                        }
+                    }
+                }
+                _ = flush_timer.tick() => {
+                    if !batch.is_empty() {
+                        flush_batch(&store, &mut batch).await;
+                    }
+                }
+            }
+        }
+    })
+}
+
+async fn flush_batch(store: &Arc<dyn AuditStore>, batch: &mut Vec<Arc<SystemEvent>>) {
+    let records: Vec<crate::db::AuditRecord> = batch
+        .iter()
+        .map(|e| crate::db::AuditRecord {
+            event_id: e.id,
+            event_type: event_type_name(&e.payload),
+            source_module: e.source.module.clone(),
+            source_component: e.source.component.clone(),
+            category: format!("{:?}", e.category),
+            session_id: e.context.session_id,
+            thread_id: e.context.thread_id,
+            job_id: e.context.job_id,
+            user_id: e.context.user_id.clone(),
+            payload: serde_json::to_value(&e.payload).unwrap_or_default(),
+            created_at: e.timestamp,
+        })
+        .collect();
+
+    if let Err(e) = store.append_audit_events(&records).await {
+        tracing::error!(count = records.len(), error = %e, "Failed to persist audit events to DB, falling back to file");
+        fallback_to_file(&records);
+    }
+
+    batch.clear();
+}
+
+/// Extract a short event type name from the payload for indexing.
+fn event_type_name(payload: &crate::event_bus::event::EventPayload) -> String {
+    use crate::event_bus::event::EventPayload;
+    match payload {
+        EventPayload::Domain(_) => "domain".to_string(),
+        EventPayload::StateChange(_) => "state_change".to_string(),
+        EventPayload::Telemetry(_) => "telemetry".to_string(),
+        EventPayload::StateTransition { .. } => "state_transition".to_string(),
+        EventPayload::ToolExecution { .. } => "tool_execution".to_string(),
+        EventPayload::AuthEvent { .. } => "auth_event".to_string(),
+        EventPayload::ConfigChange { .. } => "config_change".to_string(),
+    }
+}
+
+/// Fallback: append audit records as JSONL to a local file.
+fn fallback_to_file(records: &[crate::db::AuditRecord]) {
+    let fallback_path = crate::bootstrap::ironclaw_base_dir().join("audit.fallback.jsonl");
+    let file = match std::fs::OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open(&fallback_path)
+    {
+        Ok(f) => f,
+        Err(e) => {
+            tracing::error!(path = %fallback_path.display(), error = %e, "Cannot open audit fallback file");
+            return;
+        }
+    };
+
+    let mut writer = std::io::BufWriter::new(file);
+    for record in records {
+        if let Err(e) = serde_json::to_writer(&mut writer, record) {
+            tracing::error!(error = %e, "Failed to write audit record to fallback file");
+        } else {
+            use std::io::Write;
+            let _ = writer.write_all(b"\n");
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn event_type_names() {
+        use crate::event_bus::event::EventPayload;
+
+        assert_eq!( // safety: test-only
+            event_type_name(&EventPayload::StateTransition {
+                entity_type: "t".into(),
+                entity_id: "i".into(),
+                from_state: "a".into(),
+                to_state: "b".into(),
+                reason: None,
+            }),
+            "state_transition"
+        );
+        assert_eq!( // safety: test-only
+            event_type_name(&EventPayload::ToolExecution {
+                tool_name: "echo".into(),
+                parameters_hash: "abc".into(),
+                duration_ms: 10,
+                success: true,
+                error: None,
+            }),
+            "tool_execution"
+        );
+    }
+}

src/event_bus/sinks/metrics_sink.rs

@@ -0,0 +1,132 @@
+//! Metrics sink — filters `Telemetry`/`Metric` events and delegates to `Observer`.
+//!
+//! Bridges the unified event bus to the existing `Observer` trait so that
+//! `LogObserver`, future OpenTelemetry exporters, etc. continue to work.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::broadcast;
+
+use crate::event_bus::EventBus;
+use crate::event_bus::event::{EventCategory, EventPayload, SystemEvent, TelemetryPayload};
+use crate::observability::traits::{Observer, ObserverEvent};
+
+/// Spawn the metrics sink as a background task.
+pub fn spawn(bus: &EventBus, observer: Arc<dyn Observer>) -> tokio::task::JoinHandle<()> {
+    let mut rx = bus.subscribe();
+
+    tokio::spawn(async move {
+        loop {
+            match rx.recv().await {
+                Ok(event) => forward_if_metric(&event, &observer),
+                Err(broadcast::error::RecvError::Lagged(n)) => {
+                    tracing::warn!(skipped = n, "Metrics sink lagged behind event bus");
+                }
+                Err(broadcast::error::RecvError::Closed) => {
+                    tracing::debug!("Event bus closed, metrics sink shutting down");
+                    break;
+                }
+            }
+        }
+    })
+}
+
+fn forward_if_metric(event: &Arc<SystemEvent>, observer: &Arc<dyn Observer>) {
+    if event.category != EventCategory::Metric {
+        return;
+    }
+
+    if let EventPayload::Telemetry(ref telemetry) = event.payload {
+        match telemetry {
+            TelemetryPayload::LlmCall {
+                provider,
+                model,
+                duration_ms,
+                success,
+                ..
+            } => {
+                observer.record_event(&ObserverEvent::LlmResponse {
+                    provider: provider.clone(),
+                    model: model.clone(),
+                    duration: Duration::from_millis(*duration_ms),
+                    success: *success,
+                    error_message: None,
+                });
+            }
+            TelemetryPayload::ChannelMessage { channel, direction } => {
+                observer.record_event(&ObserverEvent::ChannelMessage {
+                    channel: channel.clone(),
+                    direction: direction.clone(),
+                });
+            }
+            TelemetryPayload::HeartbeatTick => {
+                observer.record_event(&ObserverEvent::HeartbeatTick);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::event_bus::event::{EventContext, EventSource};
+    use crate::observability::traits::ObserverMetric;
+    use std::sync::Mutex;
+
+    struct RecordingObserver {
+        events: Mutex<Vec<String>>,
+    }
+
+    impl RecordingObserver {
+        fn new() -> Self {
+            Self {
+                events: Mutex::new(Vec::new()),
+            }
+        }
+
+        fn recorded(&self) -> Vec<String> {
+            self.events.lock().expect("test lock").clone() // safety: test-only
+        }
+    }
+
+    impl Observer for RecordingObserver {
+        fn record_event(&self, event: &ObserverEvent) {
+            let name = match event {
+                ObserverEvent::LlmResponse { .. } => "llm_response",
+                ObserverEvent::ChannelMessage { .. } => "channel_message",
+                ObserverEvent::HeartbeatTick => "heartbeat_tick",
+                _ => "other",
+            };
+            self.events
+                .lock()
+                .expect("test lock") // safety: test-only
+                .push(name.to_string());
+        }
+
+        fn record_metric(&self, _metric: &ObserverMetric) {}
+        fn name(&self) -> &str {
+            "test-recorder"
+        }
+    }
+
+    #[tokio::test]
+    async fn forwards_telemetry_to_observer() {
+        let bus = EventBus::new();
+        let observer = Arc::new(RecordingObserver::new());
+
+        let _handle = spawn(&bus, Arc::clone(&observer) as Arc<dyn Observer>);
+
+        bus.emit_telemetry(
+            EventSource::new("test", "metrics"),
+            EventContext::empty(),
+            TelemetryPayload::HeartbeatTick,
+        );
+
+        // Give the sink a moment to process
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+
+        let recorded = observer.recorded();
+        assert_eq!(recorded, vec!["heartbeat_tick"]); // safety: test-only
+    }
+}

src/event_bus/sinks/mod.rs

@@ -0,0 +1,48 @@
+//! Event bus sinks — subscribers that consume events by category.
+//!
+//! Each sink runs as a background task, filtering events and routing
+//! them to the appropriate subsystem.
+
+pub mod audit_sink;
+pub mod metrics_sink;
+pub mod sse_sink;
+pub mod state_sink;
+
+use std::sync::Arc;
+
+use crate::event_bus::EventBus;
+
+/// Spawn all configured sinks as background tasks.
+///
+/// Returns `JoinHandle`s so the caller can abort them on shutdown.
+pub fn spawn_sinks(
+    bus: &EventBus,
+    sse_tx: Option<tokio::sync::broadcast::Sender<crate::events::DomainEvent>>,
+    state_bus: Option<Arc<crate::state_bus::StateBus>>,
+    observer: Option<Arc<dyn crate::observability::Observer>>,
+    audit_store: Option<Arc<dyn crate::db::AuditStore>>,
+) -> Vec<tokio::task::JoinHandle<()>> {
+    let mut handles = Vec::new();
+
+    // SSE sink — bridges Domain events to the web gateway
+    if let Some(tx) = sse_tx {
+        handles.push(sse_sink::spawn(bus, tx));
+    }
+
+    // State sink — bridges StateChange events to the StateBus
+    if let Some(sb) = state_bus {
+        handles.push(state_sink::spawn(bus, sb));
+    }
+
+    // Metrics sink — bridges Telemetry/Metric events to Observer
+    if let Some(obs) = observer {
+        handles.push(metrics_sink::spawn(bus, obs));
+    }
+
+    // Audit sink — persists Audit events to the database
+    if let Some(store) = audit_store {
+        handles.push(audit_sink::spawn(bus, store));
+    }
+
+    handles
+}

src/event_bus/sinks/sse_sink.rs

@@ -0,0 +1,96 @@
+//! SSE sink — filters `Domain` payloads and forwards to `SseManager`.
+//!
+//! Drop-in replacement for direct `sse_tx.send()` calls. The web gateway's
+//! SSE wire format is unchanged because `DomainEvent` serialization is
+//! identical.
+
+use std::sync::Arc;
+
+use tokio::sync::broadcast;
+
+use crate::event_bus::EventBus;
+use crate::event_bus::event::{EventPayload, SystemEvent};
+use crate::events::DomainEvent;
+
+/// Spawn the SSE sink as a background task.
+pub fn spawn(
+    bus: &EventBus,
+    sse_tx: broadcast::Sender<DomainEvent>,
+) -> tokio::task::JoinHandle<()> {
+    let mut rx = bus.subscribe();
+
+    tokio::spawn(async move {
+        loop {
+            match rx.recv().await {
+                Ok(event) => forward_if_domain(&event, &sse_tx),
+                Err(broadcast::error::RecvError::Lagged(n)) => {
+                    tracing::warn!(skipped = n, "SSE sink lagged behind event bus");
+                }
+                Err(broadcast::error::RecvError::Closed) => {
+                    tracing::debug!("Event bus closed, SSE sink shutting down");
+                    break;
+                }
+            }
+        }
+    })
+}
+
+fn forward_if_domain(event: &Arc<SystemEvent>, sse_tx: &broadcast::Sender<DomainEvent>) {
+    if let EventPayload::Domain(ref domain_event) = event.payload {
+        // Ignore send error — no SSE subscribers is fine.
+        let _ = sse_tx.send(domain_event.clone());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::event_bus::event::{EventContext, EventSource};
+
+    #[tokio::test]
+    async fn forwards_domain_events() {
+        let bus = EventBus::new();
+        let (sse_tx, mut sse_rx) = broadcast::channel::<DomainEvent>(16);
+
+        let _handle = spawn(&bus, sse_tx);
+
+        bus.emit_domain(
+            EventSource::new("test", "sse"),
+            EventContext::empty(),
+            DomainEvent::Heartbeat,
+        );
+
+        let received = tokio::time::timeout(std::time::Duration::from_millis(100), sse_rx.recv())
+            .await
+            .expect("should receive within timeout") // safety: test-only
+            .expect("should not error"); // safety: test-only
+
+        assert!(matches!(received, DomainEvent::Heartbeat)); // safety: test-only
+    }
+
+    #[tokio::test]
+    async fn ignores_non_domain_events() {
+        let bus = EventBus::new();
+        let (sse_tx, mut sse_rx) = broadcast::channel::<DomainEvent>(16);
+
+        let _handle = spawn(&bus, sse_tx);
+
+        // Emit a non-domain event
+        bus.emit_state_change(crate::state_bus::StateChange::ConfigReloaded);
+
+        // Then emit a domain event so we know the sink is running
+        bus.emit_domain(
+            EventSource::new("test", "sse"),
+            EventContext::empty(),
+            DomainEvent::Heartbeat,
+        );
+
+        let received = tokio::time::timeout(std::time::Duration::from_millis(100), sse_rx.recv())
+            .await
+            .expect("should receive within timeout") // safety: test-only
+            .expect("should not error"); // safety: test-only
+
+        // Only the Heartbeat should arrive, not the StateChange
+        assert!(matches!(received, DomainEvent::Heartbeat)); // safety: test-only
+    }
+}

src/event_bus/sinks/state_sink.rs

@@ -0,0 +1,63 @@
+//! State sink — filters `StateChange` payloads and forwards to `StateBus`.
+//!
+//! Replaces direct `StateBus::publish()` calls. Modules that need state
+//! invalidation subscribe to the `StateBus` as before — the sink bridges
+//! the unified event bus to the existing invalidation mechanism.
+
+use std::sync::Arc;
+
+use tokio::sync::broadcast;
+
+use crate::event_bus::EventBus;
+use crate::event_bus::event::{EventPayload, SystemEvent};
+use crate::state_bus::StateBus;
+
+/// Spawn the state sink as a background task.
+pub fn spawn(bus: &EventBus, state_bus: Arc<StateBus>) -> tokio::task::JoinHandle<()> {
+    let mut rx = bus.subscribe();
+
+    tokio::spawn(async move {
+        loop {
+            match rx.recv().await {
+                Ok(event) => forward_if_state_change(&event, &state_bus),
+                Err(broadcast::error::RecvError::Lagged(n)) => {
+                    tracing::warn!(skipped = n, "State sink lagged behind event bus");
+                }
+                Err(broadcast::error::RecvError::Closed) => {
+                    tracing::debug!("Event bus closed, state sink shutting down");
+                    break;
+                }
+            }
+        }
+    })
+}
+
+fn forward_if_state_change(event: &Arc<SystemEvent>, state_bus: &StateBus) {
+    if let EventPayload::StateChange(ref change) = event.payload {
+        state_bus.publish(change.clone());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::state_bus::StateChange;
+
+    #[tokio::test]
+    async fn forwards_state_changes() {
+        let bus = EventBus::new();
+        let state_bus = Arc::new(StateBus::new());
+        let mut state_rx = state_bus.subscribe();
+
+        let _handle = spawn(&bus, Arc::clone(&state_bus));
+
+        bus.emit_state_change(StateChange::ConfigReloaded);
+
+        let received = tokio::time::timeout(std::time::Duration::from_millis(100), state_rx.recv())
+            .await
+            .expect("should receive within timeout") // safety: test-only
+            .expect("should not error"); // safety: test-only
+
+        assert!(matches!(received, StateChange::ConfigReloaded)); // safety: test-only
+    }
+}

src/lib.rs

@@ -51,6 +51,7 @@ pub mod document_extraction;
 pub mod error;
 pub mod estimation;
 pub mod evaluation;
+pub mod event_bus;
 pub mod events;
 pub mod extensions;
 pub mod history;

src/main.rs

@@ -698,6 +698,7 @@ async fn async_main() -> anyhow::Result<()> {
         document_extraction: Some(Arc::new(
             ironclaw::document_extraction::DocumentExtractionMiddleware::new(),
         )),
+        event_bus: Some(components.event_bus.clone()),
     };
 
     let mut agent = Agent::new(

src/models/routine.rs

@@ -714,7 +714,8 @@ mod tests {
         };
         let json = trigger.to_config_json();
         let parsed = Trigger::from_db("event", json).expect("parse event"); // safety: test-only
-        assert!( // safety: test-only
+        assert!(
+            // safety: test-only
             // safety: test-only
             matches!(parsed, Trigger::Event { channel, pattern } // safety: test-only
             if channel == Some("telegram".to_string()) && pattern == r"deploy\s+\w+")
@@ -733,7 +734,8 @@ mod tests {
         };
         let json = trigger.to_config_json();
         let parsed = Trigger::from_db("system_event", json).expect("parse system_event"); // safety: test-only
-        assert!( // safety: test-only
+        assert!(
+            // safety: test-only
             // safety: test-only
             // safety: test-only
             matches!(parsed, Trigger::SystemEvent { source, event_type, filters: f }
@@ -752,7 +754,8 @@ mod tests {
         };
         let json = action.to_config_json();
         let parsed = RoutineAction::from_db("lightweight", json).expect("parse lightweight"); // safety: test-only
-        assert!( // safety: test-only
+        assert!(
+            // safety: test-only
             // safety: test-only
             // safety: test-only
             matches!(parsed, RoutineAction::Lightweight { prompt, context_paths, max_tokens, .. }
@@ -770,7 +773,8 @@ mod tests {
         };
         let json = action.to_config_json();
         let parsed = RoutineAction::from_db("full_job", json).expect("parse full_job"); // safety: test-only
-        assert!( // safety: test-only
+        assert!(
+            // safety: test-only
             // safety: test-only
             // safety: test-only
             matches!(parsed, RoutineAction::FullJob { title, max_iterations, tool_permissions, .. }
@@ -823,7 +827,8 @@ mod tests {
         };
         let json = trigger.to_config_json();
         let parsed = Trigger::from_db("cron", json).expect("parse cron"); // safety: test-only
-        assert!( // safety: test-only
+        assert!(
+            // safety: test-only
             // safety: test-only
             matches!(parsed, Trigger::Cron { schedule, timezone } // safety: test-only
                 if schedule == "0 9 * * MON-FRI"
@@ -842,7 +847,8 @@ mod tests {
     fn test_trigger_cron_invalid_timezone_coerced_to_none() {
         let json = serde_json::json!({"schedule": "0 9 * * *", "timezone": "Fake/Zone"});
         let parsed = Trigger::from_db("cron", json).expect("parse cron"); // safety: test-only
-        assert!( // safety: test-only
+        assert!(
+            // safety: test-only
             // safety: test-only
             // safety: test-only
             matches!(parsed, Trigger::Cron { timezone, .. } if timezone.is_none()),
@@ -872,7 +878,8 @@ mod tests {
 
     #[test]
     fn test_trigger_type_tag() {
-        assert_eq!( // safety: test-only
+        assert_eq!(
+            // safety: test-only
             // safety: test-only
             // safety: test-only
             Trigger::Cron {
@@ -882,7 +889,8 @@ mod tests {
             .type_tag(),
             "cron"
         );
-        assert_eq!( // safety: test-only
+        assert_eq!(
+            // safety: test-only
             // safety: test-only
             // safety: test-only
             Trigger::Event {
@@ -892,7 +900,8 @@ mod tests {
             .type_tag(),
             "event"
         );
-        assert_eq!( // safety: test-only
+        assert_eq!(
+            // safety: test-only
             // safety: test-only
             // safety: test-only
             Trigger::SystemEvent {
@@ -915,7 +924,8 @@ mod tests {
             "max_tokens": 4096
         });
         let parsed = RoutineAction::from_db("lightweight", json).expect("parse lightweight"); // safety: test-only
-        assert!( // safety: test-only
+        assert!(
+            // safety: test-only
             // safety: test-only
             // safety: test-only
             matches!(parsed, RoutineAction::Lightweight { use_tools, max_tool_rounds, .. }
@@ -936,7 +946,8 @@ mod tests {
             RoutineAction::Lightweight {
                 max_tool_rounds, ..
             } => {
-                assert_eq!( // safety: test-only
+                assert_eq!(
+                    // safety: test-only
                     // safety: test-only
                     // safety: test-only
                     max_tool_rounds,

src/orchestrator/api.rs

@@ -879,21 +879,15 @@ mod tests {
         // Insert a handle so update_worker_status has something to update
         {
             let mut containers = state.job_manager.containers.write().await;
-            containers.insert(
+            let mut handle = crate::orchestrator::job_manager::ContainerHandle::new(
                 job_id,
-                crate::orchestrator::job_manager::ContainerHandle {
-                    job_id,
-                    container_id: "test-container".to_string(),
-                    state: crate::orchestrator::job_manager::ContainerState::Running,
-                    mode: crate::orchestrator::job_manager::JobMode::Worker,
-                    created_at: chrono::Utc::now(),
-                    project_dir: None,
-                    task_description: "test".to_string(),
-                    last_worker_status: None,
-                    worker_iteration: 0,
-                    completion_result: None,
-                },
+                crate::orchestrator::job_manager::JobMode::Worker,
+                "test".to_string(),
+                None,
             );
+            handle.container_id = "test-container".to_string();
+            handle.mark_running();
+            containers.insert(job_id, handle);
         }
 
         let jm = Arc::clone(&state.job_manager);

src/orchestrator/job_manager.rs

@@ -94,6 +94,17 @@ pub enum ContainerState {
     Failed,
 }
 
+impl ContainerState {
+    /// Check whether a transition from this state to `target` is valid.
+    pub fn can_transition_to(self, target: ContainerState) -> bool {
+        use ContainerState::*;
+        matches!(
+            (self, target),
+            (Creating, Running) | (Creating, Failed) | (Running, Stopped) | (Running, Failed)
+        )
+    }
+}
+
 impl std::fmt::Display for ContainerState {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
@@ -110,7 +121,8 @@ impl std::fmt::Display for ContainerState {
 pub struct ContainerHandle {
     pub job_id: Uuid,
     pub container_id: String,
-    pub state: ContainerState,
+    /// Private — use `state()` to read, `mark_running()`/`mark_stopped()`/`mark_failed()` to mutate.
+    state: ContainerState,
     pub mode: JobMode,
     pub created_at: DateTime<Utc>,
     pub project_dir: Option<PathBuf>,
@@ -125,6 +137,49 @@ pub struct ContainerHandle {
     // It lives only in the TokenStore (never logged, serialized, or persisted).
 }
 
+impl ContainerHandle {
+    /// Create a new container handle in `Creating` state.
+    pub fn new(
+        job_id: Uuid,
+        mode: JobMode,
+        task_description: String,
+        project_dir: Option<PathBuf>,
+    ) -> Self {
+        Self {
+            job_id,
+            container_id: String::new(),
+            state: ContainerState::Creating,
+            mode,
+            created_at: Utc::now(),
+            project_dir,
+            task_description,
+            last_worker_status: None,
+            worker_iteration: 0,
+            completion_result: None,
+        }
+    }
+
+    /// Get the current container state.
+    pub fn state(&self) -> ContainerState {
+        self.state
+    }
+
+    /// Transition from Creating to Running.
+    pub fn mark_running(&mut self) {
+        self.state = ContainerState::Running;
+    }
+
+    /// Transition to Stopped.
+    pub fn mark_stopped(&mut self) {
+        self.state = ContainerState::Stopped;
+    }
+
+    /// Transition to Failed.
+    pub fn mark_failed(&mut self) {
+        self.state = ContainerState::Failed;
+    }
+}
+
 /// Result reported by a worker on completion.
 #[derive(Debug, Clone)]
 pub struct CompletionResult {
@@ -265,18 +320,7 @@ impl ContainerJobManager {
             .await;
 
         // Record the handle
-        let handle = ContainerHandle {
-            job_id,
-            container_id: String::new(), // set after container creation
-            state: ContainerState::Creating,
-            mode,
-            created_at: Utc::now(),
-            project_dir: project_dir.clone(),
-            task_description: task.to_string(),
-            last_worker_status: None,
-            worker_iteration: 0,
-            completion_result: None,
-        };
+        let handle = ContainerHandle::new(job_id, mode, task.to_string(), project_dir.clone());
         self.containers.write().await.insert(job_id, handle);
 
         // Run the actual container creation. On any failure, revoke the token
@@ -450,7 +494,7 @@ impl ContainerJobManager {
         // Update handle with container ID
         if let Some(handle) = self.containers.write().await.get_mut(&job_id) {
             handle.container_id = container_id;
-            handle.state = ContainerState::Running;
+            handle.mark_running();
         }
 
         tracing::info!(
@@ -507,7 +551,7 @@ impl ContainerJobManager {
 
         // Update state
         if let Some(handle) = self.containers.write().await.get_mut(&job_id) {
-            handle.state = ContainerState::Stopped;
+            handle.mark_stopped();
         }
 
         // Revoke the auth token
@@ -530,7 +574,7 @@ impl ContainerJobManager {
             let mut containers = self.containers.write().await;
             if let Some(handle) = containers.get_mut(&job_id) {
                 handle.completion_result = Some(result);
-                handle.state = ContainerState::Stopped;
+                handle.mark_stopped();
             }
         }
 
@@ -680,22 +724,12 @@ mod tests {
 
         // Insert a handle
         {
+            let mut handle =
+                ContainerHandle::new(job_id, JobMode::Worker, "test job".to_string(), None);
+            handle.container_id = "test".to_string();
+            handle.mark_running();
             let mut containers = mgr.containers.write().await;
-            containers.insert(
-                job_id,
-                ContainerHandle {
-                    job_id,
-                    container_id: "test".to_string(),
-                    state: ContainerState::Running,
-                    mode: JobMode::Worker,
-                    created_at: chrono::Utc::now(),
-                    project_dir: None,
-                    task_description: "test job".to_string(),
-                    last_worker_status: None,
-                    worker_iteration: 0,
-                    completion_result: None,
-                },
-            );
+            containers.insert(job_id, handle);
         }
 
         mgr.update_worker_status(job_id, Some("Iteration 3".to_string()), 3)

src/resilience/health.rs

@@ -118,7 +118,8 @@ mod tests {
     fn test_endpoint_health_becomes_unhealthy() {
         let h = EndpointHealth::new();
         for i in 0..4 {
-            assert!( // safety: test-only
+            assert!(
+                // safety: test-only
                 // safety: test-only
                 !h.record_failure(5),
                 "should not be unhealthy at failure {}",

src/state_bus.rs

@@ -14,7 +14,7 @@ use tokio::sync::broadcast;
 use uuid::Uuid;
 
 /// A state change notification.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, serde::Serialize)]
 pub enum StateChange {
     /// A routine was created, updated, toggled, or deleted.
     RoutineUpdated { routine_id: Uuid },
@@ -121,7 +121,8 @@ mod tests {
 
         // First recv should report a lag.
         let result = rx.recv().await;
-        assert!( // safety: test-only
+        assert!(
+            // safety: test-only
             // safety: test-only
             result.is_ok() || result.is_err(),
             "lagged receiver should either get an event or a Lagged error"

src/testing/mod.rs

@@ -455,6 +455,7 @@ impl TestHarnessBuilder {
             http_interceptor: None,
             transcription: None,
             document_extraction: None,
+            event_bus: None,
         };
 
         TestHarness {

src/tools/builtin/job.rs

@@ -463,7 +463,7 @@ impl CreateJobTool {
             }
 
             match jm.get_handle(job_id).await {
-                Some(handle) => match handle.state {
+                Some(handle) => match handle.state() {
                     crate::orchestrator::job_manager::ContainerState::Running
                     | crate::orchestrator::job_manager::ContainerState::Creating => {
                         tokio::time::sleep(poll_interval).await;

src/tools/registry.rs

@@ -194,6 +194,52 @@ impl ToolRegistry {
         self.tools.try_read().map(|t| t.len()).unwrap_or(0)
     }
 
+    /// Verify that expected built-in tools are registered.
+    ///
+    /// Returns a list of tool names that should be registered but aren't.
+    /// Called during startup to catch wiring bugs.
+    pub fn verify_expected_tools(&self, config: &crate::config::Config) -> Vec<String> {
+        let tools = match self.tools.try_read() {
+            Ok(t) => t,
+            Err(_) => return Vec::new(),
+        };
+
+        let mut missing = Vec::new();
+
+        // Core built-ins that should always be present
+        for name in &["echo", "time", "json", "http"] {
+            if !tools.contains_key(*name) {
+                missing.push(name.to_string());
+            }
+        }
+
+        // Memory tools should be present when tools beyond core builtins are loaded
+        // (indicates a workspace/DB is available).
+        if tools.len() > 4 {
+            for name in &[
+                "memory_search",
+                "memory_write",
+                "memory_read",
+                "memory_tree",
+            ] {
+                if !tools.contains_key(*name) {
+                    missing.push(name.to_string());
+                }
+            }
+        }
+
+        // Dev tools when local tools allowed
+        if config.agent.allow_local_tools {
+            for name in &["shell", "read_file", "write_file"] {
+                if !tools.contains_key(*name) {
+                    missing.push(name.to_string());
+                }
+            }
+        }
+
+        missing
+    }
+
     /// Get all tools.
     pub async fn all(&self) -> Vec<Arc<dyn Tool>> {
         self.tools.read().await.values().cloned().collect()

src/worker/job.rs

@@ -54,6 +54,9 @@ pub struct WorkerDeps {
     pub approval_context: Option<ApprovalContext>,
     /// HTTP interceptor for trace recording/replay (propagated to JobContext).
     pub http_interceptor: Option<Arc<dyn crate::llm::recording::HttpInterceptor>>,
+    /// Unified event bus. When present, job events are emitted through the bus
+    /// in addition to the legacy SSE + DB paths.
+    pub event_bus: Option<crate::event_bus::EventBus>,
 }
 
 /// Worker that executes a single job.
@@ -120,9 +123,63 @@ impl Worker {
     }
 
     /// Fire-and-forget persistence of a job event and SSE broadcast.
+    ///
+    /// Also emits through the unified event bus when available, so the
+    /// audit sink and other subscribers capture all job activity.
     fn log_event(&self, event_type: &str, data: serde_json::Value) {
         let job_id = self.job_id;
 
+        // Emit through unified event bus (audit + future sinks)
+        if let Some(ref bus) = self.deps.event_bus {
+            use crate::event_bus::{EventContext, EventSource};
+            use crate::events::DomainEvent;
+
+            let job_id_str = job_id.to_string();
+            let domain_event = match event_type {
+                "message" => Some(DomainEvent::JobMessage {
+                    job_id: job_id_str.clone(),
+                    role: data
+                        .get("role")
+                        .and_then(|v| v.as_str())
+                        .unwrap_or("assistant")
+                        .to_string(),
+                    content: data
+                        .get("content")
+                        .and_then(|v| v.as_str())
+                        .unwrap_or("")
+                        .to_string(),
+                }),
+                "status" => Some(DomainEvent::JobStatus {
+                    job_id: job_id_str.clone(),
+                    message: data
+                        .get("message")
+                        .and_then(|v| v.as_str())
+                        .unwrap_or("")
+                        .to_string(),
+                }),
+                "result" => Some(DomainEvent::JobResult {
+                    job_id: job_id_str.clone(),
+                    status: data
+                        .get("status")
+                        .and_then(|v| v.as_str())
+                        .unwrap_or("completed")
+                        .to_string(),
+                    session_id: data
+                        .get("session_id")
+                        .and_then(|v| v.as_str())
+                        .map(|s| s.to_string()),
+                }),
+                _ => None,
+            };
+            if let Some(event) = domain_event {
+                bus.emit_domain(
+                    EventSource::new("worker", "job"),
+                    EventContext::with_job(job_id),
+                    event,
+                );
+            }
+        }
+
         // Persist to DB
         if let Some(store) = self.store() {
             let store = store.clone();
@@ -1452,6 +1509,7 @@ mod tests {
             sse_tx: None,
             approval_context: None,
             http_interceptor: None,
+            event_bus: None,
         };
 
         Worker::new(job_id, deps)
@@ -1642,6 +1700,7 @@ mod tests {
             sse_tx: None,
             approval_context,
             http_interceptor: None,
+            event_bus: None,
         };
 
         Worker::new(job_id, deps)

tests/batch_query_tests.rs

@@ -213,8 +213,10 @@ mod tests {
         assert_eq!(counts.len(), 3, "Should return 3 routines"); // safety: test-only
         assert_eq!(counts[&r1], 2, "r1 should have 2 running"); // safety: test-only
         assert_eq!(counts[&r2], 1, "r2 should have 1 running"); // safety: test-only
- assert_eq!( // safety: test-only
-            counts[&r3], 0,
+        assert_eq!(
+            // safety: test-only
+            counts[&r3],
+            0,
             "r3 should have 0 running (Ok status is not running)"
         );
     }
@@ -360,8 +362,10 @@ mod tests {
             .await
             .expect("batch query should work"); // safety: test-only
 
- assert_eq!( // safety: test-only
-            counts[&routine_id], 2,
+        assert_eq!(
+            // safety: test-only
+            counts[&routine_id],
+            2,
             "Should only count 2 Running status runs"
         );
     }
@@ -454,12 +458,16 @@ mod tests {
             .expect("batch query should work"); // safety: test-only
 
         // Verify counts match the limits
- assert_eq!( // safety: test-only
-            counts[&r1], 1,
+        assert_eq!(
+            // safety: test-only
+            counts[&r1],
+            1,
             "r1 should have 1 running (at max_concurrent=1)"
         );
- assert_eq!( // safety: test-only
-            counts[&r2], 2,
+        assert_eq!(
+            // safety: test-only
+            counts[&r2],
+            2,
             "r2 should have 2 running (at max_concurrent=2)"
         );
 

tests/boundary_chaos.rs

@@ -310,7 +310,8 @@ fn domain_event_all_variants_serialize() {
     for variant in &variants {
         let json = serde_json::to_string(variant).unwrap(); // safety: test-only
         let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); // safety: test-only
-        assert!( // safety: test-only
+        assert!(
+            // safety: test-only
             // safety: test-only
             parsed.get("type").is_some(),
             "missing 'type' field in {:?}",

tests/support/gateway_workflow_harness.rs

@@ -252,6 +252,7 @@ impl GatewayWorkflowHarness {
                 http_interceptor: None,
                 transcription: None,
                 document_extraction: None,
+                event_bus: Some(components.event_bus.clone()),
             },
             channels,
             None,

tests/support/test_rig.rs

@@ -641,6 +641,7 @@ impl TestRigBuilder {
             },
             transcription: None,
             document_extraction: None,
+            event_bus: Some(components.event_bus.clone()),
         };
 
         // 7. Create TestChannel and ChannelManager.