Vakyartha documentation

1//! Session event streams (docs/design/48-web-client.md §4.4, §4.7).
2//!
3//! Each subscription a client can hold — a session's agent events, its
4//! presentation frames, its side-chat branch, its coworking signal, the host
5//! and config changes — is built here exactly once. The single-session
6//! routes (`/sessions/{id}/events`, `/sessions/{id}/presentation/events`)
7//! serve one of them each for clients that follow one session, such as
8//! `vak term`. `GET /stream` merges any set of them onto ONE connection.
9//!
10//! The merge exists because a browser allows six HTTP/1.1 connections per
11//! host across all of its tabs, and HTTP/2 is not available on plain-http
12//! loopback. A tab that held one EventSource per subscription exhausted the
13//! pool with two or three tabs open, and every ordinary `fetch` then queued
14//! behind streams that never end.
15
16use std::collections::BTreeMap;
17use std::pin::Pin;
18use std::sync::Arc;
19
20use axum::extract::State;
21use axum::http::StatusCode;
22use axum::response::sse::{Event, KeepAlive, Sse};
23use axum::response::{IntoResponse, Response};
24use futures::stream::{self, Stream, StreamExt};
25use tokio_stream::wrappers::{BroadcastStream, IntervalStream};
26use vak_agent::AgentEvent;
27
28use crate::{AppState, SessionHandle};
29
30type Frames<T> = Pin<Box<dyn Stream<Item = T> + Send>>;
31
32/// Most sessions one `/stream` connection may follow. A client follows the
33/// sessions on screen plus any best-of-N candidates; anything past this is a
34/// client bug, refused rather than turned into unbounded server work.
35pub(crate) const MAX_STREAM_SESSIONS: usize = 32;
36
37/// One item of a session's agent-event subscription. `Event` already carries
38/// `crate::client_events::ClientEvent`, the same projection the single-
39/// session `/sessions/{id}/events` route applies through `seq_frame` — a
40/// multiplexed connection must not become a second, unprojected path for
41/// internal `AgentEvent` traffic (retry reasons, route legs, raw error text)
42/// to reach a client (docs/audits Finding 1).
43pub(crate) enum AgentFrame {
44 /// The resume point is older than the replay ring, or this consumer
45 /// lagged: what the client holds may be missing events it cannot see.
46 Resync(&'static str),
47 Event {
48 seq: u64,
49 event: crate::client_events::ClientEvent,
50 },
51}
52
53/// A session's agent events, resuming after `resume` when given.
54///
55/// Subscribes BEFORE reading the replay ring, so an event published between
56/// the two is received live rather than falling into the gap between them.
57/// Duplicates are filtered by sequence number; a gap could not be recovered.
58/// `seq` stays monotonic across a projected-away event: it produces no
59/// frame, identically on replay and live, exactly like `seq_frame`.
60pub(crate) fn agent_frames(handle: &SessionHandle, resume: Option<u64>) -> Frames<AgentFrame> {
61 let rx = handle.events_tx.subscribe();
62
63 // `None` from the ring means it no longer reaches back that far, and the
64 // client is told to rebuild from the durable transcript rather than being
65 // handed a stream with a hole in it that it cannot see.
66 let (replay, resync) = match resume {
67 Some(seq) => match handle.events_tx.replay_after(seq) {
68 Some(missed) => (missed, false),
69 None => (Vec::new(), true),
70 },
71 None => (Vec::new(), false),
72 };
73 let highest_replayed = replay.last().map(|e| e.seq).unwrap_or(0);
74
75 handle.subscribed.notify_one();
76 handle.events_tx.send(AgentEvent::StreamOpened);
77
78 let resync = resync.then_some(AgentFrame::Resync("events older than the replay window"));
79 let live = BroadcastStream::new(rx).filter_map(move |event| {
80 std::future::ready(match event {
81 // At or below what the replay already delivered is a duplicate.
82 Ok(framed) if framed.seq <= highest_replayed => None,
83 Ok(framed) => {
84 crate::client_events::project(framed.event).map(|event| AgentFrame::Event {
85 seq: framed.seq,
86 event,
87 })
88 }
89 Err(_) => Some(AgentFrame::Resync("live event consumer lagged")),
90 })
91 });
92 Box::pin(
93 stream::iter(resync)
94 .chain(stream::iter(replay.into_iter().filter_map(|framed| {
95 crate::client_events::project(framed.event).map(|event| AgentFrame::Event {
96 seq: framed.seq,
97 event,
98 })
99 })))
100 .chain(live),
101 )
102}
103
104/// One serialized `OutputStreamFrame`, with the sequence it brings the
105/// client up to when it has one.
106pub(crate) struct PresentationFrame {
107 pub(crate) sequence: Option<u64>,
108 pub(crate) json: String,
109}
110
111fn presentation_frame(frame: &vak_delivery::OutputStreamFrame) -> PresentationFrame {
112 PresentationFrame {
113 sequence: frame.sequence,
114 json: serde_json::to_string(frame).unwrap_or_else(|error| {
115 serde_json::json!({
116 "error": "presentation serialization failed",
117 "detail": error.to_string(),
118 })
119 .to_string()
120 }),
121 }
122}
123
124/// A session's presentation frames: an authoritative snapshot first, then
125/// one frame per projected live event. A session with no live handle gets
126/// its historical snapshot and nothing after it; `None` is an unknown id.
127///
128/// Snapshot-based on purpose: a reconnect receives a fresh snapshot, whose
129/// sequence becomes the new cursor. Delta replay is never manufactured from
130/// an unknown historical baseline.
131pub(crate) fn presentation_frames(
132 state: &AppState,
133 id: &str,
134 handle: Option<Arc<SessionHandle>>,
135) -> Option<Frames<PresentationFrame>> {
136 let Some(handle) = handle else {
137 let session = crate::open_historical_session(state, id)?;
138 let frame = vak_delivery::OutputStreamFrame {
139 sequence: None,
140 delta: None,
141 snapshot: Some(crate::projection::snapshot(id, &session)),
142 };
143 return Some(Box::pin(stream::once(std::future::ready(
144 presentation_frame(&frame),
145 ))));
146 };
147 let id = id.to_owned();
148 let rx = handle.events_tx.subscribe();
149 let mut timeline = {
150 let guard = handle
151 .session
152 .lock()
153 .unwrap_or_else(std::sync::PoisonError::into_inner);
154 guard
155 .as_ref()
156 .map(|session| crate::live_presentation_snapshot(&handle.core, &id, session))
157 .unwrap_or_else(|| {
158 handle
159 .presentation
160 .lock()
161 .unwrap_or_else(std::sync::PoisonError::into_inner)
162 .clone()
163 })
164 };
165 let mut last_sequence = live_cursor(&timeline).unwrap_or(0);
166 let initial = presentation_frame(&vak_delivery::OutputStreamFrame {
167 sequence: Some(last_sequence),
168 delta: None,
169 snapshot: Some(timeline.clone()),
170 });
171 handle.subscribed.notify_one();
172 let live = BroadcastStream::new(rx).filter_map(move |event| {
173 let frame = match event {
174 Ok(framed) => {
175 if framed.seq <= last_sequence {
176 None
177 } else {
178 last_sequence = framed.seq;
179 if matches!(&framed.event, AgentEvent::RunFinished { .. }) {
180 // The run owner publishes the rebuilt durable
181 // projection before broadcasting RunFinished. Rebase
182 // this long-lived subscriber now: carrying its private
183 // live timeline into the next run would otherwise
184 // replace settled cards with the prior run's
185 // prose/progress snapshot.
186 timeline = handle
187 .presentation
188 .lock()
189 .unwrap_or_else(std::sync::PoisonError::into_inner)
190 .clone();
191 Some(crate::projection::settled_frame(last_sequence, &timeline))
192 } else {
193 crate::projection::project_frame(&mut timeline, framed)
194 }
195 }
196 }
197 Err(_) => {
198 if let Some(events) = handle.events_tx.replay_after(last_sequence) {
199 for framed in events {
200 last_sequence = framed.seq;
201 crate::projection::project_frame(&mut timeline, framed);
202 }
203 } else {
204 let guard = handle
205 .session
206 .lock()
207 .unwrap_or_else(std::sync::PoisonError::into_inner);
208 timeline = guard
209 .as_ref()
210 .map(|session| crate::projection::snapshot(&id, session))
211 .unwrap_or_else(|| {
212 handle
213 .presentation
214 .lock()
215 .unwrap_or_else(std::sync::PoisonError::into_inner)
216 .clone()
217 });
218 last_sequence = live_cursor(&timeline).unwrap_or(last_sequence);
219 timeline
220 .diagnostics
221 .push("Presentation stream resynchronized after a gap.".into());
222 }
223 Some(vak_delivery::OutputStreamFrame {
224 sequence: Some(last_sequence),
225 delta: None,
226 snapshot: Some(timeline.clone()),
227 })
228 }
229 };
230 std::future::ready(frame.map(|frame| presentation_frame(&frame)))
231 });
232 Some(Box::pin(
233 stream::once(std::future::ready(initial)).chain(live),
234 ))
235}
236
237fn live_cursor(timeline: &vak_delivery::OutputTimeline) -> Option<u64> {
238 timeline
239 .cursor
240 .as_deref()
241 .and_then(|cursor| cursor.strip_prefix("live:"))
242 .and_then(|seq| seq.parse().ok())
243}
244
245/// One item of a session's side-chat (`/btw`) subscription, projected
246/// through the same `client_events::project` as the main agent stream
247/// (docs/audits Finding 1) — a side chat is a full agent run and deserves
248/// the same protection from internal traffic.
249enum SideFrame {
250 /// The broadcast consumer lagged; distinct from a projected-away event,
251 /// which simply produces no `Event` variant.
252 Lagged,
253 Event(crate::client_events::ClientEvent),
254}
255
256fn side_frames(handle: &SessionHandle) -> Frames<SideFrame> {
257 let mut rx = handle.side_events_tx.subscribe();
258 let _ = rx.try_recv();
259 handle.side_events_tx.send(AgentEvent::StreamOpened);
260 Box::pin(BroadcastStream::new(rx).filter_map(|result| {
261 std::future::ready(match result {
262 Ok(framed) => crate::client_events::project(framed.event).map(SideFrame::Event),
263 Err(_) => Some(SideFrame::Lagged),
264 })
265 }))
266}
267
268/// Content-free wakeups for a session's shared candidate comments.
269fn coworking_frames(handle: &SessionHandle) -> Frames<()> {
270 Box::pin(BroadcastStream::new(handle.coworking_comments_tx.subscribe()).map(|_| ()))
271}
272
273/// Host-level facts (active workspace, recents, terminal), sent once on
274/// connect and then only when they change. Sampled rather than broadcast:
275/// a workspace switch is a human action, and a two-second latency on it is
276/// not worth an event family of its own.
277fn host_frames(state: AppState) -> Frames<String> {
278 let mut last: Option<String> = None;
279 let ticks = IntervalStream::new(tokio::time::interval(std::time::Duration::from_secs(2)));
280 Box::pin(ticks.filter_map(move |_| {
281 let payload = serde_json::to_string(&crate::web::host_payload(&state))
282 .unwrap_or_else(|_| "{}".into());
283 let changed = last.as_deref() != Some(payload.as_str());
284 if changed {
285 last = Some(payload.clone());
286 }
287 std::future::ready(changed.then_some(payload))
288 }))
289}
290
291/// `ConfigChanged` events from the global hub, so a setting written by
292/// another surface is reflected without a restart (docs/design/44
293/// "Liveness").
294fn config_frames(state: &AppState) -> Frames<String> {
295 Box::pin(
296 BroadcastStream::new(state.hub.subscribe()).filter_map(|event| {
297 std::future::ready(match event {
298 Ok(event @ crate::events::SystemEvent::ConfigChanged { .. }) => {
299 serde_json::to_string(&event).ok()
300 }
301 _ => None,
302 })
303 }),
304 )
305}
306
307/// What `/stream` should carry, parsed from its query string.
308#[derive(Debug, Default, PartialEq, Eq)]
309pub(crate) struct Interest {
310 pub(crate) sessions: Vec<String>,
311 pub(crate) host: bool,
312 pub(crate) config: bool,
313 pub(crate) cursor: BTreeMap<String, u64>,
314}
315
316impl Interest {
317 fn parse(query: Option<&str>) -> Self {
318 let mut interest = Self::default();
319 for part in query.unwrap_or_default().split('&') {
320 let (key, value) = part.split_once('=').unwrap_or((part, ""));
321 let value = percent_encoding::percent_decode_str(&value.replace('+', " "))
322 .decode_utf8_lossy()
323 .into_owned();
324 match key {
325 "session" if !value.is_empty() && !interest.sessions.contains(&value) => {
326 interest.sessions.push(value);
327 }
328 "host" => interest.host = value == "1",
329 "config" => interest.config = value == "1",
330 "cursor" => interest.cursor = parse_cursor(&value),
331 _ => {}
332 }
333 }
334 interest
335 }
336}
337
338/// A `/stream` resume cursor: `<session>:<seq>` pairs joined by `,`.
339///
340/// It is the SSE event id of every agent frame, so a browser's own reconnect
341/// sends the whole vector back as `Last-Event-ID` and every session resumes
342/// from its own sequence. A client that reopens the stream itself (because
343/// its set of sessions changed) passes the same string as `?cursor=`.
344pub(crate) fn parse_cursor(raw: &str) -> BTreeMap<String, u64> {
345 raw.split(',')
346 .filter_map(|pair| {
347 let (session, seq) = pair.trim().rsplit_once(':')?;
348 Some((session.to_owned(), seq.parse().ok()?))
349 })
350 .filter(|(session, _)| !session.is_empty())
351 .collect()
352}
353
354pub(crate) fn format_cursor(cursor: &BTreeMap<String, u64>) -> String {
355 cursor
356 .iter()
357 .map(|(session, seq)| format!("{session}:{seq}"))
358 .collect::<Vec<_>>()
359 .join(",")
360}
361
362enum Muxed {
363 Host(String),
364 Config(String),
365 Agent(String, AgentFrame),
366 Presentation(String, PresentationFrame),
367 Side(String, SideFrame),
368 Coworking(String),
369 Unknown(String),
370}
371
372fn session_data(session: &str, key: &str, raw_json: &str) -> String {
373 format!(
374 "{{\"session\":{},\"{key}\":{raw_json}}}",
375 serde_json::Value::from(session)
376 )
377}
378
379fn client_event_json(event: &crate::client_events::ClientEvent) -> String {
380 serde_json::to_string(event).unwrap_or_else(|error| {
381 serde_json::json!({
382 "error": "event serialization failed",
383 "detail": error.to_string(),
384 })
385 .to_string()
386 })
387}
388
389/// `GET /stream` — every subscription a client holds, on one connection.
390///
391/// Query: `session=<id>` (repeatable), `host=1`, `config=1`, and optionally
392/// `cursor=` (see [`parse_cursor`]); a `Last-Event-ID` header wins over the
393/// query cursor. Every frame is a named SSE event whose JSON names the
394/// session it belongs to: `agent` (with the cursor as its id), `presentation`,
395/// `side`, `coworking`, `resync`, `unknown`, `host`, `config`.
396///
397/// Operator-only by construction: the participant middleware admits only
398/// `/sessions/<its conversation>/…` paths.
399pub(crate) async fn stream(
400 State(state): State<AppState>,
401 headers: axum::http::HeaderMap,
402 uri: axum::http::Uri,
403) -> Response {
404 let mut interest = Interest::parse(uri.query());
405 if let Some(header) = headers
406 .get("last-event-id")
407 .and_then(|value| value.to_str().ok())
408 .filter(|value| !value.trim().is_empty())
409 {
410 interest.cursor = parse_cursor(header);
411 }
412 if interest.sessions.len() > MAX_STREAM_SESSIONS {
413 return (
414 StatusCode::BAD_REQUEST,
415 format!("a stream follows at most {MAX_STREAM_SESSIONS} sessions"),
416 )
417 .into_response();
418 }
419 if interest.sessions.is_empty() && !interest.host && !interest.config {
420 return (
421 StatusCode::BAD_REQUEST,
422 "a stream needs at least one subscription",
423 )
424 .into_response();
425 }
426
427 let mut parts: Vec<Frames<Muxed>> = Vec::new();
428 if interest.host {
429 parts.push(Box::pin(host_frames(state.clone()).map(Muxed::Host)));
430 }
431 if interest.config {
432 parts.push(Box::pin(config_frames(&state).map(Muxed::Config)));
433 }
434 // Only sessions this connection follows keep a cursor entry; a session
435 // dropped from the set must not be resumed from a stale position later.
436 let mut cursor = BTreeMap::new();
437 for id in &interest.sessions {
438 let handle = crate::ensure_session_handle(&state, id)
439 .await
440 .ok()
441 .map(|(_, handle)| handle);
442 let presentation = presentation_frames(&state, id, handle.clone());
443 let Some(handle) = handle else {
444 match presentation {
445 Some(frames) => {
446 let session = id.clone();
447 parts.push(Box::pin(
448 frames.map(move |frame| Muxed::Presentation(session.clone(), frame)),
449 ));
450 }
451 None => parts.push(Box::pin(stream::once(std::future::ready(Muxed::Unknown(
452 id.clone(),
453 ))))),
454 }
455 continue;
456 };
457 let resume = interest.cursor.get(id).copied();
458 if let Some(seq) = resume {
459 cursor.insert(id.clone(), seq);
460 }
461 let session = id.clone();
462 parts.push(Box::pin(
463 agent_frames(&handle, resume).map(move |frame| Muxed::Agent(session.clone(), frame)),
464 ));
465 if let Some(frames) = presentation {
466 let session = id.clone();
467 parts.push(Box::pin(
468 frames.map(move |frame| Muxed::Presentation(session.clone(), frame)),
469 ));
470 }
471 let session = id.clone();
472 parts.push(Box::pin(
473 side_frames(&handle).map(move |event| Muxed::Side(session.clone(), event)),
474 ));
475 let session = id.clone();
476 parts.push(Box::pin(
477 coworking_frames(&handle).map(move |()| Muxed::Coworking(session.clone())),
478 ));
479 }
480
481 let frames = stream::select_all(parts).map(move |item| {
482 let event = match item {
483 Muxed::Host(payload) => Event::default().event("host").data(payload),
484 Muxed::Config(payload) => Event::default().event("config").data(payload),
485 Muxed::Agent(session, AgentFrame::Event { seq, event }) => {
486 cursor.insert(session.clone(), seq);
487 Event::default()
488 .event("agent")
489 .id(format_cursor(&cursor))
490 .data(session_data(&session, "event", &client_event_json(&event)))
491 }
492 Muxed::Agent(session, AgentFrame::Resync(reason)) => {
493 Event::default().event("resync").data(session_data(
494 &session,
495 "reason",
496 &serde_json::Value::from(reason).to_string(),
497 ))
498 }
499 Muxed::Presentation(session, frame) => Event::default()
500 .event("presentation")
501 .data(session_data(&session, "frame", &frame.json)),
502 Muxed::Side(session, SideFrame::Event(event)) => Event::default()
503 .event("side")
504 .data(session_data(&session, "event", &client_event_json(&event))),
505 Muxed::Side(session, SideFrame::Lagged) => Event::default()
506 .event("side")
507 .data(session_data(&session, "lagged", "true")),
508 Muxed::Coworking(session) => Event::default()
509 .event("coworking")
510 .data(session_data(&session, "refresh", "true")),
511 Muxed::Unknown(session) => Event::default().event("unknown").data(session_data(
512 &session,
513 "error",
514 "\"unknown session\"",
515 )),
516 };
517 Ok::<_, std::convert::Infallible>(event)
518 });
519 Sse::new(frames)
520 .keep_alive(KeepAlive::default())
521 .into_response()
522}
523
524#[cfg(test)]
525#[allow(clippy::unwrap_used)]
526mod tests {
527 use super::*;
528
529 #[test]
530 fn a_cursor_round_trips_through_its_event_id_form() {
531 let mut cursor = BTreeMap::new();
532 cursor.insert("019a-b".to_owned(), 17);
533 cursor.insert("019a-c".to_owned(), 4);
534 let raw = format_cursor(&cursor);
535 assert_eq!(raw, "019a-b:17,019a-c:4");
536 assert_eq!(parse_cursor(&raw), cursor);
537 }
538
539 #[test]
540 fn a_malformed_cursor_pair_is_skipped_not_fatal() {
541 let parsed = parse_cursor("a:1,nonsense,:3,b:x,c:9");
542 assert_eq!(parsed.len(), 2);
543 assert_eq!(parsed.get("a"), Some(&1));
544 assert_eq!(parsed.get("c"), Some(&9));
545 }
546
547 #[test]
548 fn interest_reads_repeated_sessions_flags_and_an_encoded_cursor() {
549 let interest = Interest::parse(Some(
550 "session=a&session=b&session=a&host=1&cursor=a%3A5%2Cb%3A6",
551 ));
552 assert_eq!(interest.sessions, vec!["a", "b"]);
553 assert!(interest.host);
554 assert!(!interest.config);
555 assert_eq!(interest.cursor.get("a"), Some(&5));
556 assert_eq!(interest.cursor.get("b"), Some(&6));
557 }
558}
559

1//! Session event streams (docs/design/48-web-client.md §4.4, §4.7).
2//!
3//! Each subscription a client can hold — a session's agent events, its
4//! presentation frames, its side-chat branch, its coworking signal, the host
5//! and config changes — is built here exactly once. The single-session
6//! routes (`/sessions/{id}/events`, `/sessions/{id}/presentation/events`)
7//! serve one of them each for clients that follow one session, such as
8//! `vak term`. `GET /stream` merges any set of them onto ONE connection.
9//!
10//! The merge exists because a browser allows six HTTP/1.1 connections per
11//! host across all of its tabs, and HTTP/2 is not available on plain-http
12//! loopback. A tab that held one EventSource per subscription exhausted the
13//! pool with two or three tabs open, and every ordinary `fetch` then queued
14//! behind streams that never end.
15
16use std::collections::BTreeMap;
17use std::pin::Pin;
18use std::sync::Arc;
19
20use axum::extract::State;
21use axum::http::StatusCode;
22use axum::response::sse::{Event, KeepAlive, Sse};
23use axum::response::{IntoResponse, Response};
24use futures::stream::{self, Stream, StreamExt};
25use tokio_stream::wrappers::{BroadcastStream, IntervalStream};
26use vak_agent::AgentEvent;
27
28use crate::{AppState, SessionHandle};
29
30type Frames<T> = Pin<Box<dyn Stream<Item = T> + Send>>;
31
32/// Most sessions one `/stream` connection may follow. A client follows the
33/// sessions on screen plus any best-of-N candidates; anything past this is a
34/// client bug, refused rather than turned into unbounded server work.
35pub(crate) const MAX_STREAM_SESSIONS: usize = 32;
36
37/// One item of a session's agent-event subscription. `Event` already carries
38/// `crate::client_events::ClientEvent`, the same projection the single-
39/// session `/sessions/{id}/events` route applies through `seq_frame` — a
40/// multiplexed connection must not become a second, unprojected path for
41/// internal `AgentEvent` traffic (retry reasons, route legs, raw error text)
42/// to reach a client (docs/audits Finding 1).
43pub(crate) enum AgentFrame {
44 /// The resume point is older than the replay ring, or this consumer
45 /// lagged: what the client holds may be missing events it cannot see.
46 Resync(&'static str),
47 Event {
48 seq: u64,
49 event: crate::client_events::ClientEvent,
50 },
51}
52
53/// A session's agent events, resuming after `resume` when given.
54///
55/// Subscribes BEFORE reading the replay ring, so an event published between
56/// the two is received live rather than falling into the gap between them.
57/// Duplicates are filtered by sequence number; a gap could not be recovered.
58/// `seq` stays monotonic across a projected-away event: it produces no
59/// frame, identically on replay and live, exactly like `seq_frame`.
60pub(crate) fn agent_frames(handle: &SessionHandle, resume: Option<u64>) -> Frames<AgentFrame> {
61 let rx = handle.events_tx.subscribe();
62
63 // `None` from the ring means it no longer reaches back that far, and the
64 // client is told to rebuild from the durable transcript rather than being
65 // handed a stream with a hole in it that it cannot see.
66 let (replay, resync) = match resume {
67 Some(seq) => match handle.events_tx.replay_after(seq) {
68 Some(missed) => (missed, false),
69 None => (Vec::new(), true),
70 },
71 None => (Vec::new(), false),
72 };
73 let highest_replayed = replay.last().map(|e| e.seq).unwrap_or(0);
74
75 handle.subscribed.notify_one();
76 handle.events_tx.send(AgentEvent::StreamOpened);
77
78 let resync = resync.then_some(AgentFrame::Resync("events older than the replay window"));
79 let live = BroadcastStream::new(rx).filter_map(move |event| {
80 std::future::ready(match event {
81 // At or below what the replay already delivered is a duplicate.
82 Ok(framed) if framed.seq <= highest_replayed => None,
83 Ok(framed) => {
84 crate::client_events::project(framed.event).map(|event| AgentFrame::Event {
85 seq: framed.seq,
86 event,
87 })
88 }
89 Err(_) => Some(AgentFrame::Resync("live event consumer lagged")),
90 })
91 });
92 Box::pin(
93 stream::iter(resync)
94 .chain(stream::iter(replay.into_iter().filter_map(|framed| {
95 crate::client_events::project(framed.event).map(|event| AgentFrame::Event {
96 seq: framed.seq,
97 event,
98 })
99 })))
100 .chain(live),
101 )
102}
103
104/// One serialized `OutputStreamFrame`, with the sequence it brings the
105/// client up to when it has one.
106pub(crate) struct PresentationFrame {
107 pub(crate) sequence: Option<u64>,
108 pub(crate) json: String,
109}
110
111fn presentation_frame(frame: &vak_delivery::OutputStreamFrame) -> PresentationFrame {
112 PresentationFrame {
113 sequence: frame.sequence,
114 json: serde_json::to_string(frame).unwrap_or_else(|error| {
115 serde_json::json!({
116 "error": "presentation serialization failed",
117 "detail": error.to_string(),
118 })
119 .to_string()
120 }),
121 }
122}
123
124/// A session's presentation frames: an authoritative snapshot first, then
125/// one frame per projected live event. A session with no live handle gets
126/// its historical snapshot and nothing after it; `None` is an unknown id.
127///
128/// Snapshot-based on purpose: a reconnect receives a fresh snapshot, whose
129/// sequence becomes the new cursor. Delta replay is never manufactured from
130/// an unknown historical baseline.
131pub(crate) fn presentation_frames(
132 state: &AppState,
133 id: &str,
134 handle: Option<Arc<SessionHandle>>,
135) -> Option<Frames<PresentationFrame>> {
136 let Some(handle) = handle else {
137 let session = crate::open_historical_session(state, id)?;
138 let frame = vak_delivery::OutputStreamFrame {
139 sequence: None,
140 delta: None,
141 snapshot: Some(crate::projection::snapshot(id, &session)),
142 };
143 return Some(Box::pin(stream::once(std::future::ready(
144 presentation_frame(&frame),
145 ))));
146 };
147 let id = id.to_owned();
148 let rx = handle.events_tx.subscribe();
149 let mut timeline = {
150 let guard = handle
151 .session
152 .lock()
153 .unwrap_or_else(std::sync::PoisonError::into_inner);
154 guard
155 .as_ref()
156 .map(|session| crate::live_presentation_snapshot(&handle.core, &id, session))
157 .unwrap_or_else(|| {
158 handle
159 .presentation
160 .lock()
161 .unwrap_or_else(std::sync::PoisonError::into_inner)
162 .clone()
163 })
164 };
165 let mut last_sequence = live_cursor(&timeline).unwrap_or(0);
166 let initial = presentation_frame(&vak_delivery::OutputStreamFrame {
167 sequence: Some(last_sequence),
168 delta: None,
169 snapshot: Some(timeline.clone()),
170 });
171 handle.subscribed.notify_one();
172 let live = BroadcastStream::new(rx).filter_map(move |event| {
173 let frame = match event {
174 Ok(framed) => {
175 if framed.seq <= last_sequence {
176 None
177 } else {
178 last_sequence = framed.seq;
179 if matches!(&framed.event, AgentEvent::RunFinished { .. }) {
180 // The run owner publishes the rebuilt durable
181 // projection before broadcasting RunFinished. Rebase
182 // this long-lived subscriber now: carrying its private
183 // live timeline into the next run would otherwise
184 // replace settled cards with the prior run's
185 // prose/progress snapshot.
186 timeline = handle
187 .presentation
188 .lock()
189 .unwrap_or_else(std::sync::PoisonError::into_inner)
190 .clone();
191 Some(crate::projection::settled_frame(last_sequence, &timeline))
192 } else {
193 crate::projection::project_frame(&mut timeline, framed)
194 }
195 }
196 }
197 Err(_) => {
198 if let Some(events) = handle.events_tx.replay_after(last_sequence) {
199 for framed in events {
200 last_sequence = framed.seq;
201 crate::projection::project_frame(&mut timeline, framed);
202 }
203 } else {
204 let guard = handle
205 .session
206 .lock()
207 .unwrap_or_else(std::sync::PoisonError::into_inner);
208 timeline = guard
209 .as_ref()
210 .map(|session| crate::projection::snapshot(&id, session))
211 .unwrap_or_else(|| {
212 handle
213 .presentation
214 .lock()
215 .unwrap_or_else(std::sync::PoisonError::into_inner)
216 .clone()
217 });
218 last_sequence = live_cursor(&timeline).unwrap_or(last_sequence);
219 timeline
220 .diagnostics
221 .push("Presentation stream resynchronized after a gap.".into());
222 }
223 Some(vak_delivery::OutputStreamFrame {
224 sequence: Some(last_sequence),
225 delta: None,
226 snapshot: Some(timeline.clone()),
227 })
228 }
229 };
230 std::future::ready(frame.map(|frame| presentation_frame(&frame)))
231 });
232 Some(Box::pin(
233 stream::once(std::future::ready(initial)).chain(live),
234 ))
235}
236
237fn live_cursor(timeline: &vak_delivery::OutputTimeline) -> Option<u64> {
238 timeline
239 .cursor
240 .as_deref()
241 .and_then(|cursor| cursor.strip_prefix("live:"))
242 .and_then(|seq| seq.parse().ok())
243}
244
245/// One item of a session's side-chat (`/btw`) subscription, projected
246/// through the same `client_events::project` as the main agent stream
247/// (docs/audits Finding 1) — a side chat is a full agent run and deserves
248/// the same protection from internal traffic.
249enum SideFrame {
250 /// The broadcast consumer lagged; distinct from a projected-away event,
251 /// which simply produces no `Event` variant.
252 Lagged,
253 Event(crate::client_events::ClientEvent),
254}
255
256fn side_frames(handle: &SessionHandle) -> Frames<SideFrame> {
257 let mut rx = handle.side_events_tx.subscribe();
258 let _ = rx.try_recv();
259 handle.side_events_tx.send(AgentEvent::StreamOpened);
260 Box::pin(BroadcastStream::new(rx).filter_map(|result| {
261 std::future::ready(match result {
262 Ok(framed) => crate::client_events::project(framed.event).map(SideFrame::Event),
263 Err(_) => Some(SideFrame::Lagged),
264 })
265 }))
266}
267
268/// Content-free wakeups for a session's shared candidate comments.
269fn coworking_frames(handle: &SessionHandle) -> Frames<()> {
270 Box::pin(BroadcastStream::new(handle.coworking_comments_tx.subscribe()).map(|_| ()))
271}
272
273/// Host-level facts (active workspace, recents, terminal), sent once on
274/// connect and then only when they change. Sampled rather than broadcast:
275/// a workspace switch is a human action, and a two-second latency on it is
276/// not worth an event family of its own.
277fn host_frames(state: AppState) -> Frames<String> {
278 let mut last: Option<String> = None;
279 let ticks = IntervalStream::new(tokio::time::interval(std::time::Duration::from_secs(2)));
280 Box::pin(ticks.filter_map(move |_| {
281 let payload = serde_json::to_string(&crate::web::host_payload(&state))
282 .unwrap_or_else(|_| "{}".into());
283 let changed = last.as_deref() != Some(payload.as_str());
284 if changed {
285 last = Some(payload.clone());
286 }
287 std::future::ready(changed.then_some(payload))
288 }))
289}
290
291/// `ConfigChanged` events from the global hub, so a setting written by
292/// another surface is reflected without a restart (docs/design/44
293/// "Liveness").
294fn config_frames(state: &AppState) -> Frames<String> {
295 Box::pin(
296 BroadcastStream::new(state.hub.subscribe()).filter_map(|event| {
297 std::future::ready(match event {
298 Ok(event @ crate::events::SystemEvent::ConfigChanged { .. }) => {
299 serde_json::to_string(&event).ok()
300 }
301 _ => None,
302 })
303 }),
304 )
305}
306
307/// What `/stream` should carry, parsed from its query string.
308#[derive(Debug, Default, PartialEq, Eq)]
309pub(crate) struct Interest {
310 pub(crate) sessions: Vec<String>,
311 pub(crate) host: bool,
312 pub(crate) config: bool,
313 pub(crate) cursor: BTreeMap<String, u64>,
314}
315
316impl Interest {
317 fn parse(query: Option<&str>) -> Self {
318 let mut interest = Self::default();
319 for part in query.unwrap_or_default().split('&') {
320 let (key, value) = part.split_once('=').unwrap_or((part, ""));
321 let value = percent_encoding::percent_decode_str(&value.replace('+', " "))
322 .decode_utf8_lossy()
323 .into_owned();
324 match key {
325 "session" if !value.is_empty() && !interest.sessions.contains(&value) => {
326 interest.sessions.push(value);
327 }
328 "host" => interest.host = value == "1",
329 "config" => interest.config = value == "1",
330 "cursor" => interest.cursor = parse_cursor(&value),
331 _ => {}
332 }
333 }
334 interest
335 }
336}
337
338/// A `/stream` resume cursor: `<session>:<seq>` pairs joined by `,`.
339///
340/// It is the SSE event id of every agent frame, so a browser's own reconnect
341/// sends the whole vector back as `Last-Event-ID` and every session resumes
342/// from its own sequence. A client that reopens the stream itself (because
343/// its set of sessions changed) passes the same string as `?cursor=`.
344pub(crate) fn parse_cursor(raw: &str) -> BTreeMap<String, u64> {
345 raw.split(',')
346 .filter_map(|pair| {
347 let (session, seq) = pair.trim().rsplit_once(':')?;
348 Some((session.to_owned(), seq.parse().ok()?))
349 })
350 .filter(|(session, _)| !session.is_empty())
351 .collect()
352}
353
354pub(crate) fn format_cursor(cursor: &BTreeMap<String, u64>) -> String {
355 cursor
356 .iter()
357 .map(|(session, seq)| format!("{session}:{seq}"))
358 .collect::<Vec<_>>()
359 .join(",")
360}
361
362enum Muxed {
363 Host(String),
364 Config(String),
365 Agent(String, AgentFrame),
366 Presentation(String, PresentationFrame),
367 Side(String, SideFrame),
368 Coworking(String),
369 Unknown(String),
370}
371
372fn session_data(session: &str, key: &str, raw_json: &str) -> String {
373 format!(
374 "{{\"session\":{},\"{key}\":{raw_json}}}",
375 serde_json::Value::from(session)
376 )
377}
378
379fn client_event_json(event: &crate::client_events::ClientEvent) -> String {
380 serde_json::to_string(event).unwrap_or_else(|error| {
381 serde_json::json!({
382 "error": "event serialization failed",
383 "detail": error.to_string(),
384 })
385 .to_string()
386 })
387}
388
389/// `GET /stream` — every subscription a client holds, on one connection.
390///
391/// Query: `session=<id>` (repeatable), `host=1`, `config=1`, and optionally
392/// `cursor=` (see [`parse_cursor`]); a `Last-Event-ID` header wins over the
393/// query cursor. Every frame is a named SSE event whose JSON names the
394/// session it belongs to: `agent` (with the cursor as its id), `presentation`,
395/// `side`, `coworking`, `resync`, `unknown`, `host`, `config`.
396///
397/// Operator-only by construction: the participant middleware admits only
398/// `/sessions/<its conversation>/…` paths.
399pub(crate) async fn stream(
400 State(state): State<AppState>,
401 headers: axum::http::HeaderMap,
402 uri: axum::http::Uri,
403) -> Response {
404 let mut interest = Interest::parse(uri.query());
405 if let Some(header) = headers
406 .get("last-event-id")
407 .and_then(|value| value.to_str().ok())
408 .filter(|value| !value.trim().is_empty())
409 {
410 interest.cursor = parse_cursor(header);
411 }
412 if interest.sessions.len() > MAX_STREAM_SESSIONS {
413 return (
414 StatusCode::BAD_REQUEST,
415 format!("a stream follows at most {MAX_STREAM_SESSIONS} sessions"),
416 )
417 .into_response();
418 }
419 if interest.sessions.is_empty() && !interest.host && !interest.config {
420 return (
421 StatusCode::BAD_REQUEST,
422 "a stream needs at least one subscription",
423 )
424 .into_response();
425 }
426
427 let mut parts: Vec<Frames<Muxed>> = Vec::new();
428 if interest.host {
429 parts.push(Box::pin(host_frames(state.clone()).map(Muxed::Host)));
430 }
431 if interest.config {
432 parts.push(Box::pin(config_frames(&state).map(Muxed::Config)));
433 }
434 // Only sessions this connection follows keep a cursor entry; a session
435 // dropped from the set must not be resumed from a stale position later.
436 let mut cursor = BTreeMap::new();
437 for id in &interest.sessions {
438 let handle = crate::ensure_session_handle(&state, id)
439 .await
440 .ok()
441 .map(|(_, handle)| handle);
442 let presentation = presentation_frames(&state, id, handle.clone());
443 let Some(handle) = handle else {
444 match presentation {
445 Some(frames) => {
446 let session = id.clone();
447 parts.push(Box::pin(
448 frames.map(move |frame| Muxed::Presentation(session.clone(), frame)),
449 ));
450 }
451 None => parts.push(Box::pin(stream::once(std::future::ready(Muxed::Unknown(
452 id.clone(),
453 ))))),
454 }
455 continue;
456 };
457 let resume = interest.cursor.get(id).copied();
458 if let Some(seq) = resume {
459 cursor.insert(id.clone(), seq);
460 }
461 let session = id.clone();
462 parts.push(Box::pin(
463 agent_frames(&handle, resume).map(move |frame| Muxed::Agent(session.clone(), frame)),
464 ));
465 if let Some(frames) = presentation {
466 let session = id.clone();
467 parts.push(Box::pin(
468 frames.map(move |frame| Muxed::Presentation(session.clone(), frame)),
469 ));
470 }
471 let session = id.clone();
472 parts.push(Box::pin(
473 side_frames(&handle).map(move |event| Muxed::Side(session.clone(), event)),
474 ));
475 let session = id.clone();
476 parts.push(Box::pin(
477 coworking_frames(&handle).map(move |()| Muxed::Coworking(session.clone())),
478 ));
479 }
480
481 let frames = stream::select_all(parts).map(move |item| {
482 let event = match item {
483 Muxed::Host(payload) => Event::default().event("host").data(payload),
484 Muxed::Config(payload) => Event::default().event("config").data(payload),
485 Muxed::Agent(session, AgentFrame::Event { seq, event }) => {
486 cursor.insert(session.clone(), seq);
487 Event::default()
488 .event("agent")
489 .id(format_cursor(&cursor))
490 .data(session_data(&session, "event", &client_event_json(&event)))
491 }
492 Muxed::Agent(session, AgentFrame::Resync(reason)) => {
493 Event::default().event("resync").data(session_data(
494 &session,
495 "reason",
496 &serde_json::Value::from(reason).to_string(),
497 ))
498 }
499 Muxed::Presentation(session, frame) => Event::default()
500 .event("presentation")
501 .data(session_data(&session, "frame", &frame.json)),
502 Muxed::Side(session, SideFrame::Event(event)) => Event::default()
503 .event("side")
504 .data(session_data(&session, "event", &client_event_json(&event))),
505 Muxed::Side(session, SideFrame::Lagged) => Event::default()
506 .event("side")
507 .data(session_data(&session, "lagged", "true")),
508 Muxed::Coworking(session) => Event::default()
509 .event("coworking")
510 .data(session_data(&session, "refresh", "true")),
511 Muxed::Unknown(session) => Event::default().event("unknown").data(session_data(
512 &session,
513 "error",
514 "\"unknown session\"",
515 )),
516 };
517 Ok::<_, std::convert::Infallible>(event)
518 });
519 Sse::new(frames)
520 .keep_alive(KeepAlive::default())
521 .into_response()
522}
523
524#[cfg(test)]
525#[allow(clippy::unwrap_used)]
526mod tests {
527 use super::*;
528
529 #[test]
530 fn a_cursor_round_trips_through_its_event_id_form() {
531 let mut cursor = BTreeMap::new();
532 cursor.insert("019a-b".to_owned(), 17);
533 cursor.insert("019a-c".to_owned(), 4);
534 let raw = format_cursor(&cursor);
535 assert_eq!(raw, "019a-b:17,019a-c:4");
536 assert_eq!(parse_cursor(&raw), cursor);
537 }
538
539 #[test]
540 fn a_malformed_cursor_pair_is_skipped_not_fatal() {
541 let parsed = parse_cursor("a:1,nonsense,:3,b:x,c:9");
542 assert_eq!(parsed.len(), 2);
543 assert_eq!(parsed.get("a"), Some(&1));
544 assert_eq!(parsed.get("c"), Some(&9));
545 }
546
547 #[test]
548 fn interest_reads_repeated_sessions_flags_and_an_encoded_cursor() {
549 let interest = Interest::parse(Some(
550 "session=a&session=b&session=a&host=1&cursor=a%3A5%2Cb%3A6",
551 ));
552 assert_eq!(interest.sessions, vec!["a", "b"]);
553 assert!(interest.host);
554 assert!(!interest.config);
555 assert_eq!(interest.cursor.get("a"), Some(&5));
556 assert_eq!(interest.cursor.get("b"), Some(&6));
557 }
558}
559

Indexing the workspace…