diff --git a/migrations/V13__audit_log.sql b/migrations/V13__audit_log.sql new file mode 100644 index 00000000..450ae619 --- /dev/null +++ b/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); diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 9eaad8e5..8ee3f09d 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -75,6 +75,8 @@ pub struct AgentDeps { pub cost_guard: Arc, /// SSE broadcast sender for live job event streaming to the web gateway. pub sse_tx: Option>, + /// Unified event bus. Optional for backward compatibility with tests. + pub event_bus: Option, /// HTTP interceptor for trace recording/replay. pub http_interceptor: Option>, /// Audio transcription middleware for voice messages. diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index 9e6747f2..32b2b430 100644 --- a/src/agent/dispatcher.rs +++ b/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( diff --git a/src/agent/scheduler.rs b/src/agent/scheduler.rs index bc6cf962..95d9d3fc 100644 --- a/src/agent/scheduler.rs +++ b/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); diff --git a/src/agent/session.rs b/src/agent/session.rs index 6085b027..3a1922f3 100644 --- a/src/agent/session.rs +++ b/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, /// 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, /// Tool calls made during this turn. pub tool_calls: Vec, - /// 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, /// 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) { 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); + } } diff --git a/src/agent/session_manager.rs b/src/agent/session_manager.rs index 3db275cc..65095057 100644 --- a/src/agent/session_manager.rs +++ b/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); } } } diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index 213d2176..94677c32 100644 --- a/src/agent/thread_ops.rs +++ b/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); } } diff --git a/src/app.rs b/src/app.rs index 00804de1..e81384cf 100644 --- a/src/app.rs +++ b/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, pub catalog_entries: Vec, pub dev_loaded_tool_names: Vec, + /// 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) } } diff --git a/src/channels/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index 909a252c..79833804 100644 --- a/src/channels/web/handlers/chat.rs +++ b/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(), diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index e8cb33c2..50b8cfa8 100644 --- a/src/channels/web/server.rs +++ b/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(), diff --git a/src/config/mod.rs b/src/config/mod.rs index 52997963..59a68110 100644 --- a/src/config/mod.rs +++ b/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 { + 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. diff --git a/src/db/libsql/audit.rs b/src/db/libsql/audit.rs new file mode 100644 index 00000000..53d1a003 --- /dev/null +++ b/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 { + 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, 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 = 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) + } +} diff --git a/src/db/libsql/mod.rs b/src/db/libsql/mod.rs index dcc5a8b5..9715e7bd 100644 --- a/src/db/libsql/mod.rs +++ b/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; diff --git a/src/db/libsql_migrations.rs b/src/db/libsql_migrations.rs index fc445b7c..51b0a69f 100644 --- a/src/db/libsql_migrations.rs +++ b/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); "#, ), ]; diff --git a/src/db/mod.rs b/src/db/mod.rs index d77a1a37..5ab06508 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -641,6 +641,70 @@ pub trait WorkspaceStore: Send + Sync { ) -> Result, 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, + /// Thread ID (if applicable). + pub thread_id: Option, + /// Job ID (if applicable). + pub job_id: Option, + /// User ID (if applicable). + pub user_id: Option, + /// Full event payload as JSON. + pub payload: serde_json::Value, + /// When the event was created. + pub created_at: DateTime, +} + +/// Filter for querying the audit log. +#[derive(Debug, Default)] +pub struct AuditFilter { + /// Filter by session ID. + pub session_id: Option, + /// Filter by job ID. + pub job_id: Option, + /// Filter by user ID. + pub user_id: Option, + /// Filter by event type. + pub event_type: Option, + /// Only events after this time. + pub after: Option>, + /// Only events before this time. + pub before: Option>, + /// Maximum number of records to return. + pub limit: Option, +} + +/// 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, DatabaseError>; +} + /// Backend-agnostic database supertrait. /// /// Combines all sub-traits into one. Existing `Arc` consumers diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 8c18e252..41483aee 100644 --- a/src/db/postgres.rs +++ b/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> = 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, ¶m_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, 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> = 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, ¶m_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) + } +} diff --git a/src/event_bus/bus.rs b/src/event_bus/bus.rs new file mode 100644 index 00000000..14d2ad90 --- /dev/null +++ b/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`-wrapping avoids deep-cloning payloads across multiple sinks. +/// The monotonic sequence counter ensures total ordering. +#[derive(Clone)] +pub struct EventBus { + tx: broadcast::Sender>, + seq: Arc, +} + +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, + entity_id: impl Into, + from_state: impl Into, + to_state: impl Into, + reason: Option, + ) { + 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, + parameters_hash: impl Into, + duration_ms: u64, + success: bool, + error: Option, + ) { + 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> { + 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 + } +} diff --git a/src/event_bus/event.rs b/src/event_bus/event.rs new file mode 100644 index 00000000..8c00913a --- /dev/null +++ b/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, + /// 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, component: impl Into) -> 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, + /// Thread ID (if applicable). + #[serde(skip_serializing_if = "Option::is_none")] + pub thread_id: Option, + /// Job ID (if applicable). + #[serde(skip_serializing_if = "Option::is_none")] + pub job_id: Option, + /// User ID (if applicable). + #[serde(skip_serializing_if = "Option::is_none")] + pub user_id: Option, +} + +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) -> 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, + 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, + }, + + /// 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, + }, + + /// 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 + ); + } +} diff --git a/src/event_bus/mod.rs b/src/event_bus/mod.rs new file mode 100644 index 00000000..4b04fa0b --- /dev/null +++ b/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, +}; diff --git a/src/event_bus/sinks/audit_sink.rs b/src/event_bus/sinks/audit_sink.rs new file mode 100644 index 00000000..be59d56a --- /dev/null +++ b/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) -> tokio::task::JoinHandle<()> { + let mut rx = bus.subscribe(); + + tokio::spawn(async move { + let mut batch: Vec> = 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, batch: &mut Vec>) { + let records: Vec = 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" + ); + } +} diff --git a/src/event_bus/sinks/metrics_sink.rs b/src/event_bus/sinks/metrics_sink.rs new file mode 100644 index 00000000..34ecf187 --- /dev/null +++ b/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) -> 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, observer: &Arc) { + 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>, + } + + impl RecordingObserver { + fn new() -> Self { + Self { + events: Mutex::new(Vec::new()), + } + } + + fn recorded(&self) -> Vec { + 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); + + 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 + } +} diff --git a/src/event_bus/sinks/mod.rs b/src/event_bus/sinks/mod.rs new file mode 100644 index 00000000..e665b050 --- /dev/null +++ b/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>, + state_bus: Option>, + observer: Option>, + audit_store: Option>, +) -> Vec> { + 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 +} diff --git a/src/event_bus/sinks/sse_sink.rs b/src/event_bus/sinks/sse_sink.rs new file mode 100644 index 00000000..7b44d257 --- /dev/null +++ b/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, +) -> 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, sse_tx: &broadcast::Sender) { + 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::(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::(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 + } +} diff --git a/src/event_bus/sinks/state_sink.rs b/src/event_bus/sinks/state_sink.rs new file mode 100644 index 00000000..73afe6ca --- /dev/null +++ b/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) -> 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, 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 + } +} diff --git a/src/lib.rs b/src/lib.rs index 689d9fed..95ca6875 100644 --- a/src/lib.rs +++ b/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; diff --git a/src/main.rs b/src/main.rs index 0b469530..d5134f3a 100644 --- a/src/main.rs +++ b/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( diff --git a/src/models/routine.rs b/src/models/routine.rs index 5cfac200..c3af1caf 100644 --- a/src/models/routine.rs +++ b/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, diff --git a/src/orchestrator/api.rs b/src/orchestrator/api.rs index b46aa8c6..2996ed12 100644 --- a/src/orchestrator/api.rs +++ b/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); diff --git a/src/orchestrator/job_manager.rs b/src/orchestrator/job_manager.rs index 34b9f373..3c62aa60 100644 --- a/src/orchestrator/job_manager.rs +++ b/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, pub project_dir: Option, @@ -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, + ) -> 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) diff --git a/src/resilience/health.rs b/src/resilience/health.rs index af1822dd..b211a61a 100644 --- a/src/resilience/health.rs +++ b/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 {}", diff --git a/src/state_bus.rs b/src/state_bus.rs index 5f384759..da079b2e 100644 --- a/src/state_bus.rs +++ b/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" diff --git a/src/testing/mod.rs b/src/testing/mod.rs index 33702e67..e63d3f37 100644 --- a/src/testing/mod.rs +++ b/src/testing/mod.rs @@ -455,6 +455,7 @@ impl TestHarnessBuilder { http_interceptor: None, transcription: None, document_extraction: None, + event_bus: None, }; TestHarness { diff --git a/src/tools/builtin/job.rs b/src/tools/builtin/job.rs index 9346d14a..ff921c87 100644 --- a/src/tools/builtin/job.rs +++ b/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; diff --git a/src/tools/registry.rs b/src/tools/registry.rs index 754869c8..41c5062f 100644 --- a/src/tools/registry.rs +++ b/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 { + 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> { self.tools.read().await.values().cloned().collect() diff --git a/src/worker/job.rs b/src/worker/job.rs index 1247a552..f21570f8 100644 --- a/src/worker/job.rs +++ b/src/worker/job.rs @@ -54,6 +54,9 @@ pub struct WorkerDeps { pub approval_context: Option, /// HTTP interceptor for trace recording/replay (propagated to JobContext). pub http_interceptor: Option>, + /// Unified event bus. When present, job events are emitted through the bus + /// in addition to the legacy SSE + DB paths. + pub event_bus: Option, } /// 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) diff --git a/tests/batch_query_tests.rs b/tests/batch_query_tests.rs index fc2b63b7..4daa2d81 100644 --- a/tests/batch_query_tests.rs +++ b/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)" ); diff --git a/tests/boundary_chaos.rs b/tests/boundary_chaos.rs index fd202397..fe8d92a5 100644 --- a/tests/boundary_chaos.rs +++ b/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 {:?}", diff --git a/tests/support/gateway_workflow_harness.rs b/tests/support/gateway_workflow_harness.rs index dd9e8643..4cf687a2 100644 --- a/tests/support/gateway_workflow_harness.rs +++ b/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, diff --git a/tests/support/test_rig.rs b/tests/support/test_rig.rs index 07106e42..892aeffb 100644 --- a/tests/support/test_rig.rs +++ b/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.