ironflow_api/routes/
run_events.rs

1//! SSE endpoint for per-run workflow event streaming.
2//!
3//! A stream past the caps of [`SseLimits`](crate::sse_limiter::SseLimits) is
4//! refused with `429 SSE_CONNECTION_LIMIT`. The stream of a run ends once the
5//! run reaches a terminal status; a run already terminal gets a stream that
6//! ends at once.
7
8use std::convert::Infallible;
9use std::pin::Pin;
10use std::time::Duration;
11
12use axum::extract::{Path, Query, State};
13use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
14use futures_util::stream::{Stream, StreamExt, empty};
15use serde::Deserialize;
16use serde::de::{self, Deserializer};
17use tokio_stream::wrappers::BroadcastStream;
18use uuid::Uuid;
19
20use crate::authorization::AuthorizeExt;
21use crate::error::ApiError;
22use crate::state::AppState;
23use ironflow_auth::extractor::Authenticated;
24use ironflow_engine::notify::WorkflowEvent;
25use ironflow_store::entities::ApiKeyScope;
26
27/// Deserialize a comma-separated string into `Option<Vec<String>>`.
28fn deserialize_comma_strings<'de, D>(deserializer: D) -> Result<Option<Vec<String>>, D::Error>
29where
30    D: Deserializer<'de>,
31{
32    let opt: Option<String> = Option::deserialize(deserializer)?;
33    match opt {
34        None => Ok(None),
35        Some(raw) => {
36            let all_types = [
37                WorkflowEvent::STEP_STARTED,
38                WorkflowEvent::STEP_COMPLETED,
39                WorkflowEvent::STEP_FAILED,
40                WorkflowEvent::APPROVAL_REQUIRED,
41                WorkflowEvent::INPUT_REQUIRED,
42                WorkflowEvent::AGENT_STEP_TOKENS_USED,
43                WorkflowEvent::AGENT_STEP_RESUMED,
44            ];
45
46            let kinds: Vec<String> = raw
47                .split(',')
48                .map(|s| s.trim())
49                .filter(|s| !s.is_empty())
50                .map(|s| {
51                    if all_types.contains(&s) {
52                        Ok(s.to_string())
53                    } else {
54                        Err(de::Error::custom(format!(
55                            "unknown workflow event type: {s}"
56                        )))
57                    }
58                })
59                .collect::<Result<Vec<_>, _>>()?;
60
61            Ok(Some(kinds))
62        }
63    }
64}
65
66/// Query parameters for the per-run SSE events endpoint.
67///
68/// # Examples
69///
70/// ```
71/// use ironflow_api::routes::run_events::RunEventsQuery;
72///
73/// let query = RunEventsQuery { types: None };
74/// ```
75#[derive(Debug, Deserialize)]
76pub struct RunEventsQuery {
77    /// Comma-separated list of workflow event types to include
78    /// (e.g. `?types=step_started,step_completed`).
79    #[serde(default, deserialize_with = "deserialize_comma_strings")]
80    pub types: Option<Vec<String>>,
81}
82
83/// `GET /api/v1/runs/{id}/events` -- per-run Server-Sent Events stream.
84///
85/// Streams [`WorkflowEvent`]s for a specific workflow run in real time.
86/// Supports optional filtering via `?types=step_started,step_completed`.
87///
88/// Each SSE message has:
89/// - `event:` set to the event type (e.g. `step_started`)
90/// - `data:` JSON-serialized event payload
91///
92/// A keep-alive comment is sent every 30 seconds.
93///
94/// # Errors
95///
96/// Returns 401 if the request is not authenticated.
97/// Returns 404 if the run does not exist.
98#[cfg_attr(
99    feature = "openapi",
100    utoipa::path(
101        get,
102        path = "/api/v1/runs/{id}/events",
103        tags = ["runs"],
104        params(
105            ("id" = Uuid, Path, description = "Run ID"),
106            ("types" = Option<String>, Query, description = "Comma-separated workflow event types to filter (e.g. step_started,step_completed)")
107        ),
108        responses(
109            (status = 200, description = "SSE stream of workflow events"),
110            (status = 401, description = "Unauthorized"),
111            (status = 404, description = "Run not found")
112        ),
113        security(("Bearer" = []))
114    )
115)]
116pub async fn run_events(
117    auth: Authenticated,
118    State(state): State<AppState>,
119    Path(id): Path<Uuid>,
120    Query(query): Query<RunEventsQuery>,
121) -> Result<Sse<impl Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
122    auth.require(ApiKeyScope::RunsRead, false)?;
123    let run = state.get_run_or_404(id).await?;
124    let connection = state.sse_limiter.try_acquire(auth.user_id)?;
125
126    let type_filter = query.types;
127
128    let stream: Pin<Box<dyn Stream<Item = Result<SseEvent, Infallible>> + Send>> = match state
129        .event_bus
130    {
131        // A terminal run publishes nothing more, and its bus channel has been
132        // released: subscribing would re-create a channel nobody removes.
133        Some(ref bus) if !run.status.state.is_terminal() => {
134            let receiver = bus.subscribe(id);
135
136            Box::pin(BroadcastStream::new(receiver).filter_map(
137                move |result: Result<WorkflowEvent, _>| {
138                    let type_filter = type_filter.clone();
139                    async move {
140                        let event = result.ok()?;
141
142                        if let Some(ref kinds) = type_filter {
143                            let event_type = event.event_type();
144                            if !kinds.iter().any(|k| k == event_type) {
145                                return None;
146                            }
147                        }
148
149                        let data = serde_json::to_string(&event).ok()?;
150                        let sse_event = SseEvent::default().event(event.event_type()).data(data);
151
152                        Some(Ok::<_, Infallible>(sse_event))
153                    }
154                },
155            ))
156        }
157        _ => Box::pin(empty()),
158    };
159
160    // The stream owns the slot: axum drops it when the client disconnects.
161    let stream = stream.map(move |item| {
162        let _held = &connection;
163        item
164    });
165
166    Ok(Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(30))))
167}
168
169#[cfg(test)]
170mod tests {
171    use std::collections::HashMap;
172    use std::sync::Arc;
173    use std::time::Duration;
174
175    use axum::Router;
176    use axum::routing::get;
177    use chrono::Utc;
178    use ironflow_core::providers::claude::ClaudeCodeProvider;
179    use ironflow_engine::engine::Engine;
180    use ironflow_engine::notify::{
181        Event, WorkflowAgentStepResumedEvent, WorkflowEvent, WorkflowEventBus,
182        WorkflowStepCompletedEvent, WorkflowStepStartedEvent,
183    };
184    use ironflow_store::memory::InMemoryStore;
185    use ironflow_store::models::{NewRun, RunStatus, TriggerKind};
186    use serde_json::json;
187    use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
188    use tokio::net::{TcpListener, TcpStream};
189    use tokio::sync::broadcast;
190    use tokio::time::{sleep, timeout};
191    use uuid::Uuid;
192
193    use super::run_events;
194    use crate::routes::test_helpers::create_user_auth_header;
195    use crate::sse_limiter::SseLimits;
196    use crate::state::AppState;
197
198    fn test_state_with_bus() -> (AppState, WorkflowEventBus) {
199        let store = Arc::new(InMemoryStore::new());
200        let provider = Arc::new(ClaudeCodeProvider::new());
201        let engine = Arc::new(Engine::new(store.clone(), provider));
202        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
203            secret: "test-secret".to_string(),
204            access_token_ttl_secs: 900,
205            refresh_token_ttl_secs: 604800,
206            cookie_domain: None,
207            cookie_secure: false,
208        });
209        let (event_sender, _) = broadcast::channel::<Event>(16);
210        let bus = WorkflowEventBus::new();
211        let state = AppState::new(
212            store,
213            engine,
214            jwt_config,
215            "test-worker-token".to_string(),
216            event_sender,
217        )
218        .with_event_bus(bus.clone());
219        (state, bus)
220    }
221
222    async fn create_run(state: &AppState) -> Uuid {
223        state
224            .store
225            .create_run(NewRun {
226                created_by: None,
227                workflow_name: "test".to_string(),
228                trigger: TriggerKind::Manual,
229                payload: json!({}),
230                max_retries: 0,
231                handler_version: None,
232                labels: HashMap::new(),
233                scheduled_at: None,
234                idempotency_key: None,
235                concurrency_key: None,
236                priority: 0,
237                concurrency_limits: Vec::new(),
238                max_cost_usd: None,
239                worker_tags: Vec::new(),
240            })
241            .await
242            .unwrap()
243            .into_run()
244            .id
245    }
246
247    async fn start_sse_server(state: AppState) -> (String, String) {
248        let auth = create_user_auth_header(&state, "testuser", false).await;
249        let app = Router::new()
250            .route("/{id}/events", get(run_events))
251            .with_state(state);
252
253        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
254        let addr = listener.local_addr().unwrap().to_string();
255        tokio::spawn(async move {
256            axum::serve(listener, app).await.unwrap();
257        });
258        (addr, auth)
259    }
260
261    async fn connect_sse(addr: &str, path: &str, auth: &str) -> BufReader<TcpStream> {
262        let stream = TcpStream::connect(addr).await.unwrap();
263        let (reader, mut writer) = stream.into_split();
264
265        writer
266            .write_all(
267                format!(
268                    "GET {path} HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\nAuthorization: {auth}\r\n\r\n"
269                )
270                .as_bytes(),
271            )
272            .await
273            .unwrap();
274
275        BufReader::new(reader.reunite(writer).unwrap())
276    }
277
278    /// Reads and discards the HTTP status line and headers, blocking until
279    /// the blank line that terminates them.
280    ///
281    /// By the time a client can read any response bytes, `run_events` has
282    /// already called `bus.subscribe`, so this is a deterministic
283    /// synchronization point for tests that publish events right after
284    /// connecting -- unlike a fixed `sleep`, it cannot race under load.
285    async fn wait_for_response_headers(reader: &mut BufReader<TcpStream>) {
286        loop {
287            let mut line = String::new();
288            let n = reader.read_line(&mut line).await.unwrap();
289            if n == 0 || line == "\r\n" || line == "\n" {
290                break;
291            }
292        }
293    }
294
295    async fn read_until_contains(
296        reader: &mut BufReader<TcpStream>,
297        needle: &str,
298        dur: Duration,
299    ) -> String {
300        let mut accumulated = String::new();
301        let result = timeout(dur, async {
302            loop {
303                let mut line = String::new();
304                let n = reader.read_line(&mut line).await.unwrap();
305                if n == 0 {
306                    break;
307                }
308                accumulated.push_str(&line);
309                if accumulated.contains(needle) {
310                    break;
311                }
312            }
313        })
314        .await;
315        if result.is_err() {
316            panic!("timeout waiting for '{needle}' in SSE stream. Data so far:\n{accumulated}");
317        }
318        accumulated
319    }
320
321    #[tokio::test]
322    async fn sse_stream_receives_workflow_events() {
323        let (state, bus) = test_state_with_bus();
324        let run_id = create_run(&state).await;
325        let (addr, auth) = start_sse_server(state).await;
326
327        let mut reader = connect_sse(&addr, &format!("/{run_id}/events"), &auth).await;
328        wait_for_response_headers(&mut reader).await;
329
330        bus.publish(
331            run_id,
332            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
333                step_name: "build".to_string(),
334                step_index: 0,
335                timestamp: Utc::now(),
336            }),
337        );
338
339        let text = read_until_contains(&mut reader, "build", Duration::from_secs(5)).await;
340
341        assert!(text.contains("event: step_started"));
342        assert!(text.contains("build"));
343    }
344
345    #[tokio::test]
346    async fn returns_404_for_unknown_run() {
347        let (state, _bus) = test_state_with_bus();
348        let (addr, auth) = start_sse_server(state).await;
349
350        let unknown = Uuid::nil();
351        let mut reader = connect_sse(&addr, &format!("/{unknown}/events"), &auth).await;
352
353        let text = read_until_contains(&mut reader, "404", Duration::from_secs(5)).await;
354        assert!(text.contains("404"));
355    }
356
357    #[tokio::test]
358    async fn rejects_unauthenticated() {
359        let (state, _bus) = test_state_with_bus();
360        let run_id = create_run(&state).await;
361        let (addr, _auth) = start_sse_server(state).await;
362
363        let stream = TcpStream::connect(&addr).await.unwrap();
364        let (reader, mut writer) = stream.into_split();
365
366        writer
367            .write_all(
368                format!(
369                    "GET /{run_id}/events HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\n\r\n"
370                )
371                .as_bytes(),
372            )
373            .await
374            .unwrap();
375
376        let mut buf_reader = BufReader::new(reader.reunite(writer).unwrap());
377        let text = read_until_contains(&mut buf_reader, "401", Duration::from_secs(5)).await;
378        assert!(text.contains("401"));
379    }
380
381    #[tokio::test]
382    async fn filters_by_event_type() {
383        let (state, bus) = test_state_with_bus();
384        let run_id = create_run(&state).await;
385        let (addr, auth) = start_sse_server(state).await;
386
387        let mut reader = connect_sse(
388            &addr,
389            &format!("/{run_id}/events?types=step_completed"),
390            &auth,
391        )
392        .await;
393        wait_for_response_headers(&mut reader).await;
394
395        bus.publish(
396            run_id,
397            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
398                step_name: "build".to_string(),
399                step_index: 0,
400                timestamp: Utc::now(),
401            }),
402        );
403        bus.publish(
404            run_id,
405            WorkflowEvent::StepCompleted(WorkflowStepCompletedEvent {
406                step_name: "build".to_string(),
407                step_index: 0,
408                duration_ms: 1234,
409                output_summary: None,
410            }),
411        );
412
413        let text = read_until_contains(&mut reader, "step_completed", Duration::from_secs(5)).await;
414
415        assert!(text.contains("step_completed"));
416        assert!(!text.contains("event: step_started"));
417    }
418
419    #[tokio::test]
420    async fn filters_by_agent_step_resumed_event_type() {
421        let (state, bus) = test_state_with_bus();
422        let run_id = create_run(&state).await;
423        let (addr, auth) = start_sse_server(state).await;
424
425        let mut reader = connect_sse(
426            &addr,
427            &format!("/{run_id}/events?types=agent_step_resumed"),
428            &auth,
429        )
430        .await;
431        wait_for_response_headers(&mut reader).await;
432
433        bus.publish(
434            run_id,
435            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
436                step_name: "review".to_string(),
437                step_index: 0,
438                timestamp: Utc::now(),
439            }),
440        );
441        bus.publish(
442            run_id,
443            WorkflowEvent::AgentStepResumed(WorkflowAgentStepResumedEvent {
444                step_name: "review".to_string(),
445                step_index: 0,
446                session_id: "0192f0c1-7d2e-7a4b-9c3d-1e2f3a4b5c6d".to_string(),
447                timestamp: Utc::now(),
448            }),
449        );
450
451        let text = read_until_contains(
452            &mut reader,
453            "0192f0c1-7d2e-7a4b-9c3d-1e2f3a4b5c6d",
454            Duration::from_secs(5),
455        )
456        .await;
457
458        assert!(text.contains("event: agent_step_resumed"));
459        assert!(!text.contains("event: step_started"));
460    }
461
462    #[tokio::test]
463    async fn events_isolated_between_runs() {
464        let (state, bus) = test_state_with_bus();
465        let run_a = create_run(&state).await;
466        let run_b = create_run(&state).await;
467        let (addr, auth) = start_sse_server(state).await;
468
469        let mut reader_a = connect_sse(&addr, &format!("/{run_a}/events"), &auth).await;
470        wait_for_response_headers(&mut reader_a).await;
471
472        bus.publish(
473            run_b,
474            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
475                step_name: "only-for-b".to_string(),
476                step_index: 0,
477                timestamp: Utc::now(),
478            }),
479        );
480        bus.publish(
481            run_a,
482            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
483                step_name: "only-for-a".to_string(),
484                step_index: 0,
485                timestamp: Utc::now(),
486            }),
487        );
488
489        let text = read_until_contains(&mut reader_a, "only-for-a", Duration::from_secs(5)).await;
490
491        assert!(text.contains("only-for-a"));
492        assert!(!text.contains("only-for-b"));
493    }
494
495    /// Read the HTTP status line of a response.
496    async fn status_line(reader: &mut BufReader<TcpStream>) -> String {
497        read_until_contains(reader, "\r\n", Duration::from_secs(5)).await
498    }
499
500    /// Read the headers left after the status line, then the body they announce.
501    async fn read_body(reader: &mut BufReader<TcpStream>) -> String {
502        timeout(Duration::from_secs(5), async {
503            let mut content_length = 0;
504            loop {
505                let mut line = String::new();
506                reader.read_line(&mut line).await.unwrap();
507                if line.is_empty() || line == "\r\n" {
508                    break;
509                }
510                if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") {
511                    content_length = value.trim().parse().unwrap();
512                }
513            }
514            let mut body = vec![0; content_length];
515            reader.read_exact(&mut body).await.unwrap();
516            String::from_utf8(body).unwrap()
517        })
518        .await
519        .expect("timeout reading the response body")
520    }
521
522    fn limited_state(per_principal: usize) -> (AppState, WorkflowEventBus) {
523        let (state, bus) = test_state_with_bus();
524        let state = state.with_sse_limits(SseLimits {
525            per_principal: Some(per_principal),
526            global: None,
527        });
528        (state, bus)
529    }
530
531    #[tokio::test]
532    async fn run_events_returns_429_past_the_per_principal_limit() {
533        let (state, _bus) = limited_state(2);
534        let run_id = create_run(&state).await;
535        let other_auth = create_user_auth_header(&state, "otheruser", false).await;
536        let (addr, auth) = start_sse_server(state).await;
537        let path = format!("/{run_id}/events");
538
539        let mut first = connect_sse(&addr, &path, &auth).await;
540        assert!(status_line(&mut first).await.starts_with("HTTP/1.1 200"));
541        let mut second = connect_sse(&addr, &path, &auth).await;
542        assert!(status_line(&mut second).await.starts_with("HTTP/1.1 200"));
543
544        let mut third = connect_sse(&addr, &path, &auth).await;
545        let status = status_line(&mut third).await;
546        assert!(status.starts_with("HTTP/1.1 429"), "status line: {status}");
547        let body = read_body(&mut third).await;
548        assert!(body.contains("SSE_CONNECTION_LIMIT"), "body: {body}");
549
550        let mut other = connect_sse(&addr, &path, &other_auth).await;
551        let status = status_line(&mut other).await;
552        assert!(
553            status.starts_with("HTTP/1.1 200"),
554            "another user keeps its own quota: {status}"
555        );
556    }
557
558    #[tokio::test]
559    async fn run_events_frees_the_slot_when_the_client_disconnects() {
560        let (state, _bus) = limited_state(1);
561        let run_id = create_run(&state).await;
562        let (addr, auth) = start_sse_server(state).await;
563        let path = format!("/{run_id}/events");
564
565        let mut first = connect_sse(&addr, &path, &auth).await;
566        assert!(status_line(&mut first).await.starts_with("HTTP/1.1 200"));
567        let mut refused = connect_sse(&addr, &path, &auth).await;
568        assert!(status_line(&mut refused).await.starts_with("HTTP/1.1 429"));
569
570        drop(first);
571
572        let reopened = timeout(Duration::from_secs(5), async {
573            loop {
574                let mut retry = connect_sse(&addr, &path, &auth).await;
575                if status_line(&mut retry).await.starts_with("HTTP/1.1 200") {
576                    return retry;
577                }
578                sleep(Duration::from_millis(20)).await;
579            }
580        })
581        .await;
582        assert!(reopened.is_ok(), "the slot was not freed after disconnect");
583    }
584
585    #[tokio::test]
586    async fn run_events_unknown_run_is_404_before_the_limit() {
587        let (state, _bus) = limited_state(0);
588        let (addr, auth) = start_sse_server(state).await;
589
590        let unknown = Uuid::nil();
591        let mut reader = connect_sse(&addr, &format!("/{unknown}/events"), &auth).await;
592
593        let status = status_line(&mut reader).await;
594        assert!(status.starts_with("HTTP/1.1 404"), "status line: {status}");
595    }
596
597    #[tokio::test]
598    async fn run_events_on_terminal_run_ends_without_creating_a_channel() {
599        let (state, bus) = test_state_with_bus();
600        let run_id = create_run(&state).await;
601        state
602            .store
603            .update_run_status(run_id, RunStatus::Running)
604            .await
605            .unwrap();
606        state
607            .store
608            .update_run_status(run_id, RunStatus::Completed)
609            .await
610            .unwrap();
611        let (addr, auth) = start_sse_server(state).await;
612
613        let mut stream = TcpStream::connect(&addr).await.unwrap();
614        stream
615            .write_all(
616                format!(
617                    "GET /{run_id}/events HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\nAuthorization: {auth}\r\nConnection: close\r\n\r\n"
618                )
619                .as_bytes(),
620            )
621            .await
622            .unwrap();
623
624        let mut response = String::new();
625        timeout(Duration::from_secs(5), stream.read_to_string(&mut response))
626            .await
627            .expect("the stream of a terminal run must end")
628            .unwrap();
629
630        assert!(response.starts_with("HTTP/1.1 200"), "response: {response}");
631        assert!(!bus.has_channel(run_id));
632    }
633
634    #[tokio::test]
635    async fn run_events_on_live_run_subscribes_to_the_bus() {
636        let (state, bus) = test_state_with_bus();
637        let run_id = create_run(&state).await;
638        let (addr, auth) = start_sse_server(state).await;
639
640        let mut reader = connect_sse(&addr, &format!("/{run_id}/events"), &auth).await;
641        wait_for_response_headers(&mut reader).await;
642
643        assert!(bus.has_channel(run_id));
644    }
645}
ironflow_api/routes/run_events.rs

ironflow_api/routes/
run_events.rs
