prospero_api/
sse.rs

1//! Server-Sent Events: replay an agent's history from the store, then tail the
2//! live event bus — joined on the monotonic `seq` with no gap or dup.
3//!
4//! The stream closes right after the agent's terminal `AgentFinished` event, so
5//! `prospero follow` behaves like `tail` of a finite run and the dashboard
6//! shows the finished run then closes cleanly (rather than hanging forever).
7
8mod tail;
9
10use std::convert::Infallible;
11
12use async_stream::stream;
13use axum::extract::{Path, Query, State};
14use axum::response::sse::{Event, KeepAlive, Sse};
15use prospero_core::FleetEvent;
16use prospero_core::event::EventKind;
17use tokio_stream::{Stream, StreamExt};
18
19use crate::AppState;
20use crate::dto::FromSeq;
21use tail::{Frame, GapSignal, Step, Tailer};
22
23/// `GET /api/agents/{id}/stream` — replay-then-tail SSE of `FleetEvent`s.
24pub async fn agent_stream(
25    State(st): State<AppState>,
26    Path(id): Path<String>,
27    Query(q): Query<FromSeq>,
28) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
29    // Subscribe BEFORE reading history so no live event is missed in the gap:
30    // InProcessBus registers its receiver here (eagerly); DistributedBus replays
31    // from seq 0 on its first doorbell. Either way the live tail covers every
32    // event after this point, and the `seq` dedup below drops the history overlap.
33    let mut sub = st.bus.subscribe(&id);
34    let history = st.store.replay(&id, q.from).await.unwrap_or_default();
35
36    let body = stream! {
37        // 1) Replay persisted history, stopping if it already contains the
38        //    terminal event. Track the last seq delivered as the dedup
39        //    high-water mark for the live tail. Seed it from the client's
40        //    `from` floor so a later self-heal replay never re-sends events
41        //    below what the client asked for (seq is monotonic per stream, so an
42        //    agent can legitimately have no events at or above `from` yet).
43        let mut last_delivered = q.from.saturating_sub(1);
44        for ev in history {
45            let terminal = is_terminal(&ev);
46            last_delivered = ev.seq;
47            yield Ok(to_event(&ev));
48            if terminal {
49                return;
50            }
51        }
52
53        // 2) Tail live events, self-healing across a slow-consumer `Lagged`.
54        //    The per-subscriber broadcast buffer is the lag tolerance
55        //    (`FleetConfig::event_buffer`, default 1024). Exceed it and the
56        //    `Tailer` emits a `gap` signal plus replays the missed events from
57        //    the durable store, rather than silently skipping them.
58        let mut tailer = Tailer::new(id, last_delivered, st.store.clone());
59        loop {
60            match tailer.on_recv(sub.next().await).await {
61                Step::Emit(frames) => {
62                    for f in frames {
63                        yield Ok(frame_to_event(&f));
64                    }
65                }
66                Step::EmitAndClose(frames) => {
67                    for f in frames {
68                        yield Ok(frame_to_event(&f));
69                    }
70                    break;
71                }
72                Step::Skip => continue,
73                Step::Close => break,
74            }
75        }
76    };
77
78    Sse::new(body).keep_alive(KeepAlive::default())
79}
80
81fn is_terminal(ev: &FleetEvent) -> bool {
82    matches!(ev.kind, EventKind::AgentFinished { .. })
83}
84
85fn to_event(ev: &FleetEvent) -> Event {
86    // json_data only fails if serialization fails, which FleetEvent never does.
87    Event::default()
88        .json_data(ev)
89        .unwrap_or_else(|_| Event::default().data("{}"))
90}
91
92fn frame_to_event(frame: &Frame) -> Event {
93    match frame {
94        Frame::Event(ev) => to_event(ev),
95        Frame::Gap { skipped, last_seq } => Event::default()
96            .event("gap")
97            .json_data(GapSignal {
98                skipped: *skipped,
99                last_seq: *last_seq,
100            })
101            .unwrap_or_else(|_| Event::default().event("gap").data("{}")),
102    }
103}
prospero_api/sse.rs

prospero_api/
sse.rs
