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    /// A chunk of the compression summary being streamed.
407    CompressionDelta {
408        /// The session being compressed.
409        session_id: String,
410        /// Incremental summary text.
411        delta: String,
412    },
413    /// Compression completed successfully.
414    CompressionDone {
415        /// The session that was compressed.
416        session_id: String,
417    },
418    /// Compression failed.
419    CompressionFailed {
420        /// The session that failed compression.
421        session_id: String,
422        /// Error description.
423        error: String,
424    },
425}
426
427/// Convert a KernelEvent to an AuditAction for the audit trail.
428pub fn kernel_event_to_audit_action(event: &KernelEvent) -> AuditAction {
429    match event {
430        KernelEvent::AgentCreated { name, .. } => AuditAction::AgentSpawn {
431            task_type: name.clone(),
432        },
433        KernelEvent::AgentStarted { .. } => AuditAction::AgentSpawn {
434            task_type: "started".to_string(),
435        },
436        KernelEvent::AgentStopped { success, .. } => AuditAction::AgentExit {
437            reason: if *success {
438                "completed".to_string()
439            } else {
440                "stopped".to_string()
441            },
442        },
443        KernelEvent::AgentFailed { error, .. } => AuditAction::AgentExit {
444            reason: error.clone(),
445        },
446        KernelEvent::MessageReceived { content, .. } => AuditAction::Other {
447            detail: format!("message: {content}"),
448        },
449        KernelEvent::AgentOutput { output, .. } => AuditAction::Other {
450            detail: format!("agent_output:{output}"),
451        },
452        KernelEvent::ApprovalRequested {
453            id,
454            action,
455            resource,
456            ..
457        } => AuditAction::Other {
458            detail: format!("approval_requested:{id}:{action}:{resource}"),
459        },
460        KernelEvent::ApprovalResolved { id, approved } => AuditAction::Other {
461            detail: format!("approval_resolved:{id}:{approved}"),
462        },
463        KernelEvent::PathAccessRequested {
464            id,
465            path,
466            tool_name,
467            ..
468        } => AuditAction::Other {
469            detail: format!("path_access_requested:{id}:{tool_name}:{path}"),
470        },
471        KernelEvent::MemoryStored {
472            id, memory_type, ..
473        } => AuditAction::MemoryWrite {
474            entry_id: format!("{id}:{memory_type}"),
475        },
476        KernelEvent::MemoryRecalled { query, count } => AuditAction::MemoryRead {
477            entry_id: format!("query:{query}:{count}results"),
478        },
479        KernelEvent::AgentGroupCreated {
480            group_id,
481            agent_count,
482        } => AuditAction::Other {
483            detail: format!("group_created:{group_id}:{agent_count}agents"),
484        },
485        KernelEvent::AgentGroupMemberCompleted {
486            group_id,
487            agent_id,
488            success,
489        } => AuditAction::Other {
490            detail: format!("group_member_completed:{group_id}:{agent_id}:{success}"),
491        },
492        KernelEvent::ProjectCreated {
493            project_id: _,
494            name,
495            source,
496        } => AuditAction::Other {
497            detail: format!("project_created:{name}:{source}"),
498        },
499        KernelEvent::ProjectActivated {
500            project_id: _,
501            name,
502        } => AuditAction::Other {
503            detail: format!("project_activated:{name}"),
504        },
505        // ── RFC-015 ──
506        KernelEvent::ToolExecutionStarted { tool_name, .. } => AuditAction::Other {
507            detail: format!("tool_started:{tool_name}"),
508        },
509        KernelEvent::ToolExecutionFinished {
510            tool_name,
511            is_error,
512            ..
513        } => AuditAction::Other {
514            detail: format!(
515                "tool_finished:{tool_name}:{}",
516                if *is_error { "error" } else { "ok" }
517            ),
518        },
519        KernelEvent::ToolExecutionProgress {
520            tool_name,
521            tab_id,
522            context,
523            ..
524        } => AuditAction::Other {
525            detail: {
526                let mut d = format!("tool_progress:{tool_name}");
527                if let Some(id) = tab_id {
528                    d.push_str(&format!(":tab={id}"));
529                }
530                if let Some(ctx) = context
531                    .as_ref()
532                    .and_then(|c| c.get("kind"))
533                    .and_then(|k| k.as_str())
534                {
535                    d.push_str(&format!(":{ctx}"));
536                }
537                d
538            },
539        },
540        KernelEvent::MemoryRecallUsed { query, count, .. } => AuditAction::MemoryRead {
541            entry_id: format!("recall:{query}:{count}results"),
542        },
543        KernelEvent::TokenUsageUpdate {
544            input_tokens,
545            output_tokens,
546            ..
547        } => AuditAction::Other {
548            detail: format!("tokens:in={input_tokens}:out={output_tokens}"),
549        },
550        KernelEvent::ReasoningFragment { source, .. } => AuditAction::Other {
551            detail: format!("reasoning:{source}"),
552        },
553        KernelEvent::ToolArgsDelta { tool_call_id, .. } => AuditAction::Other {
554            detail: format!("tool_args_delta:{tool_call_id}"),
555        },
556        KernelEvent::CompactionTriggered { source, .. } => AuditAction::Other {
557            detail: format!("compaction:triggered:{source}"),
558        },
559        KernelEvent::CalendarEventCreated { uid, title, .. } => AuditAction::Other {
560            detail: format!("calendar:created:{uid}:{title}"),
561        },
562        KernelEvent::CalendarEventUpdated { uid, title } => AuditAction::Other {
563            detail: format!("calendar:updated:{uid}:{title}"),
564        },
565        KernelEvent::CalendarEventDeleted { uid, title } => AuditAction::Other {
566            detail: format!("calendar:deleted:{uid}:{title}"),
567        },
568        KernelEvent::EmailSent {
569            subject,
570            message_id,
571            template_name,
572        } => AuditAction::Other {
573            detail: format!("email:sent:{subject} (msg={message_id}, tpl={template_name:?})"),
574        },
575        KernelEvent::KnowledgePersisted {
576            session_id,
577            message_index,
578            path,
579            source,
580        } => AuditAction::Other {
581            detail: format!("knowledge:persisted:{session_id}:{message_index}:{path}:{source}"),
582        },
583        KernelEvent::KnowledgeRemoved {
584            session_id,
585            message_index,
586        } => AuditAction::Other {
587            detail: format!("knowledge:removed:{session_id}:{message_index}"),
588        },
589        KernelEvent::AskUserRequest { id, question, .. } => AuditAction::Other {
590            detail: format!("ask_user:{id}:{question}"),
591        },
592        KernelEvent::PersonaCreated {
593            id, name, source, ..
594        } => AuditAction::Other {
595            detail: format!("persona:created:{id}:{name}:{source}"),
596        },
597        KernelEvent::PersonaUpdated { id, name, source } => AuditAction::Other {
598            detail: format!("persona:updated:{id}:{name}:{source}"),
599        },
600        KernelEvent::IntegrationInstallStarted {
601            job_id,
602            integration_id,
603            ..
604        } => AuditAction::Other {
605            detail: format!("install:started:{integration_id}:{job_id}"),
606        },
607        KernelEvent::IntegrationInstallProgress {
608            job_id,
609            integration_id,
610            ..
611        } => AuditAction::Other {
612            detail: format!("install:progress:{integration_id}:{job_id}"),
613        },
614        KernelEvent::IntegrationInstallCompleted {
615            job_id,
616            integration_id,
617            exit_code,
618            ..
619        } => AuditAction::Other {
620            detail: format!(
621                "install:completed:{integration_id}:{job_id}:exit={}",
622                exit_code
623                    .map(|c| c.to_string())
624                    .unwrap_or_else(|| "?".into())
625            ),
626        },
627        KernelEvent::IntegrationInstallFailed {
628            job_id,
629            integration_id,
630            ..
631        } => AuditAction::Other {
632            detail: format!("install:failed:{integration_id}:{job_id}"),
633        },
634        KernelEvent::CompressionDelta { .. }
635        | KernelEvent::CompressionDone { .. }
636        | KernelEvent::CompressionFailed { .. } => AuditAction::Other {
637            detail: "compression".to_string(),
638        },
639    }
640}
641
642/// Extract agent ID from a KernelEvent variant.
643fn extract_agent_id(event: &KernelEvent) -> String {
644    match event {
645        KernelEvent::AgentCreated { id, .. } => id.to_string(),
646        KernelEvent::AgentStarted { id, .. } => id.to_string(),
647        KernelEvent::AgentStopped { id, .. } => id.to_string(),
648        KernelEvent::AgentFailed { id, .. } => id.to_string(),
649        KernelEvent::MessageReceived { from, .. } => from.to_string(),
650        KernelEvent::AgentOutput { agent_id, .. } => agent_id.to_string(),
651        KernelEvent::AgentGroupMemberCompleted { agent_id, .. } => agent_id.to_string(),
652        KernelEvent::ProjectActivated { project_id, .. } => format!("project:{project_id}"),
653        // RFC-015: session-scoped events use session_id as the subject
654        KernelEvent::ToolExecutionStarted { session_id, .. } => format!("session:{session_id}"),
655        KernelEvent::ToolExecutionFinished { session_id, .. } => format!("session:{session_id}"),
656        KernelEvent::ToolExecutionProgress { session_id, .. } => format!("session:{session_id}"),
657        KernelEvent::MemoryRecallUsed { session_id, .. } => format!("session:{session_id}"),
658        KernelEvent::TokenUsageUpdate { session_id, .. } => format!("session:{session_id}"),
659        KernelEvent::ReasoningFragment { session_id, .. } => format!("session:{session_id}"),
660        KernelEvent::ToolArgsDelta { session_id, .. } => format!("session:{session_id}"),
661        KernelEvent::KnowledgePersisted { session_id, .. } => format!("session:{session_id}"),
662        KernelEvent::KnowledgeRemoved { session_id, .. } => format!("session:{session_id}"),
663        KernelEvent::CompactionTriggered { session_id, .. } => session_id
664            .as_ref()
665            .map(|s| format!("session:{s}"))
666            .unwrap_or_else(|| "system".to_string()),
667        _ => "system".to_string(),
668    }
669}
670
671/// Subscribe the audit trail to all kernel events.
672///
673/// The bus is broadcast-based; this spawns a long-running task that
674/// forwards every event into the audit trail as a structured entry.
675/// Lagged subscribers are logged and recovered.
676pub fn attach_audit_trail(bus: &EventBus, audit: Arc<AuditTrail>) {
677    let mut rx = bus.subscribe();
678    tokio::spawn(async move {
679        loop {
680            match rx.recv().await {
681                Ok(event) => {
682                    // Skip high-frequency streaming variants that are UI data,
683                    // not audit-relevant. ToolCallDelta fires per-token (raw
684                    // JSON fragments) — appending each would flood the Merkle
685                    // chain + JSONL with partial-JSON Debug strings per call.
686                    if matches!(event, KernelEvent::ToolArgsDelta { .. }) {
687                        continue;
688                    }
689                    let actor = extract_agent_id(&event);
690                    let action = kernel_event_to_audit_action(&event);
691                    let resource = format!("{event:?}");
692                    audit.append(actor, action, resource);
693                }
694                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
695                    // Surface the drop as a metric so operators can detect
696                    // incomplete audit trails instead of the events
697                    // vanishing silently (state-area F4).
698                    crate::metrics::get_metrics().audit_lagged_events.inc_by(n);
699                    tracing::warn!(
700                        skipped = n,
701                        "Audit trail subscriber lagged, skipping events"
702                    );
703                    continue;
704                }
705                Err(tokio::sync::broadcast::error::RecvError::Closed) => {
706                    tracing::info!("Audit trail event bus closed, exiting");
707                    break;
708                }
709            }
710        }
711    });
712}
713
714#[cfg(test)]
715mod tests {
716    use super::*;
717
718    fn sample_event(name: &str) -> KernelEvent {
719        KernelEvent::AgentCreated {
720            id: AgentId::new_v4(),
721            name: name.to_string(),
722        }
723    }
724
725    #[test]
726    fn test_event_bus_uses_sdk() {
727        let bus: EventBus = EventBus::new(256);
728        assert!(format!("{:?}", bus).contains("EventBus"));
729    }
730
731    #[tokio::test]
732    async fn test_publish_no_subscribers_ok() {
733        let bus = EventBus::new(16);
734        let result = bus.publish(sample_event("orphan"));
735        assert!(result.is_ok());
736    }
737
738    #[tokio::test]
739    async fn test_single_subscriber_receives_event() {
740        let bus = EventBus::new(16);
741        let mut rx = bus.subscribe();
742
743        let event = sample_event("test-agent");
744        bus.publish(event.clone()).unwrap();
745
746        let received = rx.try_recv().expect("should receive event");
747        match received {
748            KernelEvent::AgentCreated { name, .. } => assert_eq!(name, "test-agent"),
749            _ => panic!("wrong event type"),
750        }
751    }
752
753    #[tokio::test]
754    async fn test_multiple_subscribers_receive_events() {
755        let bus = EventBus::new(16);
756        let mut rx1 = bus.subscribe();
757        let mut rx2 = bus.subscribe();
758
759        let event = sample_event("multi");
760        bus.publish(event.clone()).unwrap();
761
762        let r1 = rx1.try_recv().expect("rx1 should receive event");
763        let r2 = rx2.try_recv().expect("rx2 should receive event");
764
765        assert!(matches!(r1, KernelEvent::AgentCreated { .. }));
766        assert!(matches!(r2, KernelEvent::AgentCreated { .. }));
767    }
768
769    #[tokio::test]
770    async fn test_kernel_event_to_audit_action() {
771        let event = KernelEvent::AgentFailed {
772            id: AgentId::new_v4(),
773            error: "boom".to_string(),
774        };
775        let action = kernel_event_to_audit_action(&event);
776        match action {
777            AuditAction::AgentExit { reason } => assert_eq!(reason, "boom"),
778            other => panic!("expected AgentExit, got {other:?}"),
779        }
780    }
781
782    // ── RFC-015 chat transparency event coverage ──
783
784    /// Round-trip JSON serialization for every new RFC-015 variant. This
785    /// guards against accidental renames that would break the WebSocket
786    /// wire format on the frontend.
787    #[test]
788    fn test_rfc015_event_round_trip_json() {
789        let cases: Vec<KernelEvent> = vec![
790            KernelEvent::ToolExecutionStarted {
791                session_id: "s1".into(),
792                tool_name: "read_file".into(),
793                tool_call_id: "call_1".into(),
794                tool_args: serde_json::json!({"path": "/src/main.rs"}),
795                context: None,
796            },
797            KernelEvent::ToolExecutionFinished {
798                session_id: "s1".into(),
799                tool_call_id: "call_1".into(),
800                tool_name: "read_file".into(),
801                duration_ms: 234,
802                is_error: false,
803                output_summary: "fn main() {}".into(),
804            },
805            KernelEvent::ToolExecutionProgress {
806                session_id: "s1".into(),
807                tool_call_id: "call_1".into(),
808                tool_name: "read_file".into(),
809                progress: "reading line 42/100".into(),
810                tab_id: None,
811                context: None,
812            },
813            KernelEvent::MemoryRecallUsed {
814                session_id: "s1".into(),
815                query: "rust errors".into(),
816                count: 3,
817                source: "warm".into(),
818            },
819            KernelEvent::TokenUsageUpdate {
820                session_id: "s1".into(),
821                input_tokens: 1234,
822                output_tokens: 567,
823            },
824            KernelEvent::ReasoningFragment {
825                session_id: "s1".into(),
826                content: "compaction done".into(),
827                source: "compaction".into(),
828            },
829        ];
830        for event in cases {
831            let json = serde_json::to_string(&event).expect("serialize");
832            let back: KernelEvent = serde_json::from_str(&json).expect("deserialize");
833            let json2 = serde_json::to_string(&back).expect("serialize round-trip");
834            assert_eq!(json, json2, "round-trip should be stable");
835        }
836    }
837
838    /// Tool progress events serialize/deserialize cleanly and round-trip
839    /// stable JSON, matching the wire format the WS layer expects.
840    #[test]
841    fn test_tool_execution_progress_serde_round_trip() {
842        let event = KernelEvent::ToolExecutionProgress {
843            session_id: "s-abc".into(),
844            tool_call_id: "call_42".into(),
845            tool_name: "browse".into(),
846            progress: "loading https://example.com".into(),
847            tab_id: Some(Uuid::new_v4()),
848            context: None,
849        };
850        let json = serde_json::to_string(&event).expect("serialize");
851        let back: KernelEvent = serde_json::from_str(&json).expect("deserialize");
852        match back {
853            KernelEvent::ToolExecutionProgress {
854                ref session_id,
855                ref tool_call_id,
856                ref tool_name,
857                ref progress,
858                tab_id,
859                ..
860            } => {
861                assert_eq!(session_id, "s-abc");
862                assert_eq!(tool_call_id, "call_42");
863                assert_eq!(tool_name, "browse");
864                assert_eq!(progress, "loading https://example.com");
865                assert!(tab_id.is_some(), "tab_id should round-trip when present");
866            }
867            other => panic!("expected ToolExecutionProgress, got {other:?}"),
868        }
869    }
870
871    /// The audit-action mapping for tool progress should produce a stable,
872    /// searchable detail string (used by the audit-trail UI to filter).
873    /// When `tab_id` is set, the detail includes `:tab=<id>`; when absent,
874    /// the original `tool_progress:<tool>` form is preserved (back-compat
875    /// for older oxi-agent versions that don't propagate tabs).
876    #[test]
877    fn test_tool_execution_progress_audit_action() {
878        let with_tab = KernelEvent::ToolExecutionProgress {
879            session_id: "s1".into(),
880            tool_call_id: "c1".into(),
881            tool_name: "browse".into(),
882            progress: "navigating".into(),
883            tab_id: Some(Uuid::new_v4()),
884            context: None,
885        };
886        match kernel_event_to_audit_action(&with_tab) {
887            AuditAction::Other { detail } => {
888                assert!(detail.contains("tool_progress"), "detail: {detail}");
889                assert!(detail.contains("browse"), "detail: {detail}");
890                assert!(
891                    detail.contains(":tab="),
892                    "detail should include tab id: {detail}"
893                );
894            }
895            other => panic!("expected Other, got {other:?}"),
896        }
897        let without_tab = KernelEvent::ToolExecutionProgress {
898            session_id: "s1".into(),
899            tool_call_id: "c1".into(),
900            tool_name: "browse".into(),
901            progress: "navigating".into(),
902            tab_id: None,
903            context: None,
904        };
905        match kernel_event_to_audit_action(&without_tab) {
906            AuditAction::Other { detail } => {
907                assert_eq!(detail, "tool_progress:browse");
908            }
909            other => panic!("expected Other, got {other:?}"),
910        }
911    }
912
913    /// `tab_id` is optional in serde (`#[serde(default)]`) so older oxi-agent
914    /// versions that don't emit it still round-trip cleanly. This guards the
915    /// backwards-compat contract explicitly.
916    #[test]
917    fn test_tool_execution_progress_tab_id_optional_in_serde() {
918        // Simulate a payload from a legacy oxi-agent (no tab_id key).
919        // KernelEvent is externally tagged, so the variant is the JSON key.
920        let legacy_json = r#"{
921            "ToolExecutionProgress": {
922                "session_id": "s-old",
923                "tool_call_id": "call_legacy",
924                "tool_name": "browse",
925                "progress": "step 1"
926            }
927        }"#;
928        let event: KernelEvent = serde_json::from_str(legacy_json).expect("deserialize legacy");
929        match &event {
930            KernelEvent::ToolExecutionProgress {
931                session_id,
932                tool_call_id,
933                tool_name,
934                progress,
935                tab_id,
936                ..
937            } => {
938                assert_eq!(session_id, "s-old");
939                assert_eq!(tool_call_id, "call_legacy");
940                assert_eq!(tool_name, "browse");
941                assert_eq!(progress, "step 1");
942                assert!(tab_id.is_none(), "missing field should default to None");
943            }
944            other => panic!("expected ToolExecutionProgress, got {other:?}"),
945        }
946        // And re-serialise — `skip_serializing_if = "Option::is_none"` keeps
947        // the wire format clean when downstream tools don't set tab_id.
948        let json = serde_json::to_string(&event).expect("serialize");
949        assert!(
950            !json.contains("tab_id"),
951            "tab_id should be omitted when None: {json}"
952        );
953    }
954
955    /// The agent_id extractor should map session-scoped RFC-015 events to
956    /// `session:<id>` for audit-trail grouping, while non-session events
957    /// keep their existing behaviour.
958    #[test]
959    fn test_rfc015_extract_agent_id() {
960        let event = KernelEvent::ToolExecutionStarted {
961            session_id: "abc-123".into(),
962            tool_name: "bash".into(),
963            tool_call_id: "c1".into(),
964            tool_args: serde_json::Value::Null,
965            context: None,
966        };
967        // The function is private; verify via the public AuditAction mapping
968        // that session-scoped events do not collide with real agent ids.
969        let action = kernel_event_to_audit_action(&event);
970        match action {
971            AuditAction::Other { detail } => {
972                assert!(
973                    detail.contains("bash"),
974                    "tool name in audit detail: {detail}"
975                );
976            }
977            other => panic!("expected Other, got {other:?}"),
978        }
979    }
980}
oxios_kernel/event_bus.rs

oxios_kernel/
event_bus.rs
