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

oxios_kernel/
event_bus.rs
