klieo-workflow-api 3.3.0

//! `GET /runs/{id}/events` — Server-Sent Events of a run's live progress.
//!
//! This is **live tailing, not replay**: events emitted before a client
//! connects are lost — a client that attaches mid-run sees only events from
//! attach onward. For the complete history use the run log; this endpoint is
//! a per-process best-effort overlay (a run's stream lives only in the
//! process running it).
//!
//! Each run's channel buffers up to a bounded number of events (see
//! [`crate::progress`]); a subscriber that lags past it has the gap skipped (a
//! `Lagged`) and the stream CONTINUES — it never closes on lag. The stream
//! ends only when the run finalizes (its sender is dropped).

use crate::auth::can_read_run;
use crate::error::WorkflowApiError;
use crate::state::WorkflowRunState;
use axum::extract::{Path, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::Extension;
use klieo_auth_common::Identity;
use klieo_core::agent::AgentEvent;
use std::convert::Infallible;
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::{Stream, StreamExt};

/// `GET /runs/{id}/events` — stream the run's live progress. Returns `404`
/// when no run is currently live under `run_id` (unknown, already
/// finalized — poll `GET /runs/{id}` for terminal state — or authored by
/// someone else and the caller lacks the cross-author read scope; see
/// [`can_read_run`]).
pub(crate) async fn get_run_events(
    State(state): State<Arc<WorkflowRunState>>,
    Extension(identity): Extension<Identity>,
    Path(run_id): Path<String>,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, WorkflowApiError> {
    let author = run_author(&state, &run_id).await?;
    if !can_read_run(&identity, &author) {
        return Err(WorkflowApiError::RunNotFound);
    }
    let receiver = state
        .progress
        .subscribe(&run_id)
        .ok_or(WorkflowApiError::RunNotFound)?;
    let stream = event_json_stream(receiver).map(|data| Ok(Event::default().data(data)));
    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}

/// The run's author, read from the (authoritative-for-identity) run store —
/// the progress hub is keyed only by `run_id`, with no author of its own.
async fn run_author(state: &WorkflowRunState, run_id: &str) -> Result<String, WorkflowApiError> {
    match state.run_store.get(run_id).await {
        Ok(Some(view)) => Ok(view.author),
        Ok(None) => Err(WorkflowApiError::RunNotFound),
        Err(e) => Err(WorkflowApiError::Internal(Box::new(e))),
    }
}

/// Map a run's broadcast receiver to a stream of JSON event payloads. A
/// `Lagged` (slow subscriber) or any recv error is skipped and the stream
/// continues; it ends only when the sender is dropped at run finalize.
fn event_json_stream(receiver: broadcast::Receiver<AgentEvent>) -> impl Stream<Item = String> {
    BroadcastStream::new(receiver).filter_map(|result| {
        let event = result.ok()?;
        serde_json::to_string(&event).ok()
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn lagged_receiver_skips_gap_and_keeps_streaming() {
        let (sender, receiver) = broadcast::channel(2);
        // Overflow the 2-slot buffer without consuming → the receiver lags.
        for _ in 0..3 {
            let _ = sender.send(AgentEvent::LlmCallStarted);
        }
        sender
            .send(AgentEvent::ToolCallStarted {
                name: "post-lag".into(),
            })
            .unwrap();
        drop(sender); // close so the stream terminates once drained

        let events: Vec<String> = event_json_stream(receiver).collect().await;
        assert!(
            !events.is_empty(),
            "the stream must continue past the Lagged gap"
        );
        assert!(
            events.last().unwrap().contains("post-lag"),
            "a post-lag event is still delivered: {events:?}"
        );
    }
}