oxios_kernel/
event_bus.rs

1//! Event bus: inter-agent communication via `oxi_sdk::EventBus<KernelEvent>`.
2//!
3//! The event bus is the "pipe" of Oxios. All agents communicate
4//! through kernel events published on the bus.
5//!
6//! After RFC-014 Phase C, this module no longer owns the broadcast channel —
7//! it reuses `oxi_sdk::EventBus<E>`, which is a generic wrapper over
8//! `tokio::sync::broadcast`. The only Oxios-specific bits are:
9//!
10//! - `KernelEvent` enum (oxios-internal event vocabulary)
11//! - `kernel_event_to_audit_action` mapping for the audit trail
12//! - `attach_audit_trail` helper (subscribes the bus to the trail)
13
14use oxi_sdk::EventBus as SdkEventBus;
15use oxi_sdk::observability::{AuditAction, AuditTrail};
16use serde::{Deserialize, Serialize};
17use std::sync::Arc;
18use uuid::Uuid;
19
20use crate::types::AgentId;
21
22/// Kernel event bus — generic SDK bus specialised for `KernelEvent`.
23///
24/// The broadcast channel is owned by `oxi_sdk::EventBus`; this type alias
25/// just makes the call sites read more naturally (`crate::event_bus::EventBus`
26/// instead of `oxi_sdk::EventBus<KernelEvent>`).
27pub type EventBus = SdkEventBus<KernelEvent>;
28
29/// Events that flow through the kernel event bus.
30#[derive(Debug, Clone, Serialize, Deserialize)]
31pub enum KernelEvent {
32    /// A new agent has been created.
33    AgentCreated {
34        /// The new agent's ID.
35        id: AgentId,
36        /// The agent's name/goal.
37        name: String,
38    },
39    /// An agent has started executing.
40    AgentStarted {
41        /// The agent's ID.
42        id: AgentId,
43    },
44    /// An agent has been stopped.
45    ///
46    /// Carries `success` so consumers can distinguish a normal completion
47    /// (`success: true`) from an evaluation/assessment failure
48    /// (`success: false`). Infrastructure errors (panic, timeout) emit
49    /// `AgentFailed` instead.
50    AgentStopped {
51        /// The agent's ID.
52        id: AgentId,
53        /// Whether the agent's result passed evaluation. Mirrors
54        /// `ExecutionResult.success` from the Ok path; `false` on the
55        /// kill/terminate path (user-initiated stop).
56        #[serde(default)]
57        success: bool,
58    },
59    /// An agent has encountered a failure.
60    AgentFailed {
61        /// The agent's ID.
62        id: AgentId,
63        /// Description of the error.
64        error: String,
65    },
66    /// A message has been received from an agent.
67    MessageReceived {
68        /// The sending agent's ID.
69        from: AgentId,
70        /// Message content.
71        content: String,
72    },
73    /// An agent has produced output.
74    AgentOutput {
75        /// The session this output belongs to.
76        session_id: String,
77        /// The agent's ID.
78        agent_id: AgentId,
79        /// The output content.
80        output: String,
81    },
82    /// A HitL approval request has been submitted.
83    ApprovalRequested {
84        /// The approval request ID.
85        id: uuid::Uuid,
86        /// The tool requesting approval.
87        tool_name: String,
88        /// The action requiring approval.
89        action: String,
90        /// The resource involved.
91        resource: String,
92        /// Reason for the request.
93        reason: String,
94        /// The session ID that triggered this request.
95        session_id: Option<String>,
96    },
97    /// A HitL approval has been resolved (approved or rejected).
98    ApprovalResolved {
99        /// The approval request ID.
100        id: uuid::Uuid,
101        /// Whether it was approved (true) or rejected (false).
102        approved: bool,
103    },
104    /// An agent tried to access a file path outside its allowed_paths.
105    /// The frontend renders a path-access card offering: create a Mount,
106    /// temporarily allow, or deny. Mirrors `ApprovalRequested` but carries
107    /// path-specific context so the card and resolve endpoint know what to do.
108    PathAccessRequested {
109        /// The request ID (matches the PendingPathApprovals entry).
110        id: uuid::Uuid,
111        /// The tool that tried to access the path (read, write, edit …).
112        tool_name: String,
113        /// The denied path (absolute).
114        path: String,
115        /// Access mode: "read" or "write".
116        mode: String,
117        /// The agent whose `allowed_paths` would need updating.
118        agent_name: String,
119        /// Human-readable denial reason from the AccessGate.
120        reason: String,
121        /// The session ID that triggered this request.
122        session_id: Option<String>,
123    },
124    /// A memory entry was stored.
125    MemoryStored {
126        /// Memory entry ID.
127        id: String,
128        /// Memory type label.
129        memory_type: String,
130        /// Source of the memory.
131        source: String,
132    },
133    /// Memories were recalled for a new session.
134    MemoryRecalled {
135        /// The recall query.
136        query: String,
137        /// Number of memories returned.
138        count: usize,
139    },
140    /// Multi-agent group created.
141    AgentGroupCreated {
142        /// The group's ID.
143        group_id: uuid::Uuid,
144        /// Number of agents in the group.
145        agent_count: usize,
146    },
147    /// An agent in a group completed.
148    AgentGroupMemberCompleted {
149        /// The group's ID.
150        group_id: uuid::Uuid,
151        /// The agent's ID.
152        agent_id: uuid::Uuid,
153        /// Whether the agent succeeded.
154        success: bool,
155    },
156    /// A new Project has been created (RFC-011).
157    ProjectCreated {
158        /// The project's ID.
159        project_id: uuid::Uuid,
160        /// The project's name.
161        name: String,
162        /// How it was created.
163        source: String,
164    },
165    /// A Project has been activated (RFC-011).
166    ProjectActivated {
167        /// The project's ID.
168        project_id: uuid::Uuid,
169        /// The project's name.
170        name: String,
171    },
172
173    // ── RFC-015 Chat Transparency ─────────────────────────────
174    // Real-time events emitted by AgentRuntime during tool execution
175    // and streaming. Web channel converts these to WS chunks.
176    /// A tool execution has started (real-time, RFC-015).
177    ToolExecutionStarted {
178        /// Session this tool call belongs to.
179        session_id: String,
180        /// Name of the tool (e.g. "read_file", "bash", "memory_recall").
181        tool_name: String,
182        /// Provider-specific tool call ID used to correlate start/end.
183        tool_call_id: String,
184        /// Tool input arguments (JSON).
185        tool_args: serde_json::Value,
186        /// Semantic context inferred by oxi-agent 0.32+ from tool name/args
187        /// (e.g. WebSearch, PageVisit). `None` for tools without context mapping.
188        #[serde(default, skip_serializing_if = "Option::is_none")]
189        context: Option<serde_json::Value>,
190    },
191    /// A tool execution has finished (real-time, RFC-015).
192    ToolExecutionFinished {
193        /// Session this tool call belongs to.
194        session_id: String,
195        /// Provider-specific tool call ID.
196        tool_call_id: String,
197        /// Name of the tool.
198        tool_name: String,
199        /// Wall-clock duration in milliseconds.
200        duration_ms: u64,
201        /// Whether the tool returned an error.
202        is_error: bool,
203        /// Truncated output (max ~500 chars) for streaming.
204        output_summary: String,
205    },
206    /// A tool execution emitted a progress update (real-time, RFC-015).
207    ToolExecutionProgress {
208        /// Session this tool call belongs to.
209        session_id: String,
210        /// Provider-specific tool call ID.
211        tool_call_id: String,
212        /// Name of the tool.
213        tool_name: String,
214        /// Human-readable progress text (already-formatted by the tool).
215        progress: String,
216        /// Tab that emitted this progress event, if the upstream tool tracks
217        /// tabs. `None` for tools that don't have a tab concept (e.g. legacy
218        /// oxi-agent versions that don't propagate `tab_id`).
219        #[serde(default, skip_serializing_if = "Option::is_none")]
220        tab_id: Option<Uuid>,
221        /// Semantic context from the tool call (e.g. PageVisit, WebSearch).
222        /// Stored as `serde_json::Value` to decouple kernel events from
223        /// oxi-sdk's internal `ToolCallContext` enum. UI consumers that
224        /// understand a context variant render it richly; older consumers
225        /// simply ignore the field.
226        #[serde(default, skip_serializing_if = "Option::is_none")]
227        context: Option<serde_json::Value>,
228    },
229    /// Memory was recalled during agent execution (RFC-015).
230    MemoryRecallUsed {
231        /// Session this recall belongs to.
232        session_id: String,
233        /// The recall query.
234        query: String,
235        /// Number of memories returned.
236        count: usize,
237        /// Memory tier source ("hot" | "warm" | "cold").
238        source: String,
239    },
240    /// Token usage update (RFC-015).
241    TokenUsageUpdate {
242        /// Session this usage belongs to.
243        session_id: String,
244        /// Cumulative input tokens.
245        input_tokens: u64,
246        /// Cumulative output tokens.
247        output_tokens: u64,
248    },
249    /// Reasoning/compaction fragment (RFC-015).
250    ReasoningFragment {
251        /// Session this fragment belongs to.
252        session_id: String,
253        /// The fragment text (chain-of-thought, compaction summary, etc).
254        content: String,
255        /// Source label: "chain_of_thought" | "compaction" | "reflection".
256        source: String,
257    },
258    /// Partial tool-call arguments streamed by the LLM (RFC-015 Phase C).
259    ///
260    /// Emitted by oxi 0.58+ (`AgentEvent::ToolCallDelta`) while the model is
261    /// still constructing a tool call, before `ToolExecutionStarted`. Each
262    /// `args_delta` is a raw JSON fragment; consumers accumulate per
263    /// `tool_call_id`.
264    ToolArgsDelta {
265        /// Session this delta belongs to.
266        session_id: String,
267        /// Tool call identifier (matches the later `ToolExecutionStarted`).
268        tool_call_id: String,
269        /// Raw JSON argument fragment from the LLM stream.
270        args_delta: String,
271    },
272
273    /// Compaction was triggered (RFC-035 gap 2 observability).
274    ///
275    /// Emitted when `oxi_sdk::CompactionEvent::Triggered` fires (0.53.0+).
276    /// `source` is one of:
277    /// - `"provider-reported"` — provider-reported `usage.input_tokens` drove
278    ///   the trigger (ground truth; gap 2's primary signal)
279    /// - `"bytes/4 heuristic (cold start)"` — legacy heuristic; only on turn 1
280    ///   before any `ProviderEvent::Done` has been observed
281    /// - `"empty"` — empty context (no trigger source)
282    CompactionTriggered {
283        /// Session this compaction belongs to.
284        session_id: Option<String>,
285        /// The trigger source label from `CompactionEvent::Triggered::source`.
286        source: String,
287    },
288    // ── Calendar ──────────────────────────────────────────────
289    /// A calendar event was created.
290    CalendarEventCreated {
291        /// Event UID.
292        uid: String,
293        /// Event title.
294        title: String,
295        /// Start time.
296        start: String,
297        /// End time.
298        end: String,
299    },
300    /// A calendar event was updated.
301    CalendarEventUpdated {
302        /// Event UID.
303        uid: String,
304        /// Event title.
305        title: String,
306    },
307    /// A calendar event was deleted.
308    CalendarEventDeleted {
309        /// Event UID.
310        uid: String,
311        /// Event title.
312        title: String,
313    },
314    /// An email has been sent.
315    EmailSent {
316        /// Email subject.
317        subject: String,
318        /// SMTP message ID.
319        message_id: String,
320        /// Template name (if template was used/saved).
321        #[serde(default, skip_serializing_if = "Option::is_none")]
322        template_name: Option<String>,
323    },
324
325    // ── Knowledge ──────────────────────────────────────────────
326    /// A knowledge note was persisted (hook, user, or tool).
327    KnowledgePersisted {
328        session_id: String,
329        message_index: usize,
330        path: String,
331        source: String, // "hook", "user", "tool"
332    },
333    /// A knowledge note was removed by user action.
334    KnowledgeRemoved {
335        session_id: String,
336        message_index: usize,
337    },
338    /// A question was posed to the user by the agent (RFC-027, `ask_user`).
339    /// The frontend renders an input/option picker and resolves the
340    /// pending oneshot via a separate response endpoint.
341    AskUserRequest {
342        /// Unique request ID — used by the response handler to resolve
343        /// the oneshot the tool is awaiting.
344        id: String,
345        /// The question text the user sees.
346        question: String,
347        /// Optional structured options. Empty when the question is open-ended.
348        options: Vec<String>,
349    },
350    // ── Persona (agent-authored writes are security-reviewed) ───────────
351    /// A persona was created (by an agent tool, the HTTP API, or the UI).
352    PersonaCreated {
353        /// Persona ID.
354        id: String,
355        /// Persona display name.
356        name: String,
357        /// Whether it was registered enabled.
358        enabled: bool,
359        /// Origin of the change: "agent" | "api" | "ui".
360        source: String,
361    },
362    /// A persona was updated.
363    PersonaUpdated {
364        /// Persona ID.
365        id: String,
366        /// Persona display name.
367        name: String,
368        /// Origin of the change: "agent" | "api" | "ui".
369        source: String,
370    },
371    // ── Integration install (RFC-041 M3) ────────────────────────────────────
372    // These ride the same SSE channel as tool execution progress so the UI
373    // can stream install output without opening a second connection. Routing
374    // is by `job_id` (opaque to clients; the daemon mints it via uuid).
375    /// An integration install job has started.
376    IntegrationInstallStarted {
377        /// Opaque job ID; clients correlate all subsequent events with this.
378        job_id: String,
379        /// Integration registry id (e.g. "github").
380        integration_id: String,
381        /// Human-readable label.
382        label: String,
383    },
384    /// Incremental progress — a stdout line or stage transition.
385    IntegrationInstallProgress {
386        job_id: String,
387        integration_id: String,
388        /// One line of output or a stage label (e.g. "fetching", "extracting").
389        line: String,
390    },
391    /// Install completed successfully.
392    IntegrationInstallCompleted {
393        job_id: String,
394        integration_id: String,
395        /// Final command + summary output for the audit log and UI.
396        command: String,
397        output: String,
398        exit_code: Option<i32>,
399    },
400    /// Install failed (non-zero exit or spawn error).
401    IntegrationInstallFailed {
402        job_id: String,
403        integration_id: String,
404        error: String,
405    },
406}
407
408/// Convert a KernelEvent to an AuditAction for the audit trail.
409pub fn kernel_event_to_audit_action(event: &KernelEvent) -> AuditAction {
410    match event {
411        KernelEvent::AgentCreated { name, .. } => AuditAction::AgentSpawn {
412            task_type: name.clone(),
413        },
414        KernelEvent::AgentStarted { .. } => AuditAction::AgentSpawn {
415            task_type: "started".to_string(),
416        },
417        KernelEvent::AgentStopped { success, .. } => AuditAction::AgentExit {
418            reason: if *success {
419                "completed".to_string()
420            } else {
421                "stopped".to_string()
422            },
423        },
424        KernelEvent::AgentFailed { error, .. } => AuditAction::AgentExit {
425            reason: error.clone(),
426        },
427        KernelEvent::MessageReceived { content, .. } => AuditAction::Other {
428            detail: format!("message: {content}"),
429        },
430        KernelEvent::AgentOutput { output, .. } => AuditAction::Other {
431            detail: format!("agent_output:{output}"),
432        },
433        KernelEvent::ApprovalRequested {
434            id,
435            action,
436            resource,
437            ..
438        } => AuditAction::Other {
439            detail: format!("approval_requested:{id}:{action}:{resource}"),
440        },
441        KernelEvent::ApprovalResolved { id, approved } => AuditAction::Other {
442            detail: format!("approval_resolved:{id}:{approved}"),
443        },
444        KernelEvent::PathAccessRequested {
445            id,
446            path,
447            tool_name,
448            ..
449        } => AuditAction::Other {
450            detail: format!("path_access_requested:{id}:{tool_name}:{path}"),
451        },
452        KernelEvent::MemoryStored {
453            id, memory_type, ..
454        } => AuditAction::MemoryWrite {
455            entry_id: format!("{id}:{memory_type}"),
456        },
457        KernelEvent::MemoryRecalled { query, count } => AuditAction::MemoryRead {
458            entry_id: format!("query:{query}:{count}results"),
459        },
460        KernelEvent::AgentGroupCreated {
461            group_id,
462            agent_count,
463        } => AuditAction::Other {
464            detail: format!("group_created:{group_id}:{agent_count}agents"),
465        },
466        KernelEvent::AgentGroupMemberCompleted {
467            group_id,
468            agent_id,
469            success,
470        } => AuditAction::Other {
471            detail: format!("group_member_completed:{group_id}:{agent_id}:{success}"),
472        },
473        KernelEvent::ProjectCreated {
474            project_id: _,
475            name,
476            source,
477        } => AuditAction::Other {
478            detail: format!("project_created:{name}:{source}"),
479        },
480        KernelEvent::ProjectActivated {
481            project_id: _,
482            name,
483        } => AuditAction::Other {
484            detail: format!("project_activated:{name}"),
485        },
486        // ── RFC-015 ──
487        KernelEvent::ToolExecutionStarted { tool_name, .. } => AuditAction::Other {
488            detail: format!("tool_started:{tool_name}"),
489        },
490        KernelEvent::ToolExecutionFinished {
491            tool_name,
492            is_error,
493            ..
494        } => AuditAction::Other {
495            detail: format!(
496                "tool_finished:{tool_name}:{}",
497                if *is_error { "error" } else { "ok" }
498            ),
499        },
500        KernelEvent::ToolExecutionProgress {
501            tool_name,
502            tab_id,
503            context,
504            ..
505        } => AuditAction::Other {
506            detail: {
507                let mut d = format!("tool_progress:{tool_name}");
508                if let Some(id) = tab_id {
509                    d.push_str(&format!(":tab={id}"));
510                }
511                if let Some(ctx) = context
512                    .as_ref()
513                    .and_then(|c| c.get("kind"))
514                    .and_then(|k| k.as_str())
515                {
516                    d.push_str(&format!(":{ctx}"));
517                }
518                d
519            },
520        },
521        KernelEvent::MemoryRecallUsed { query, count, .. } => AuditAction::MemoryRead {
522            entry_id: format!("recall:{query}:{count}results"),
523        },
524        KernelEvent::TokenUsageUpdate {
525            input_tokens,
526            output_tokens,
527            ..
528        } => AuditAction::Other {
529            detail: format!("tokens:in={input_tokens}:out={output_tokens}"),
530        },
531        KernelEvent::ReasoningFragment { source, .. } => AuditAction::Other {
532            detail: format!("reasoning:{source}"),
533        },
534        KernelEvent::ToolArgsDelta { tool_call_id, .. } => AuditAction::Other {
535            detail: format!("tool_args_delta:{tool_call_id}"),
536        },
537        KernelEvent::CompactionTriggered { source, .. } => AuditAction::Other {
538            detail: format!("compaction:triggered:{source}"),
539        },
540        KernelEvent::CalendarEventCreated { uid, title, .. } => AuditAction::Other {
541            detail: format!("calendar:created:{uid}:{title}"),
542        },
543        KernelEvent::CalendarEventUpdated { uid, title } => AuditAction::Other {
544            detail: format!("calendar:updated:{uid}:{title}"),
545        },
546        KernelEvent::CalendarEventDeleted { uid, title } => AuditAction::Other {
547            detail: format!("calendar:deleted:{uid}:{title}"),
548        },
549        KernelEvent::EmailSent {
550            subject,
551            message_id,
552            template_name,
553        } => AuditAction::Other {
554            detail: format!("email:sent:{subject} (msg={message_id}, tpl={template_name:?})"),
555        },
556        KernelEvent::KnowledgePersisted {
557            session_id,
558            message_index,
559            path,
560            source,
561        } => AuditAction::Other {
562            detail: format!("knowledge:persisted:{session_id}:{message_index}:{path}:{source}"),
563        },
564        KernelEvent::KnowledgeRemoved {
565            session_id,
566            message_index,
567        } => AuditAction::Other {
568            detail: format!("knowledge:removed:{session_id}:{message_index}"),
569        },
570        KernelEvent::AskUserRequest { id, question, .. } => AuditAction::Other {
571            detail: format!("ask_user:{id}:{question}"),
572        },
573        KernelEvent::PersonaCreated {
574            id, name, source, ..
575        } => AuditAction::Other {
576            detail: format!("persona:created:{id}:{name}:{source}"),
577        },
578        KernelEvent::PersonaUpdated { id, name, source } => AuditAction::Other {
579            detail: format!("persona:updated:{id}:{name}:{source}"),
580        },
581        KernelEvent::IntegrationInstallStarted {
582            job_id,
583            integration_id,
584            ..
585        } => AuditAction::Other {
586            detail: format!("install:started:{integration_id}:{job_id}"),
587        },
588        KernelEvent::IntegrationInstallProgress {
589            job_id,
590            integration_id,
591            ..
592        } => AuditAction::Other {
593            detail: format!("install:progress:{integration_id}:{job_id}"),
594        },
595        KernelEvent::IntegrationInstallCompleted {
596            job_id,
597            integration_id,
598            exit_code,
599            ..
600        } => AuditAction::Other {
601            detail: format!(
602                "install:completed:{integration_id}:{job_id}:exit={}",
603                exit_code
604                    .map(|c| c.to_string())
605                    .unwrap_or_else(|| "?".into())
606            ),
607        },
608        KernelEvent::IntegrationInstallFailed {
609            job_id,
610            integration_id,
611            ..
612        } => AuditAction::Other {
613            detail: format!("install:failed:{integration_id}:{job_id}"),
614        },
615    }
616}
617
618/// Extract agent ID from a KernelEvent variant.
619fn extract_agent_id(event: &KernelEvent) -> String {
620    match event {
621        KernelEvent::AgentCreated { id, .. } => id.to_string(),
622        KernelEvent::AgentStarted { id, .. } => id.to_string(),
623        KernelEvent::AgentStopped { id, .. } => id.to_string(),
624        KernelEvent::AgentFailed { id, .. } => id.to_string(),
625        KernelEvent::MessageReceived { from, .. } => from.to_string(),
626        KernelEvent::AgentOutput { agent_id, .. } => agent_id.to_string(),
627        KernelEvent::AgentGroupMemberCompleted { agent_id, .. } => agent_id.to_string(),
628        KernelEvent::ProjectActivated { project_id, .. } => format!("project:{project_id}"),
629        // RFC-015: session-scoped events use session_id as the subject
630        KernelEvent::ToolExecutionStarted { session_id, .. } => format!("session:{session_id}"),
631        KernelEvent::ToolExecutionFinished { session_id, .. } => format!("session:{session_id}"),
632        KernelEvent::ToolExecutionProgress { session_id, .. } => format!("session:{session_id}"),
633        KernelEvent::MemoryRecallUsed { session_id, .. } => format!("session:{session_id}"),
634        KernelEvent::TokenUsageUpdate { session_id, .. } => format!("session:{session_id}"),
635        KernelEvent::ReasoningFragment { session_id, .. } => format!("session:{session_id}"),
636        KernelEvent::ToolArgsDelta { session_id, .. } => format!("session:{session_id}"),
637        KernelEvent::KnowledgePersisted { session_id, .. } => format!("session:{session_id}"),
638        KernelEvent::KnowledgeRemoved { session_id, .. } => format!("session:{session_id}"),
639        KernelEvent::CompactionTriggered { session_id, .. } => session_id
640            .as_ref()
641            .map(|s| format!("session:{s}"))
642            .unwrap_or_else(|| "system".to_string()),
643        _ => "system".to_string(),
644    }
645}
646
647/// Subscribe the audit trail to all kernel events.
648///
649/// The bus is broadcast-based; this spawns a long-running task that
650/// forwards every event into the audit trail as a structured entry.
651/// Lagged subscribers are logged and recovered.
652pub fn attach_audit_trail(bus: &EventBus, audit: Arc<AuditTrail>) {
653    let mut rx = bus.subscribe();
654    tokio::spawn(async move {
655        loop {
656            match rx.recv().await {
657                Ok(event) => {
658                    // Skip high-frequency streaming variants that are UI data,
659                    // not audit-relevant. ToolCallDelta fires per-token (raw
660                    // JSON fragments) — appending each would flood the Merkle
661                    // chain + JSONL with partial-JSON Debug strings per call.
662                    if matches!(event, KernelEvent::ToolArgsDelta { .. }) {
663                        continue;
664                    }
665                    let actor = extract_agent_id(&event);
666                    let action = kernel_event_to_audit_action(&event);
667                    let resource = format!("{event:?}");
668                    audit.append(actor, action, resource);
669                }
670                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
671                    // Surface the drop as a metric so operators can detect
672                    // incomplete audit trails instead of the events
673                    // vanishing silently (state-area F4).
674                    crate::metrics::get_metrics().audit_lagged_events.inc_by(n);
675                    tracing::warn!(
676                        skipped = n,
677                        "Audit trail subscriber lagged, skipping events"
678                    );
679                    continue;
680                }
681                Err(tokio::sync::broadcast::error::RecvError::Closed) => {
682                    tracing::info!("Audit trail event bus closed, exiting");
683                    break;
684                }
685            }
686        }
687    });
688}
689
690#[cfg(test)]
691mod tests {
692    use super::*;
693
694    fn sample_event(name: &str) -> KernelEvent {
695        KernelEvent::AgentCreated {
696            id: AgentId::new_v4(),
697            name: name.to_string(),
698        }
699    }
700
701    #[test]
702    fn test_event_bus_uses_sdk() {
703        let bus: EventBus = EventBus::new(256);
704        assert!(format!("{:?}", bus).contains("EventBus"));
705    }
706
707    #[tokio::test]
708    async fn test_publish_no_subscribers_ok() {
709        let bus = EventBus::new(16);
710        let result = bus.publish(sample_event("orphan"));
711        assert!(result.is_ok());
712    }
713
714    #[tokio::test]
715    async fn test_single_subscriber_receives_event() {
716        let bus = EventBus::new(16);
717        let mut rx = bus.subscribe();
718
719        let event = sample_event("test-agent");
720        bus.publish(event.clone()).unwrap();
721
722        let received = rx.try_recv().expect("should receive event");
723        match received {
724            KernelEvent::AgentCreated { name, .. } => assert_eq!(name, "test-agent"),
725            _ => panic!("wrong event type"),
726        }
727    }
728
729    #[tokio::test]
730    async fn test_multiple_subscribers_receive_events() {
731        let bus = EventBus::new(16);
732        let mut rx1 = bus.subscribe();
733        let mut rx2 = bus.subscribe();
734
735        let event = sample_event("multi");
736        bus.publish(event.clone()).unwrap();
737
738        let r1 = rx1.try_recv().expect("rx1 should receive event");
739        let r2 = rx2.try_recv().expect("rx2 should receive event");
740
741        assert!(matches!(r1, KernelEvent::AgentCreated { .. }));
742        assert!(matches!(r2, KernelEvent::AgentCreated { .. }));
743    }
744
745    #[tokio::test]
746    async fn test_kernel_event_to_audit_action() {
747        let event = KernelEvent::AgentFailed {
748            id: AgentId::new_v4(),
749            error: "boom".to_string(),
750        };
751        let action = kernel_event_to_audit_action(&event);
752        match action {
753            AuditAction::AgentExit { reason } => assert_eq!(reason, "boom"),
754            other => panic!("expected AgentExit, got {other:?}"),
755        }
756    }
757
758    // ── RFC-015 chat transparency event coverage ──
759
760    /// Round-trip JSON serialization for every new RFC-015 variant. This
761    /// guards against accidental renames that would break the WebSocket
762    /// wire format on the frontend.
763    #[test]
764    fn test_rfc015_event_round_trip_json() {
765        let cases: Vec<KernelEvent> = vec![
766            KernelEvent::ToolExecutionStarted {
767                session_id: "s1".into(),
768                tool_name: "read_file".into(),
769                tool_call_id: "call_1".into(),
770                tool_args: serde_json::json!({"path": "/src/main.rs"}),
771                context: None,
772            },
773            KernelEvent::ToolExecutionFinished {
774                session_id: "s1".into(),
775                tool_call_id: "call_1".into(),
776                tool_name: "read_file".into(),
777                duration_ms: 234,
778                is_error: false,
779                output_summary: "fn main() {}".into(),
780            },
781            KernelEvent::ToolExecutionProgress {
782                session_id: "s1".into(),
783                tool_call_id: "call_1".into(),
784                tool_name: "read_file".into(),
785                progress: "reading line 42/100".into(),
786                tab_id: None,
787                context: None,
788            },
789            KernelEvent::MemoryRecallUsed {
790                session_id: "s1".into(),
791                query: "rust errors".into(),
792                count: 3,
793                source: "warm".into(),
794            },
795            KernelEvent::TokenUsageUpdate {
796                session_id: "s1".into(),
797                input_tokens: 1234,
798                output_tokens: 567,
799            },
800            KernelEvent::ReasoningFragment {
801                session_id: "s1".into(),
802                content: "compaction done".into(),
803                source: "compaction".into(),
804            },
805        ];
806        for event in cases {
807            let json = serde_json::to_string(&event).expect("serialize");
808            let back: KernelEvent = serde_json::from_str(&json).expect("deserialize");
809            let json2 = serde_json::to_string(&back).expect("serialize round-trip");
810            assert_eq!(json, json2, "round-trip should be stable");
811        }
812    }
813
814    /// Tool progress events serialize/deserialize cleanly and round-trip
815    /// stable JSON, matching the wire format the WS layer expects.
816    #[test]
817    fn test_tool_execution_progress_serde_round_trip() {
818        let event = KernelEvent::ToolExecutionProgress {
819            session_id: "s-abc".into(),
820            tool_call_id: "call_42".into(),
821            tool_name: "browse".into(),
822            progress: "loading https://example.com".into(),
823            tab_id: Some(Uuid::new_v4()),
824            context: None,
825        };
826        let json = serde_json::to_string(&event).expect("serialize");
827        let back: KernelEvent = serde_json::from_str(&json).expect("deserialize");
828        match back {
829            KernelEvent::ToolExecutionProgress {
830                ref session_id,
831                ref tool_call_id,
832                ref tool_name,
833                ref progress,
834                tab_id,
835                ..
836            } => {
837                assert_eq!(session_id, "s-abc");
838                assert_eq!(tool_call_id, "call_42");
839                assert_eq!(tool_name, "browse");
840                assert_eq!(progress, "loading https://example.com");
841                assert!(tab_id.is_some(), "tab_id should round-trip when present");
842            }
843            other => panic!("expected ToolExecutionProgress, got {other:?}"),
844        }
845    }
846
847    /// The audit-action mapping for tool progress should produce a stable,
848    /// searchable detail string (used by the audit-trail UI to filter).
849    /// When `tab_id` is set, the detail includes `:tab=<id>`; when absent,
850    /// the original `tool_progress:<tool>` form is preserved (back-compat
851    /// for older oxi-agent versions that don't propagate tabs).
852    #[test]
853    fn test_tool_execution_progress_audit_action() {
854        let with_tab = KernelEvent::ToolExecutionProgress {
855            session_id: "s1".into(),
856            tool_call_id: "c1".into(),
857            tool_name: "browse".into(),
858            progress: "navigating".into(),
859            tab_id: Some(Uuid::new_v4()),
860            context: None,
861        };
862        match kernel_event_to_audit_action(&with_tab) {
863            AuditAction::Other { detail } => {
864                assert!(detail.contains("tool_progress"), "detail: {detail}");
865                assert!(detail.contains("browse"), "detail: {detail}");
866                assert!(
867                    detail.contains(":tab="),
868                    "detail should include tab id: {detail}"
869                );
870            }
871            other => panic!("expected Other, got {other:?}"),
872        }
873        let without_tab = KernelEvent::ToolExecutionProgress {
874            session_id: "s1".into(),
875            tool_call_id: "c1".into(),
876            tool_name: "browse".into(),
877            progress: "navigating".into(),
878            tab_id: None,
879            context: None,
880        };
881        match kernel_event_to_audit_action(&without_tab) {
882            AuditAction::Other { detail } => {
883                assert_eq!(detail, "tool_progress:browse");
884            }
885            other => panic!("expected Other, got {other:?}"),
886        }
887    }
888
889    /// `tab_id` is optional in serde (`#[serde(default)]`) so older oxi-agent
890    /// versions that don't emit it still round-trip cleanly. This guards the
891    /// backwards-compat contract explicitly.
892    #[test]
893    fn test_tool_execution_progress_tab_id_optional_in_serde() {
894        // Simulate a payload from a legacy oxi-agent (no tab_id key).
895        // KernelEvent is externally tagged, so the variant is the JSON key.
896        let legacy_json = r#"{
897            "ToolExecutionProgress": {
898                "session_id": "s-old",
899                "tool_call_id": "call_legacy",
900                "tool_name": "browse",
901                "progress": "step 1"
902            }
903        }"#;
904        let event: KernelEvent = serde_json::from_str(legacy_json).expect("deserialize legacy");
905        match &event {
906            KernelEvent::ToolExecutionProgress {
907                session_id,
908                tool_call_id,
909                tool_name,
910                progress,
911                tab_id,
912                ..
913            } => {
914                assert_eq!(session_id, "s-old");
915                assert_eq!(tool_call_id, "call_legacy");
916                assert_eq!(tool_name, "browse");
917                assert_eq!(progress, "step 1");
918                assert!(tab_id.is_none(), "missing field should default to None");
919            }
920            other => panic!("expected ToolExecutionProgress, got {other:?}"),
921        }
922        // And re-serialise — `skip_serializing_if = "Option::is_none"` keeps
923        // the wire format clean when downstream tools don't set tab_id.
924        let json = serde_json::to_string(&event).expect("serialize");
925        assert!(
926            !json.contains("tab_id"),
927            "tab_id should be omitted when None: {json}"
928        );
929    }
930
931    /// The agent_id extractor should map session-scoped RFC-015 events to
932    /// `session:<id>` for audit-trail grouping, while non-session events
933    /// keep their existing behaviour.
934    #[test]
935    fn test_rfc015_extract_agent_id() {
936        let event = KernelEvent::ToolExecutionStarted {
937            session_id: "abc-123".into(),
938            tool_name: "bash".into(),
939            tool_call_id: "c1".into(),
940            tool_args: serde_json::Value::Null,
941            context: None,
942        };
943        // The function is private; verify via the public AuditAction mapping
944        // that session-scoped events do not collide with real agent ids.
945        let action = kernel_event_to_audit_action(&event);
946        match action {
947            AuditAction::Other { detail } => {
948                assert!(
949                    detail.contains("bash"),
950                    "tool name in audit detail: {detail}"
951                );
952            }
953            other => panic!("expected Other, got {other:?}"),
954        }
955    }
956}
oxios_kernel/event_bus.rs

oxios_kernel/
event_bus.rs
