ironflow_engine/notify/
event_bus.rs

1//! Per-workflow broadcast event bus for real-time monitoring.
2//!
3//! [`WorkflowEventBus`] maintains one [`tokio::sync::broadcast`] channel per
4//! workflow run. Consumers (dashboards, SSE routes) subscribe to a specific
5//! `run_id` and receive only the events for that run.
6//!
7//! # Architecture
8//!
9//! - [`WorkflowEvent`] -- granular step-level events (started, completed,
10//!   failed, approval, human input, token usage, agent session resume).
11//! - [`WorkflowEventBus`] -- per-run broadcast channels with subscribe /
12//!   publish / remove lifecycle.
13//!
14//! # Examples
15//!
16//! ```
17//! use ironflow_engine::notify::{WorkflowEvent, WorkflowEventBus, WorkflowStepStartedEvent};
18//! use uuid::Uuid;
19//! use chrono::Utc;
20//!
21//! let bus = WorkflowEventBus::new();
22//! let run_id = Uuid::now_v7();
23//!
24//! let mut rx = bus.subscribe(run_id);
25//!
26//! bus.publish(run_id, WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
27//!     step_name: "build".to_string(),
28//!     step_index: 0,
29//!     timestamp: Utc::now(),
30//! }));
31//! ```
32
33use std::collections::HashMap;
34use std::sync::RwLock;
35
36use chrono::{DateTime, Utc};
37use rust_decimal::Decimal;
38use serde::{Deserialize, Serialize};
39use serde_json::Value;
40use tokio::sync::broadcast;
41use uuid::Uuid;
42
43use super::{Event, EventSubscriber, SubscriberFuture};
44
45/// Default broadcast channel buffer size per run.
46const DEFAULT_BUFFER_SIZE: usize = 64;
47
48/// Payload of the `WorkflowEvent::StepStarted` workflow event.
49///
50/// # Examples
51///
52/// ```
53/// use chrono::Utc;
54/// use ironflow_engine::notify::WorkflowStepStartedEvent;
55///
56/// let payload = WorkflowStepStartedEvent {
57///     step_name: "build".to_string(),
58///     step_index: 0,
59///     timestamp: Utc::now(),
60/// };
61/// assert_eq!(payload.step_index, 0);
62/// ```
63#[derive(Debug, Clone, Serialize, Deserialize)]
64#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
65pub struct WorkflowStepStartedEvent {
66    /// Human-readable step name.
67    pub step_name: String,
68    /// Zero-based position in the workflow.
69    pub step_index: u32,
70    /// When the step started.
71    pub timestamp: DateTime<Utc>,
72}
73
74/// Payload of the `WorkflowEvent::StepCompleted` workflow event.
75///
76/// # Examples
77///
78/// ```
79/// use ironflow_engine::notify::WorkflowStepCompletedEvent;
80///
81/// let payload = WorkflowStepCompletedEvent {
82///     step_name: "deploy".to_string(),
83///     step_index: 1,
84///     duration_ms: 5000,
85///     output_summary: Some("deployed v1.2.3".to_string()),
86/// };
87/// assert_eq!(payload.duration_ms, 5000);
88/// ```
89#[derive(Debug, Clone, Serialize, Deserialize)]
90#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
91pub struct WorkflowStepCompletedEvent {
92    /// Human-readable step name.
93    pub step_name: String,
94    /// Zero-based position in the workflow.
95    pub step_index: u32,
96    /// Step duration in milliseconds.
97    pub duration_ms: u64,
98    /// Optional summary of the step output.
99    pub output_summary: Option<String>,
100}
101
102/// Payload of the `WorkflowEvent::StepFailed` workflow event.
103///
104/// # Examples
105///
106/// ```
107/// use ironflow_engine::notify::WorkflowStepFailedEvent;
108///
109/// let payload = WorkflowStepFailedEvent {
110///     step_name: "test".to_string(),
111///     step_index: 2,
112///     error: "exit code 1".to_string(),
113///     duration_ms: 3000,
114/// };
115/// assert_eq!(payload.error, "exit code 1");
116/// ```
117#[derive(Debug, Clone, Serialize, Deserialize)]
118#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
119pub struct WorkflowStepFailedEvent {
120    /// Human-readable step name.
121    pub step_name: String,
122    /// Zero-based position in the workflow.
123    pub step_index: u32,
124    /// Error description.
125    pub error: String,
126    /// Step duration in milliseconds.
127    pub duration_ms: u64,
128}
129
130/// Payload of the `WorkflowEvent::ApprovalRequired` workflow event.
131///
132/// # Examples
133///
134/// ```
135/// use ironflow_engine::notify::WorkflowApprovalRequiredEvent;
136/// use uuid::Uuid;
137///
138/// let payload = WorkflowApprovalRequiredEvent {
139///     step_name: "prod-gate".to_string(),
140///     step_index: 3,
141///     approval_id: Uuid::now_v7(),
142/// };
143/// assert_eq!(payload.step_name, "prod-gate");
144/// ```
145#[derive(Debug, Clone, Serialize, Deserialize)]
146#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
147pub struct WorkflowApprovalRequiredEvent {
148    /// Human-readable step name.
149    pub step_name: String,
150    /// Zero-based position in the workflow.
151    pub step_index: u32,
152    /// Identifier of the approval gate.
153    pub approval_id: Uuid,
154}
155
156/// Payload of the `WorkflowEvent::InputRequired` workflow event.
157///
158/// # Examples
159///
160/// ```
161/// use ironflow_engine::notify::WorkflowInputRequiredEvent;
162/// use serde_json::json;
163/// use uuid::Uuid;
164///
165/// let payload = WorkflowInputRequiredEvent {
166///     run_id: Uuid::now_v7(),
167///     step_id: Uuid::now_v7(),
168///     step_name: "clarify".to_string(),
169///     step_index: 2,
170///     message: "Answer the questions".to_string(),
171///     schema: json!({"type": "object"}),
172/// };
173/// assert_eq!(payload.step_name, "clarify");
174/// ```
175#[derive(Debug, Clone, Serialize, Deserialize)]
176#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
177pub struct WorkflowInputRequiredEvent {
178    /// The run waiting for the input.
179    pub run_id: Uuid,
180    /// Identifier of the human input step; the answer is posted to it.
181    pub step_id: Uuid,
182    /// Human-readable step name.
183    pub step_name: String,
184    /// Zero-based position in the workflow.
185    pub step_index: u32,
186    /// Message displayed to the person answering.
187    pub message: String,
188    /// JSON schema the answer must match.
189    #[cfg_attr(feature = "openapi", schema(value_type = Object))]
190    pub schema: Value,
191}
192
193/// Payload of the `WorkflowEvent::AgentStepTokensUsed` workflow event.
194///
195/// # Examples
196///
197/// ```
198/// use ironflow_engine::notify::WorkflowAgentStepTokensUsedEvent;
199/// use rust_decimal::Decimal;
200///
201/// let payload = WorkflowAgentStepTokensUsedEvent {
202///     step_name: "review".to_string(),
203///     tokens: 15_000,
204///     cost_usd: Decimal::new(42, 4),
205/// };
206/// assert_eq!(payload.tokens, 15_000);
207/// ```
208#[derive(Debug, Clone, Serialize, Deserialize)]
209#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
210pub struct WorkflowAgentStepTokensUsedEvent {
211    /// Human-readable step name.
212    pub step_name: String,
213    /// Total tokens consumed.
214    pub tokens: u64,
215    /// Estimated cost in USD.
216    pub cost_usd: Decimal,
217}
218
219/// Payload of the `WorkflowEvent::AgentStepResumed` workflow event.
220///
221/// Published when an agent step interrupted by a lost lease is re-executed
222/// on the Claude Code session of the interrupted attempt instead of starting
223/// from scratch.
224///
225/// # Examples
226///
227/// ```
228/// use ironflow_engine::notify::WorkflowAgentStepResumedEvent;
229/// use chrono::Utc;
230///
231/// let payload = WorkflowAgentStepResumedEvent {
232///     step_name: "review".to_string(),
233///     step_index: 2,
234///     session_id: "0192f0c1-7d2e-7a4b-9c3d-1e2f3a4b5c6d".to_string(),
235///     timestamp: Utc::now(),
236/// };
237/// assert_eq!(payload.step_index, 2);
238/// ```
239#[derive(Debug, Clone, Serialize, Deserialize)]
240#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
241pub struct WorkflowAgentStepResumedEvent {
242    /// Human-readable step name.
243    pub step_name: String,
244    /// Zero-based position in the workflow.
245    pub step_index: u32,
246    /// Claude Code session the step resumes.
247    pub session_id: String,
248    /// When the step resumed.
249    pub timestamp: DateTime<Utc>,
250}
251
252/// A granular step-level event for real-time workflow monitoring.
253///
254/// Unlike [`Event`](super::Event) which covers the full system lifecycle
255/// (runs, auth, audit), `WorkflowEvent` tracks individual step transitions
256/// within a single run. Serialized with a `type` discriminant for UI
257/// consumption.
258///
259/// Each variant wraps a dedicated payload struct; the serialized form stays
260/// flat, with `type` sitting next to the payload fields.
261///
262/// # Examples
263///
264/// ```
265/// use ironflow_engine::notify::{WorkflowEvent, WorkflowStepStartedEvent};
266/// use chrono::Utc;
267///
268/// let event = WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
269///     step_name: "deploy".to_string(),
270///     step_index: 0,
271///     timestamp: Utc::now(),
272/// });
273/// assert_eq!(event.event_type(), "step_started");
274///
275/// let json = serde_json::to_string(&event)?;
276/// assert!(json.contains("\"type\":\"step_started\""));
277/// # Ok::<(), serde_json::Error>(())
278/// ```
279#[derive(Debug, Clone, Serialize, Deserialize)]
280#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
281#[serde(tag = "type", rename_all = "snake_case")]
282pub enum WorkflowEvent {
283    /// A step began execution.
284    StepStarted(WorkflowStepStartedEvent),
285
286    /// A step completed successfully.
287    StepCompleted(WorkflowStepCompletedEvent),
288
289    /// A step failed.
290    StepFailed(WorkflowStepFailedEvent),
291
292    /// A step requires human approval before the run can continue.
293    ApprovalRequired(WorkflowApprovalRequiredEvent),
294
295    /// A step waits for a typed human input before the run can continue.
296    InputRequired(WorkflowInputRequiredEvent),
297
298    /// Token usage report for an agent step.
299    AgentStepTokensUsed(WorkflowAgentStepTokensUsedEvent),
300
301    /// An interrupted agent step resumed its Claude Code session.
302    AgentStepResumed(WorkflowAgentStepResumedEvent),
303}
304
305impl WorkflowEvent {
306    /// Event type constant for [`StepStarted`](WorkflowEvent::StepStarted).
307    pub const STEP_STARTED: &'static str = "step_started";
308    /// Event type constant for [`StepCompleted`](WorkflowEvent::StepCompleted).
309    pub const STEP_COMPLETED: &'static str = "step_completed";
310    /// Event type constant for [`StepFailed`](WorkflowEvent::StepFailed).
311    pub const STEP_FAILED: &'static str = "step_failed";
312    /// Event type constant for [`ApprovalRequired`](WorkflowEvent::ApprovalRequired).
313    pub const APPROVAL_REQUIRED: &'static str = "approval_required";
314    /// Event type constant for [`InputRequired`](WorkflowEvent::InputRequired).
315    pub const INPUT_REQUIRED: &'static str = "input_required";
316    /// Event type constant for [`AgentStepTokensUsed`](WorkflowEvent::AgentStepTokensUsed).
317    pub const AGENT_STEP_TOKENS_USED: &'static str = "agent_step_tokens_used";
318    /// Event type constant for [`AgentStepResumed`](WorkflowEvent::AgentStepResumed).
319    pub const AGENT_STEP_RESUMED: &'static str = "agent_step_resumed";
320
321    /// Returns the event type as a static string (e.g. `"step_started"`).
322    ///
323    /// # Examples
324    ///
325    /// ```
326    /// use ironflow_engine::notify::{WorkflowEvent, WorkflowStepStartedEvent};
327    /// use chrono::Utc;
328    ///
329    /// let event = WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
330    ///     step_name: "build".to_string(),
331    ///     step_index: 0,
332    ///     timestamp: Utc::now(),
333    /// });
334    /// assert_eq!(event.event_type(), "step_started");
335    /// ```
336    #[deny(unreachable_patterns)]
337    pub fn event_type(&self) -> &'static str {
338        match self {
339            WorkflowEvent::StepStarted(_) => Self::STEP_STARTED,
340            WorkflowEvent::StepCompleted(_) => Self::STEP_COMPLETED,
341            WorkflowEvent::StepFailed(_) => Self::STEP_FAILED,
342            WorkflowEvent::ApprovalRequired(_) => Self::APPROVAL_REQUIRED,
343            WorkflowEvent::InputRequired(_) => Self::INPUT_REQUIRED,
344            WorkflowEvent::AgentStepTokensUsed(_) => Self::AGENT_STEP_TOKENS_USED,
345            WorkflowEvent::AgentStepResumed(_) => Self::AGENT_STEP_RESUMED,
346        }
347    }
348}
349
350/// Per-workflow broadcast event bus for real-time monitoring.
351///
352/// Maintains one [`tokio::sync::broadcast`] channel per workflow run.
353/// Consumers call [`subscribe`](Self::subscribe) to receive events for a
354/// specific run; producers call [`publish`](Self::publish) to broadcast
355/// an event to all subscribers of that run.
356///
357/// Thread-safe and cheaply cloneable (`Clone` shares the same inner state).
358///
359/// # Examples
360///
361/// ```
362/// use ironflow_engine::notify::{WorkflowEvent, WorkflowEventBus, WorkflowStepStartedEvent};
363/// use uuid::Uuid;
364/// use chrono::Utc;
365///
366/// let bus = WorkflowEventBus::new();
367/// let run_id = Uuid::now_v7();
368///
369/// let mut rx = bus.subscribe(run_id);
370/// bus.publish(run_id, WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
371///     step_name: "build".to_string(),
372///     step_index: 0,
373///     timestamp: Utc::now(),
374/// }));
375/// ```
376#[derive(Clone)]
377pub struct WorkflowEventBus {
378    channels: std::sync::Arc<RwLock<HashMap<Uuid, broadcast::Sender<WorkflowEvent>>>>,
379}
380
381impl WorkflowEventBus {
382    /// Create a new empty event bus.
383    ///
384    /// # Examples
385    ///
386    /// ```
387    /// use ironflow_engine::notify::WorkflowEventBus;
388    ///
389    /// let bus = WorkflowEventBus::new();
390    /// ```
391    pub fn new() -> Self {
392        Self {
393            channels: std::sync::Arc::new(RwLock::new(HashMap::new())),
394        }
395    }
396
397    /// Subscribe to events for a specific workflow run.
398    ///
399    /// If no channel exists for this `run_id`, one is created on demand.
400    /// Returns a broadcast receiver that yields [`WorkflowEvent`]s for
401    /// that run only.
402    ///
403    /// # Examples
404    ///
405    /// ```
406    /// use ironflow_engine::notify::WorkflowEventBus;
407    /// use uuid::Uuid;
408    ///
409    /// let bus = WorkflowEventBus::new();
410    /// let run_id = Uuid::now_v7();
411    /// let _rx = bus.subscribe(run_id);
412    /// ```
413    pub fn subscribe(&self, run_id: Uuid) -> broadcast::Receiver<WorkflowEvent> {
414        let mut channels = self.channels.write().expect("event bus lock poisoned");
415        let sender = channels
416            .entry(run_id)
417            .or_insert_with(|| broadcast::channel(DEFAULT_BUFFER_SIZE).0);
418        sender.subscribe()
419    }
420
421    /// Broadcast an event to all subscribers of a specific workflow run.
422    ///
423    /// If no channel exists for `run_id` (no subscriber has called
424    /// [`subscribe`](Self::subscribe)), the event is silently dropped.
425    /// If subscribers exist but none are actively listening, the send
426    /// error is ignored.
427    ///
428    /// # Examples
429    ///
430    /// ```
431    /// use ironflow_engine::notify::{WorkflowEvent, WorkflowEventBus, WorkflowStepStartedEvent};
432    /// use uuid::Uuid;
433    /// use chrono::Utc;
434    ///
435    /// let bus = WorkflowEventBus::new();
436    /// let run_id = Uuid::now_v7();
437    ///
438    /// // No subscriber -- silently dropped.
439    /// bus.publish(run_id, WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
440    ///     step_name: "build".to_string(),
441    ///     step_index: 0,
442    ///     timestamp: Utc::now(),
443    /// }));
444    /// ```
445    pub fn publish(&self, run_id: Uuid, event: WorkflowEvent) {
446        let channels = self.channels.read().expect("event bus lock poisoned");
447        if let Some(sender) = channels.get(&run_id) {
448            let _ = sender.send(event);
449        }
450    }
451
452    /// Remove the channel for a workflow run.
453    ///
454    /// Called by the engine when a run reaches a terminal status, once the
455    /// bus is attached with [`Engine::set_event_bus`](crate::engine::Engine::set_event_bus).
456    /// Dropping the sender lets every receiver drain the events already
457    /// buffered, then get [`broadcast::error::RecvError::Closed`].
458    /// If no channel exists for `run_id`, this is a no-op.
459    ///
460    /// # Examples
461    ///
462    /// ```
463    /// use ironflow_engine::notify::WorkflowEventBus;
464    /// use uuid::Uuid;
465    ///
466    /// let bus = WorkflowEventBus::new();
467    /// let run_id = Uuid::now_v7();
468    /// let _rx = bus.subscribe(run_id);
469    /// bus.remove(run_id);
470    /// ```
471    pub fn remove(&self, run_id: Uuid) {
472        let mut channels = self.channels.write().expect("event bus lock poisoned");
473        channels.remove(&run_id);
474    }
475
476    /// Whether a channel is currently open for a workflow run.
477    ///
478    /// # Panics
479    ///
480    /// Panics if the bus lock was poisoned by a panic in another thread.
481    ///
482    /// # Examples
483    ///
484    /// ```
485    /// use ironflow_engine::notify::WorkflowEventBus;
486    /// use uuid::Uuid;
487    ///
488    /// let bus = WorkflowEventBus::new();
489    /// let run_id = Uuid::now_v7();
490    /// assert!(!bus.has_channel(run_id));
491    /// let _rx = bus.subscribe(run_id);
492    /// assert!(bus.has_channel(run_id));
493    /// bus.remove(run_id);
494    /// assert!(!bus.has_channel(run_id));
495    /// ```
496    pub fn has_channel(&self, run_id: Uuid) -> bool {
497        let channels = self.channels.read().expect("event bus lock poisoned");
498        channels.contains_key(&run_id)
499    }
500}
501
502impl Default for WorkflowEventBus {
503    fn default() -> Self {
504        Self::new()
505    }
506}
507
508/// [`EventSubscriber`] that releases the [`WorkflowEventBus`] channel of a run
509/// once the run reaches a terminal status.
510///
511/// Terminal statuses are published from several places (engine, cancel,
512/// reaper, worker status updates), all through
513/// [`Event::RunStatusChanged`], so one subscriber covers them all.
514pub(crate) struct EventBusReleaser {
515    bus: WorkflowEventBus,
516}
517
518impl EventBusReleaser {
519    /// Build a releaser for the given bus.
520    pub(crate) fn new(bus: WorkflowEventBus) -> Self {
521        Self { bus }
522    }
523}
524
525impl EventSubscriber for EventBusReleaser {
526    fn name(&self) -> &str {
527        "event-bus-releaser"
528    }
529
530    fn handle<'a>(&'a self, event: &'a Event) -> SubscriberFuture<'a> {
531        Box::pin(async move {
532            if let Event::RunStatusChanged(change) = event
533                && change.to.is_terminal()
534            {
535                self.bus.remove(change.run_id);
536            }
537        })
538    }
539}
540
541impl std::fmt::Debug for WorkflowEventBus {
542    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
543        let count = self.channels.read().map(|c| c.len()).unwrap_or(0);
544        f.debug_struct("WorkflowEventBus")
545            .field("active_channels", &count)
546            .finish()
547    }
548}
549
550#[cfg(test)]
551mod tests {
552    use ironflow_store::models::RunStatus;
553    use serde_json::json;
554    use tokio::sync::broadcast::error::RecvError;
555
556    use crate::notify::RunStatusChangedEvent;
557
558    use super::*;
559
560    fn step_started(step_name: &str) -> WorkflowEvent {
561        WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
562            step_name: step_name.to_string(),
563            step_index: 0,
564            timestamp: Utc::now(),
565        })
566    }
567
568    #[tokio::test]
569    async fn subscribe_receives_published_events() {
570        let bus = WorkflowEventBus::new();
571        let run_id = Uuid::now_v7();
572
573        let mut rx = bus.subscribe(run_id);
574
575        bus.publish(run_id, step_started("build"));
576
577        let received = rx.recv().await.expect("should receive event");
578        assert_eq!(received.event_type(), "step_started");
579        match received {
580            WorkflowEvent::StepStarted(e) => {
581                assert_eq!(e.step_name, "build");
582                assert_eq!(e.step_index, 0);
583            }
584            _ => panic!("expected StepStarted"),
585        }
586    }
587
588    #[test]
589    fn subscribe_creates_channel_on_demand() {
590        let bus = WorkflowEventBus::new();
591        let run_id = Uuid::now_v7();
592
593        let count_before = bus.channels.read().unwrap().len();
594        assert_eq!(count_before, 0);
595
596        let _rx = bus.subscribe(run_id);
597
598        let count_after = bus.channels.read().unwrap().len();
599        assert_eq!(count_after, 1);
600    }
601
602    #[test]
603    fn publish_unknown_run_is_noop() {
604        let bus = WorkflowEventBus::new();
605        let unknown_run = Uuid::now_v7();
606
607        bus.publish(unknown_run, step_started("build"));
608    }
609
610    #[test]
611    fn remove_cleans_up_channel() {
612        let bus = WorkflowEventBus::new();
613        let run_id = Uuid::now_v7();
614
615        let _rx = bus.subscribe(run_id);
616        assert_eq!(bus.channels.read().unwrap().len(), 1);
617
618        bus.remove(run_id);
619        assert_eq!(bus.channels.read().unwrap().len(), 0);
620    }
621
622    #[test]
623    fn remove_unknown_is_noop() {
624        let bus = WorkflowEventBus::new();
625        bus.remove(Uuid::now_v7());
626    }
627
628    #[test]
629    fn workflow_event_serde_roundtrip() {
630        let cases: Vec<WorkflowEvent> = vec![
631            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
632                step_name: "build".to_string(),
633                step_index: 0,
634                timestamp: Utc::now(),
635            }),
636            WorkflowEvent::StepCompleted(WorkflowStepCompletedEvent {
637                step_name: "deploy".to_string(),
638                step_index: 1,
639                duration_ms: 5000,
640                output_summary: Some("deployed v1.2.3".to_string()),
641            }),
642            WorkflowEvent::StepFailed(WorkflowStepFailedEvent {
643                step_name: "test".to_string(),
644                step_index: 2,
645                error: "exit code 1".to_string(),
646                duration_ms: 3000,
647            }),
648            WorkflowEvent::ApprovalRequired(WorkflowApprovalRequiredEvent {
649                step_name: "prod-gate".to_string(),
650                step_index: 3,
651                approval_id: Uuid::now_v7(),
652            }),
653            WorkflowEvent::InputRequired(WorkflowInputRequiredEvent {
654                run_id: Uuid::now_v7(),
655                step_id: Uuid::now_v7(),
656                step_name: "clarify".to_string(),
657                step_index: 4,
658                message: "Answer the questions".to_string(),
659                schema: json!({"type": "object"}),
660            }),
661            WorkflowEvent::AgentStepTokensUsed(WorkflowAgentStepTokensUsedEvent {
662                step_name: "review".to_string(),
663                tokens: 15000,
664                cost_usd: Decimal::new(42, 4),
665            }),
666            WorkflowEvent::AgentStepResumed(WorkflowAgentStepResumedEvent {
667                step_name: "review".to_string(),
668                step_index: 5,
669                session_id: "0192f0c1-7d2e-7a4b-9c3d-1e2f3a4b5c6d".to_string(),
670                timestamp: Utc::now(),
671            }),
672        ];
673
674        for event in &cases {
675            let json = serde_json::to_string(event).expect("serialize");
676            let back: WorkflowEvent = serde_json::from_str(&json).expect("deserialize");
677
678            assert_eq!(back.event_type(), event.event_type());
679            assert!(json.contains(&format!("\"type\":\"{}\"", event.event_type())));
680        }
681    }
682
683    /// The pre-refactor wire format used flat inline-struct variants. Newtype
684    /// variants produce and accept the same JSON, so SSE consumers and stored
685    /// payloads need no migration.
686    #[test]
687    fn workflow_event_legacy_flat_json_deserializes() {
688        let approval_id: Uuid = "01890000-0000-7000-8000-000000000002"
689            .parse()
690            .expect("valid uuid");
691
692        let raw = r#"{"type":"step_started","step_name":"build","step_index":0,"timestamp":"2026-01-01T00:00:00Z"}"#;
693        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
694            WorkflowEvent::StepStarted(e) => {
695                assert_eq!(e.step_name, "build");
696                assert_eq!(e.step_index, 0);
697            }
698            other => panic!("expected StepStarted, got {other:?}"),
699        }
700
701        let raw = r#"{"type":"step_completed","step_name":"deploy","step_index":1,"duration_ms":5000,"output_summary":"deployed v1.2.3"}"#;
702        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
703            WorkflowEvent::StepCompleted(e) => {
704                assert_eq!(e.duration_ms, 5000);
705                assert_eq!(e.output_summary.as_deref(), Some("deployed v1.2.3"));
706            }
707            other => panic!("expected StepCompleted, got {other:?}"),
708        }
709
710        let raw = r#"{"type":"step_failed","step_name":"test","step_index":2,"error":"exit code 1","duration_ms":3000}"#;
711        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
712            WorkflowEvent::StepFailed(e) => {
713                assert_eq!(e.error, "exit code 1");
714                assert_eq!(e.duration_ms, 3000);
715            }
716            other => panic!("expected StepFailed, got {other:?}"),
717        }
718
719        let raw = r#"{"type":"approval_required","step_name":"prod-gate","step_index":3,"approval_id":"01890000-0000-7000-8000-000000000002"}"#;
720        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
721            WorkflowEvent::ApprovalRequired(e) => {
722                assert_eq!(e.approval_id, approval_id);
723            }
724            other => panic!("expected ApprovalRequired, got {other:?}"),
725        }
726
727        let raw = r#"{"type":"agent_step_tokens_used","step_name":"review","tokens":15000,"cost_usd":0.5}"#;
728        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
729            WorkflowEvent::AgentStepTokensUsed(e) => {
730                assert_eq!(e.tokens, 15000);
731                assert_eq!(e.cost_usd, Decimal::new(5, 1));
732            }
733            other => panic!("expected AgentStepTokensUsed, got {other:?}"),
734        }
735    }
736
737    /// Guards the internally-tagged representation: payload fields must stay
738    /// siblings of `type`, never nested under a variant key.
739    #[test]
740    fn serialized_workflow_event_is_flat_with_type_tag() {
741        let event = WorkflowEvent::StepFailed(WorkflowStepFailedEvent {
742            step_name: "test".to_string(),
743            step_index: 2,
744            error: "exit code 1".to_string(),
745            duration_ms: 3000,
746        });
747
748        let value: serde_json::Value = serde_json::to_value(&event).expect("serialize");
749        let object = value.as_object().expect("event serializes to an object");
750
751        assert_eq!(
752            object.get("type").and_then(|v| v.as_str()),
753            Some("step_failed")
754        );
755        assert_eq!(
756            object.get("step_name").and_then(|v| v.as_str()),
757            Some("test")
758        );
759        assert_eq!(object.get("step_index").and_then(|v| v.as_u64()), Some(2));
760        assert_eq!(
761            object.get("error").and_then(|v| v.as_str()),
762            Some("exit code 1")
763        );
764        assert_eq!(
765            object.get("duration_ms").and_then(|v| v.as_u64()),
766            Some(3000)
767        );
768        assert_eq!(object.len(), 5, "no nesting: {object:?}");
769    }
770
771    #[test]
772    fn input_required_event_serializes_flat_with_its_schema() {
773        let step_id = Uuid::now_v7();
774        let event = WorkflowEvent::InputRequired(WorkflowInputRequiredEvent {
775            run_id: Uuid::now_v7(),
776            step_id,
777            step_name: "clarify".to_string(),
778            step_index: 1,
779            message: "Answer the questions".to_string(),
780            schema: json!({"type": "object", "required": ["answers"]}),
781        });
782
783        let value = serde_json::to_value(&event).expect("serialize");
784        assert_eq!(value["type"], "input_required");
785        assert_eq!(value["step_id"], step_id.to_string());
786        assert_eq!(value["message"], "Answer the questions");
787        assert_eq!(value["schema"]["required"][0], "answers");
788        assert_eq!(event.event_type(), WorkflowEvent::INPUT_REQUIRED);
789    }
790
791    #[test]
792    fn event_type_all_variants() {
793        let cases: Vec<(WorkflowEvent, &str)> = vec![
794            (
795                WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
796                    step_name: "s".to_string(),
797                    step_index: 0,
798                    timestamp: Utc::now(),
799                }),
800                "step_started",
801            ),
802            (
803                WorkflowEvent::StepCompleted(WorkflowStepCompletedEvent {
804                    step_name: "s".to_string(),
805                    step_index: 0,
806                    duration_ms: 0,
807                    output_summary: None,
808                }),
809                "step_completed",
810            ),
811            (
812                WorkflowEvent::StepFailed(WorkflowStepFailedEvent {
813                    step_name: "s".to_string(),
814                    step_index: 0,
815                    error: "e".to_string(),
816                    duration_ms: 0,
817                }),
818                "step_failed",
819            ),
820            (
821                WorkflowEvent::ApprovalRequired(WorkflowApprovalRequiredEvent {
822                    step_name: "s".to_string(),
823                    step_index: 0,
824                    approval_id: Uuid::now_v7(),
825                }),
826                "approval_required",
827            ),
828            (
829                WorkflowEvent::InputRequired(WorkflowInputRequiredEvent {
830                    run_id: Uuid::now_v7(),
831                    step_id: Uuid::now_v7(),
832                    step_name: "s".to_string(),
833                    step_index: 0,
834                    message: "m".to_string(),
835                    schema: Value::Null,
836                }),
837                "input_required",
838            ),
839            (
840                WorkflowEvent::AgentStepTokensUsed(WorkflowAgentStepTokensUsedEvent {
841                    step_name: "s".to_string(),
842                    tokens: 0,
843                    cost_usd: Decimal::ZERO,
844                }),
845                "agent_step_tokens_used",
846            ),
847            (
848                WorkflowEvent::AgentStepResumed(WorkflowAgentStepResumedEvent {
849                    step_name: "s".to_string(),
850                    step_index: 0,
851                    session_id: "sid".to_string(),
852                    timestamp: Utc::now(),
853                }),
854                "agent_step_resumed",
855            ),
856        ];
857
858        for (event, expected) in cases {
859            assert_eq!(event.event_type(), expected);
860        }
861    }
862
863    #[tokio::test]
864    async fn multiple_subscribers_receive_same_event() {
865        let bus = WorkflowEventBus::new();
866        let run_id = Uuid::now_v7();
867
868        let mut rx1 = bus.subscribe(run_id);
869        let mut rx2 = bus.subscribe(run_id);
870
871        bus.publish(run_id, step_started("build"));
872
873        let e1 = rx1.recv().await.expect("rx1 should receive");
874        let e2 = rx2.recv().await.expect("rx2 should receive");
875
876        assert_eq!(e1.event_type(), "step_started");
877        assert_eq!(e2.event_type(), "step_started");
878    }
879
880    #[tokio::test]
881    async fn events_isolated_between_runs() {
882        let bus = WorkflowEventBus::new();
883        let run_a = Uuid::now_v7();
884        let run_b = Uuid::now_v7();
885
886        let mut rx_a = bus.subscribe(run_a);
887        let mut rx_b = bus.subscribe(run_b);
888
889        bus.publish(run_a, step_started("only-for-a"));
890
891        let received = rx_a.recv().await.expect("rx_a should receive");
892        match received {
893            WorkflowEvent::StepStarted(e) => {
894                assert_eq!(e.step_name, "only-for-a");
895            }
896            _ => panic!("expected StepStarted"),
897        }
898
899        // rx_b should have nothing -- try_recv returns Empty.
900        assert!(rx_b.try_recv().is_err());
901    }
902
903    fn status_changed(run_id: Uuid, to: RunStatus) -> Event {
904        Event::RunStatusChanged(RunStatusChangedEvent {
905            run_id,
906            workflow_name: "wf".to_string(),
907            from: RunStatus::Running,
908            to,
909            error: None,
910            cost_usd: Decimal::ZERO,
911            duration_ms: 0,
912            labels: HashMap::new(),
913            at: Utc::now(),
914        })
915    }
916
917    #[test]
918    fn has_channel_tracks_subscribe_and_remove() {
919        let bus = WorkflowEventBus::new();
920        let run_id = Uuid::now_v7();
921
922        assert!(!bus.has_channel(run_id));
923        let _rx = bus.subscribe(run_id);
924        assert!(bus.has_channel(run_id));
925        bus.remove(run_id);
926        assert!(!bus.has_channel(run_id));
927    }
928
929    #[tokio::test]
930    async fn remove_lets_receivers_drain_then_close() {
931        let bus = WorkflowEventBus::new();
932        let run_id = Uuid::now_v7();
933        let mut rx = bus.subscribe(run_id);
934
935        bus.publish(run_id, step_started("build"));
936        bus.remove(run_id);
937
938        let received = rx.recv().await.expect("buffered event survives removal");
939        assert_eq!(received.event_type(), "step_started");
940        assert!(matches!(rx.recv().await, Err(RecvError::Closed)));
941    }
942
943    #[tokio::test]
944    async fn releaser_removes_channel_on_terminal_status() {
945        for to in [
946            RunStatus::Completed,
947            RunStatus::Failed,
948            RunStatus::Cancelled,
949        ] {
950            let bus = WorkflowEventBus::new();
951            let releaser = EventBusReleaser::new(bus.clone());
952            let run_id = Uuid::now_v7();
953            let _rx = bus.subscribe(run_id);
954
955            releaser.handle(&status_changed(run_id, to)).await;
956
957            assert!(!bus.has_channel(run_id), "{to} must release the channel");
958        }
959    }
960
961    #[tokio::test]
962    async fn releaser_keeps_channel_on_non_terminal_status() {
963        for to in [RunStatus::Running, RunStatus::Pending, RunStatus::Paused] {
964            let bus = WorkflowEventBus::new();
965            let releaser = EventBusReleaser::new(bus.clone());
966            let run_id = Uuid::now_v7();
967            let _rx = bus.subscribe(run_id);
968
969            releaser.handle(&status_changed(run_id, to)).await;
970
971            assert!(bus.has_channel(run_id), "{to} must keep the channel");
972        }
973    }
974
975    #[tokio::test]
976    async fn releaser_ignores_other_runs() {
977        let bus = WorkflowEventBus::new();
978        let releaser = EventBusReleaser::new(bus.clone());
979        let run_id = Uuid::now_v7();
980        let _rx = bus.subscribe(run_id);
981
982        releaser
983            .handle(&status_changed(Uuid::now_v7(), RunStatus::Completed))
984            .await;
985
986        assert!(bus.has_channel(run_id));
987        assert_eq!(releaser.name(), "event-bus-releaser");
988    }
989}
ironflow_engine/notify/event_bus.rs

ironflow_engine/notify/
event_bus.rs
