capsule-runtime 0.6.0

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};

use serde::Serialize;
use tokio::sync::broadcast;

pub(crate) type SseBroadcast = broadcast::Sender<Arc<String>>;

/// Cadence at which an idle SSE connection writes [`SSE_HEARTBEAT_COMMENT`].
///
/// Shared by `message/stream` and `stream/watch` so the two endpoints present one
/// liveness contract. Not configurable.
pub(crate) const SSE_HEARTBEAT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(15);

/// The liveness signal written to an idle SSE connection.
///
/// An SSE comment, not a frame: it carries no `id:` line, is written straight to the
/// socket rather than through [`emit_frame`], and so consumes
/// no event id and never enters the replay buffer. A client that counts events sees
/// none of these.
pub(crate) const SSE_HEARTBEAT_COMMENT: &[u8] = b":heartbeat\n\n";

#[derive(Debug, Clone, Serialize)]
pub(crate) struct TaskStatusUpdateEvent {
    pub id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub context_id: Option<String>,
    pub status: StreamStatus,
    pub r#final: bool,
}

#[derive(Debug, Clone, Serialize)]
pub(crate) struct StreamStatus {
    pub state: String,
    pub message: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub response: Option<String>,
    /// The 1-based ordinal of the reopen this frame announces, on the `working` frame the reopen
    /// loop writes between two attempts of one task; `None`, and absent from the wire, on every
    /// other status.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reopen: Option<u32>,
}

#[derive(Debug, Clone, Serialize)]
pub(crate) struct TaskArtifactUpdateEvent {
    pub id: String,
    pub artifact: StreamArtifact,
}

/// A tool call the model has begun writing. Carries the call's id and name and nothing of its
/// input, which a hook or the call gate has not yet seen.
#[derive(Debug, Clone, Serialize)]
pub(crate) struct TaskToolCallStartedEvent {
    pub id: String,
    /// The id the call's answering `artifact` frame carries as its `tool_call_id`.
    pub tool_call_id: String,
    /// The capsule's tool name, the one the answering `artifact` frame carries.
    pub tool_name: String,
}

/// How much of a started call's input the model has written: a byte count, never the bytes.
#[derive(Debug, Clone, Serialize)]
pub(crate) struct TaskToolCallProgressEvent {
    pub id: String,
    pub tool_call_id: String,
    /// Bytes of the call's input JSON written so far. Never smaller than an earlier frame's for
    /// the same call.
    pub input_bytes: u64,
}

#[derive(Debug, Clone, Serialize)]
pub(crate) struct StreamArtifact {
    pub tool_name: String,
    pub content: String,
    /// The fence source naming what produced `content` — `tool:<name>` — or `None` when
    /// `content` carries no fence. It tells apart the frame shapes that are otherwise
    /// identical on the wire: an ordinary tool result, a skill result, a dispatch failure and
    /// a hook artifact all arrive as a `tool_name` and a `content`.
    ///
    /// Serialized on every frame, `null` included, without `skip_serializing_if`: a consumer
    /// that sees the key knows it is talking to a runtime that labels its frames, and one that
    /// sees no key knows it is not — which is a different thing from "this frame is unfenced".
    /// The conversation record's `fence` key is absent-when-unset instead, because a record
    /// line is read alongside four other envelope keys that all work that way.
    ///
    /// Naming a fence never justifies rewriting one: `content` is the bytes the model received,
    /// markers and all.
    pub fence_source: Option<String>,
    /// The provider's id for the call this frame reports, or `None` for a frame no call is
    /// behind (a hook artifact) and for a call the provider issued no id for. An empty provider
    /// id is `None` too, by the rule `TraceWriter::write_tool_call` applies to the same id.
    pub tool_call_id: Option<String>,
    /// Whether the call failed: the tool returned a status other than `passed`, or the dispatch
    /// never reached a tool. The call's `tool_call` or `skill_call` trace record spells the same
    /// fact as `"status":"error"`.
    pub is_error: bool,
    /// Wall-clock time of the dispatch, the same measurement the call's `tool_call` or
    /// `skill_call` trace record carries. `None` for a hook artifact.
    pub duration_ms: Option<u64>,
    /// The exit status of the subprocess this dispatch ran to completion, or `None` when it ran
    /// none — a WASM tool, a skill, a command demoted to the background, a failed dispatch, a
    /// hook artifact.
    pub exit_code: Option<i32>,
    /// The tool result's `truncated` flag, passed through unchanged: a tool's declaration that its
    /// output was cut short. The shell tool sets it when output exceeds its capture limit.
    pub truncated: bool,
}

// Every field above is serialized on every frame, `null` included, for the reason
// `fence_source` documents: an absent key tells a consumer the runtime does not report the fact,
// which is not the same as the fact not applying to this frame.
impl StreamArtifact {
    /// The frame for one dispatched tool call, whether it reached a tool or failed before one.
    // One parameter per wire field, so a call site names every value the frame carries.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn tool_call(
        tool_name: String,
        content: String,
        fence_source: Option<String>,
        tool_call_id: &str,
        is_error: bool,
        duration_ms: u64,
        exit_code: Option<i32>,
        truncated: bool,
    ) -> Self {
        Self {
            tool_name,
            content,
            fence_source,
            tool_call_id: (!tool_call_id.is_empty()).then(|| tool_call_id.to_string()),
            is_error,
            duration_ms: Some(duration_ms),
            exit_code,
            truncated,
        }
    }

    /// The frame forwarding one hook artifact. No tool call is behind it, so it has no id, no
    /// duration, no exit code and nothing that could have been truncated, and it never reports
    /// an error. A hook artifact is the capsule operator's own declared hook speaking, not
    /// content a tool fetched, so it reaches the model unfenced.
    pub(crate) fn hook(hook_name: String, payload: String) -> Self {
        Self {
            tool_name: hook_name,
            content: payload,
            fence_source: None,
            tool_call_id: None,
            is_error: false,
            duration_ms: None,
            exit_code: None,
            truncated: false,
        }
    }
}

/// An SSE event type the capsule's stream writes: the `event:` line of every frame on
/// `message/stream` and `stream/watch`. Every writer takes one, so this is the whole vocabulary
/// the agent card's stream extension is derived from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum StreamFrame {
    Status,
    Artifact,
    Text,
    Thinking,
    ToolCallStarted,
    ToolCallProgress,
    Gap,
    Lagged,
    ConnectionAck,
    CapsuleClosed,
    Error,
}

impl StreamFrame {
    /// Every frame, in the order the stream extension's `params.frames` lists them.
    pub(crate) const ALL: [StreamFrame; 11] = [
        StreamFrame::Status,
        StreamFrame::Artifact,
        StreamFrame::Text,
        StreamFrame::Thinking,
        StreamFrame::ToolCallStarted,
        StreamFrame::ToolCallProgress,
        StreamFrame::Gap,
        StreamFrame::Lagged,
        StreamFrame::ConnectionAck,
        StreamFrame::CapsuleClosed,
        StreamFrame::Error,
    ];

    /// The `event:` line's value, and the frame's name in `params.frames`.
    pub(crate) fn wire_name(self) -> &'static str {
        match self {
            StreamFrame::Status => "status",
            StreamFrame::Artifact => "artifact",
            StreamFrame::Text => "text",
            StreamFrame::Thinking => "thinking",
            StreamFrame::ToolCallStarted => "tool-call-started",
            StreamFrame::ToolCallProgress => "tool-call-progress",
            StreamFrame::Gap => "gap",
            StreamFrame::Lagged => "lagged",
            StreamFrame::ConnectionAck => "connection-ack",
            StreamFrame::CapsuleClosed => "capsule-closed",
            StreamFrame::Error => "error",
        }
    }
}

/// Format a single SSE event frame:
///   id: <event_id>\n
///   event: <event_type>\n
///   data: <json>\n\n
pub(crate) fn format_sse_event(event_id: u64, frame: StreamFrame, data: &str) -> String {
    let event_type = frame.wire_name();
    format!("id: {event_id}\nevent: {event_type}\ndata: {data}\n\n")
}

/// Format an SSE event frame with no `id:` line, for a frame that is written straight to one
/// connection and never takes a place in the session's sequence.
pub(crate) fn format_unnumbered_sse_event(frame: StreamFrame, data: &str) -> String {
    let event_type = frame.wire_name();
    format!("event: {event_type}\ndata: {data}\n\n")
}

/// The id of a frame formatted by [`format_sse_event`], read from its leading `id:` line, or
/// `None` for a frame that carries no id.
pub(crate) fn frame_id(frame: &str) -> Option<u64> {
    frame.strip_prefix("id: ")?.split('\n').next()?.parse().ok()
}

/// Decide whether a live frame is written to a connection that has already written every id up
/// to `last_written`, and advance `last_written` when it is.
///
/// A frame whose id is at or below `last_written` was already written by the replay (a frame
/// emitted between subscribing and reading the buffer arrives both ways), so it is dropped. A
/// frame with no id is always written and leaves `last_written` unchanged.
pub(crate) fn admit_live_frame(last_written: &mut Option<u64>, frame: &str) -> bool {
    let Some(id) = frame_id(frame) else {
        return true;
    };
    if last_written.is_some_and(|written| id <= written) {
        return false;
    }
    *last_written = Some(id);
    true
}

/// Result of a replay request from `SseEventBuffer::replay_from`.
pub(crate) enum ReplayResult {
    Complete(Vec<Arc<String>>),
    /// Frames after the cursor are no longer buffered, or the cursor was never issued by this
    /// session. `events` is the whole buffer; callers write a gap event before it.
    WithGap {
        first_available_id: u64,
        events: Vec<Arc<String>>,
    },
}

/// Format a gap SSE event indicating that buffered history starts at `first_available_id`.
pub(crate) fn format_gap_event(first_available_id: u64) -> String {
    format_unnumbered_sse_event(
        StreamFrame::Gap,
        &format!("{{\"first_available_id\":{first_available_id}}}"),
    )
}

/// Format the frame telling one SSE connection it lost `missed` live frames.
///
/// Written straight to that connection's socket, immediately before the next live frame
/// it receives: it carries no `id:` line, never passes through [`emit_frame`], and so
/// consumes no event id, never appears in a replay and never reaches any other connection.
/// `missed` is the count carried by the broadcast receiver's `RecvError::Lagged`, always at
/// least 1.
pub(crate) fn format_lagged_event(missed: u64) -> String {
    format_unnumbered_sse_event(StreamFrame::Lagged, &format!("{{\"missed\":{missed}}}"))
}

/// The session's event-id sequence and a bounded ring-buffer of the most recent frames for
/// reconnect replay.
///
/// One buffer serves a whole capsule session, so every numbered frame of every task takes its
/// id from `next_id`: the first frame is `1` and each later one is exactly one higher.
pub(crate) struct SseEventBuffer {
    events: VecDeque<(u64, Arc<String>)>,
    capacity: usize,
    /// Starts at `1`, not `0`: replay writes frames with an id strictly greater than the
    /// client's `Last-Event-ID`, and `0` is the cursor for "from the start", so a frame numbered
    /// `0` could never be replayed.
    next_id: u64,
}

impl SseEventBuffer {
    pub fn new(capacity: usize) -> Self {
        Self {
            events: VecDeque::new(),
            capacity,
            next_id: 1,
        }
    }

    /// Number, format and store one frame, returning its id and the formatted text.
    fn append(&mut self, frame: StreamFrame, data: &str) -> (u64, Arc<String>) {
        let id = self.next_id;
        self.next_id += 1;
        let event = Arc::new(format_sse_event(id, frame, data));
        if self.events.len() >= self.capacity {
            self.events.pop_front();
        }
        self.events.push_back((id, Arc::clone(&event)));
        (id, event)
    }

    /// Return the frames written after `last_id`, distinguishing a clean replay from one that
    /// cannot be: frames after `last_id` were evicted, or `last_id` was never issued by this
    /// session. Either of those returns the whole buffer as [`ReplayResult::WithGap`].
    pub fn replay_from(&self, last_id: u64) -> ReplayResult {
        let Some(oldest_id) = self.events.front().map(|(id, _)| *id) else {
            return ReplayResult::Complete(vec![]);
        };

        if last_id >= self.next_id || last_id < oldest_id - 1 {
            let events = self.events.iter().map(|(_, s)| Arc::clone(s)).collect();
            return ReplayResult::WithGap {
                first_available_id: oldest_id,
                events,
            };
        }

        let events = self
            .events
            .iter()
            .filter(|(id, _)| *id > last_id)
            .map(|(_, s)| Arc::clone(s))
            .collect();
        ReplayResult::Complete(events)
    }
}

/// Give one frame the session's next id, buffer it and broadcast it, returning the id.
///
/// The buffer lock is held across the broadcast send, so frames emitted concurrently — by the
/// agent loop, a tool's chunk host function and a `request-input` wait — enter the buffer and
/// reach every live receiver in id order. A send with no receivers is not an error.
pub(crate) fn emit_frame(
    tx: &SseBroadcast,
    buf: &Arc<Mutex<SseEventBuffer>>,
    frame: StreamFrame,
    data: &str,
) -> u64 {
    let mut buf = buf.lock().unwrap();
    let (id, event) = buf.append(frame, data);
    let _ = tx.send(event);
    id
}

/// Serialize and emit one SSE event through [`emit_frame`]. No-op when sse is None.
pub(crate) async fn emit_sse(
    sse: &Option<(SseBroadcast, Arc<Mutex<SseEventBuffer>>)>,
    frame: StreamFrame,
    data: &impl Serialize,
) {
    let Some((sse_tx, sse_buffer)) = sse else {
        return;
    };
    let data_str = match serde_json::to_string(data) {
        Ok(s) => s,
        Err(_) => return,
    };
    emit_frame(sse_tx, sse_buffer, frame, &data_str);
}

/// Whether a pre-formatted SSE event string is `task_id`'s final status: an `event: status`
/// frame whose data carries `"final": true` and `"id"` equal to `task_id`. A `text` frame with
/// `"final":true` (a cursor removal or the non-streaming fallback) is not, and neither is
/// another task's final status.
pub(crate) fn is_final_status_for(event: &str, task_id: &str) -> bool {
    let mut is_status = false;
    let mut data = None;
    for line in event.lines() {
        if line.split_once(": ") == Some(("event", StreamFrame::Status.wire_name())) {
            is_status = true;
        } else if let Some(rest) = line.strip_prefix("data: ") {
            data = Some(rest);
        }
    }
    if !is_status {
        return false;
    }
    let Some(parsed) = data.and_then(|data| serde_json::from_str::<serde_json::Value>(data).ok())
    else {
        return false;
    };
    parsed.get("final") == Some(&serde_json::Value::Bool(true))
        && parsed.get("id").and_then(serde_json::Value::as_str) == Some(task_id)
}

/// Emit one text chunk SSE event (synchronous, for use in `func_wrap` callbacks).
///
/// Wire format:
///   event: text
///   data: {"id":"<task_id>","text":"<chunk>","final":false}
pub(crate) fn emit_chunk_sse(
    tx: &SseBroadcast,
    buf: &Arc<Mutex<SseEventBuffer>>,
    task_id: &str,
    chunk: &str,
) {
    emit_frame(
        tx,
        buf,
        StreamFrame::Text,
        &format_text_sse_data(task_id, chunk, false),
    );
}

/// Emit one thinking chunk SSE event (synchronous).
///
/// Wire format:
///   event: thinking
///   data: {"id":"<task_id>","text":"<chunk>","final":false}
pub(crate) fn emit_thinking_chunk_sse(
    tx: &SseBroadcast,
    buf: &Arc<Mutex<SseEventBuffer>>,
    task_id: &str,
    chunk: &str,
) {
    emit_frame(
        tx,
        buf,
        StreamFrame::Thinking,
        &format_text_sse_data(task_id, chunk, false),
    );
}

/// Emit a final text SSE event (synchronous) — either cursor-removal (empty text) or
/// full-text fallback for non-streaming drivers.
///
/// Wire format:
///   event: text
///   data: {"id":"<task_id>","text":"<text>","final":true}
pub(crate) fn emit_chunk_sse_final(
    tx: &SseBroadcast,
    buf: &Arc<Mutex<SseEventBuffer>>,
    task_id: &str,
    text: &str,
) {
    emit_frame(
        tx,
        buf,
        StreamFrame::Text,
        &format_text_sse_data(task_id, text, true),
    );
}

fn format_text_sse_data(task_id: &str, text: &str, is_final: bool) -> String {
    let text_json = serde_json::to_string(text).unwrap_or_else(|_| "\"\"".to_string());
    format!(r#"{{"id":"{task_id}","text":{text_json},"final":{is_final}}}"#)
}

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

    fn frame_json(artifact: StreamArtifact) -> String {
        serde_json::to_string(&TaskArtifactUpdateEvent {
            id: "task_1".into(),
            artifact,
        })
        .unwrap()
    }

    #[test]
    fn tool_call_frame_serializes_every_key_in_declaration_order() {
        let json = frame_json(StreamArtifact::tool_call(
            "bash".into(),
            "hello".into(),
            Some("tool:bash".into()),
            "call_1",
            false,
            12,
            Some(0),
            false,
        ));
        assert_eq!(
            json,
            r#"{"id":"task_1","artifact":{"tool_name":"bash","content":"hello","fence_source":"tool:bash","tool_call_id":"call_1","is_error":false,"duration_ms":12,"exit_code":0,"truncated":false}}"#
        );
    }

    #[test]
    fn tool_call_started_frame_serializes_every_key_in_declaration_order() {
        let json = serde_json::to_string(&TaskToolCallStartedEvent {
            id: "task_1".into(),
            tool_call_id: "call_1".into(),
            tool_name: "write_file".into(),
        })
        .unwrap();
        assert_eq!(
            json,
            r#"{"id":"task_1","tool_call_id":"call_1","tool_name":"write_file"}"#
        );
        assert_eq!(
            format_sse_event(7, StreamFrame::ToolCallStarted, &json),
            "id: 7\nevent: tool-call-started\ndata: {\"id\":\"task_1\",\"tool_call_id\":\"call_1\",\"tool_name\":\"write_file\"}\n\n"
        );
    }

    #[test]
    fn tool_call_progress_frame_serializes_every_key_in_declaration_order() {
        let json = serde_json::to_string(&TaskToolCallProgressEvent {
            id: "task_1".into(),
            tool_call_id: "call_1".into(),
            input_bytes: 3172,
        })
        .unwrap();
        assert_eq!(
            json,
            r#"{"id":"task_1","tool_call_id":"call_1","input_bytes":3172}"#
        );
        assert_eq!(
            format_sse_event(8, StreamFrame::ToolCallProgress, &json),
            "id: 8\nevent: tool-call-progress\ndata: {\"id\":\"task_1\",\"tool_call_id\":\"call_1\",\"input_bytes\":3172}\n\n"
        );
    }

    #[test]
    fn tool_call_frame_passes_truncated_through() {
        let json = frame_json(StreamArtifact::tool_call(
            "read".into(),
            "partial".into(),
            None,
            "call_1",
            false,
            3,
            None,
            true,
        ));
        assert!(json.contains(r#""truncated":true"#), "{json}");
    }

    #[test]
    fn tool_call_frame_keeps_the_exit_code_key_with_or_without_a_subprocess() {
        let ran = frame_json(StreamArtifact::tool_call(
            "bash".into(),
            String::new(),
            None,
            "call_1",
            true,
            1204,
            Some(2),
            false,
        ));
        assert!(ran.contains(r#""exit_code":2"#), "{ran}");
        assert!(ran.contains(r#""is_error":true"#), "{ran}");

        let none = frame_json(StreamArtifact::tool_call(
            "write-file".into(),
            String::new(),
            None,
            "call_1",
            true,
            3,
            None,
            false,
        ));
        assert!(none.contains(r#""exit_code":null"#), "{none}");
    }

    #[test]
    fn tool_call_frame_records_an_empty_provider_id_as_null() {
        let empty = frame_json(StreamArtifact::tool_call(
            "bash".into(),
            String::new(),
            None,
            "",
            false,
            1,
            None,
            false,
        ));
        assert!(empty.contains(r#""tool_call_id":null"#), "{empty}");

        let issued = frame_json(StreamArtifact::tool_call(
            "bash".into(),
            String::new(),
            None,
            "toolu_01",
            false,
            1,
            None,
            false,
        ));
        assert!(issued.contains(r#""tool_call_id":"toolu_01""#), "{issued}");
    }

    #[test]
    fn hook_frame_reports_no_call_behind_it() {
        let json = frame_json(StreamArtifact::hook(
            "my-hook".into(),
            r#"{"reviewed":true}"#.into(),
        ));
        assert_eq!(
            json,
            r#"{"id":"task_1","artifact":{"tool_name":"my-hook","content":"{\"reviewed\":true}","fence_source":null,"tool_call_id":null,"is_error":false,"duration_ms":null,"exit_code":null,"truncated":false}}"#
        );
    }

    #[test]
    fn format_sse_event_produces_correct_frame() {
        let out = format_sse_event(3, StreamFrame::Status, r#"{"id":"x"}"#);
        assert_eq!(out, "id: 3\nevent: status\ndata: {\"id\":\"x\"}\n\n");
    }

    /// A buffer of `capacity` holding `count` frames, ids `1..=count`, each frame's data its id.
    fn buffer_with(capacity: usize, count: u64) -> SseEventBuffer {
        let mut buf = SseEventBuffer::new(capacity);
        for n in 1..=count {
            buf.append(StreamFrame::Status, &n.to_string());
        }
        buf
    }

    fn ids(events: &[Arc<String>]) -> Vec<u64> {
        events.iter().map(|e| frame_id(e).unwrap()).collect()
    }

    fn complete(result: ReplayResult) -> Vec<u64> {
        match result {
            ReplayResult::Complete(events) => ids(&events),
            ReplayResult::WithGap { .. } => panic!("expected Complete"),
        }
    }

    fn with_gap(result: ReplayResult) -> (u64, Vec<u64>) {
        match result {
            ReplayResult::WithGap {
                first_available_id,
                events,
            } => (first_available_id, ids(&events)),
            ReplayResult::Complete(_) => panic!("expected WithGap"),
        }
    }

    #[test]
    fn ids_start_at_one_and_rise_by_one() {
        let (tx, mut rx) = broadcast::channel(8);
        let buf = Arc::new(Mutex::new(SseEventBuffer::new(8)));
        assert_eq!(emit_frame(&tx, &buf, StreamFrame::Status, "{}"), 1);
        emit_chunk_sse(&tx, &buf, "t", "a");
        emit_thinking_chunk_sse(&tx, &buf, "t", "b");
        emit_chunk_sse_final(&tx, &buf, "t", "");
        assert_eq!(emit_frame(&tx, &buf, StreamFrame::Artifact, "{}"), 5);
        let received: Vec<u64> = std::iter::from_fn(|| rx.try_recv().ok())
            .map(|e| frame_id(&e).unwrap())
            .collect();
        assert_eq!(received, vec![1, 2, 3, 4, 5]);
    }

    #[test]
    fn replay_after_eviction_writes_a_gap() {
        let buf = buffer_with(3, 5);
        assert_eq!(with_gap(buf.replay_from(1)), (3, vec![3, 4, 5]));
        assert_eq!(with_gap(buf.replay_from(0)), (3, vec![3, 4, 5]));
        assert_eq!(complete(buf.replay_from(2)), vec![3, 4, 5]);
    }

    #[test]
    fn replay_from_zero_without_eviction_is_every_frame() {
        let buf = buffer_with(8, 5);
        assert_eq!(complete(buf.replay_from(0)), vec![1, 2, 3, 4, 5]);
        assert_eq!(complete(buf.replay_from(3)), vec![4, 5]);
        assert_eq!(complete(buf.replay_from(5)), Vec::<u64>::new());
    }

    #[test]
    fn replay_from_a_cursor_never_issued_writes_a_gap() {
        assert_eq!(
            with_gap(buffer_with(8, 5).replay_from(99)),
            (1, vec![1, 2, 3, 4, 5])
        );
        assert_eq!(
            with_gap(buffer_with(3, 5).replay_from(6)),
            (3, vec![3, 4, 5])
        );
    }

    #[test]
    fn replay_from_an_empty_buffer_is_empty() {
        let buf = SseEventBuffer::new(8);
        assert_eq!(complete(buf.replay_from(0)), Vec::<u64>::new());
        assert_eq!(complete(buf.replay_from(99)), Vec::<u64>::new());
    }

    #[test]
    fn concurrent_emitters_share_one_ascending_sequence() {
        const THREADS: u64 = 8;
        const PER_THREAD: u64 = 200;
        let total = THREADS * PER_THREAD;
        let (tx, mut rx) = broadcast::channel(total as usize);
        let buf = Arc::new(Mutex::new(SseEventBuffer::new(total as usize)));
        std::thread::scope(|scope| {
            for t in 0..THREADS {
                let (tx, buf) = (tx.clone(), Arc::clone(&buf));
                scope.spawn(move || {
                    for _ in 0..PER_THREAD {
                        if t % 2 == 0 {
                            emit_chunk_sse(&tx, &buf, "t", "x");
                        } else {
                            emit_frame(&tx, &buf, StreamFrame::Status, "{}");
                        }
                    }
                });
            }
        });

        let buffered = complete(buf.lock().unwrap().replay_from(0));
        let received: Vec<u64> = std::iter::from_fn(|| rx.try_recv().ok())
            .map(|e| frame_id(&e).unwrap())
            .collect();
        let expected: Vec<u64> = (1..=total).collect();
        assert_eq!(buffered, expected, "buffer order is id order");
        assert_eq!(received, expected, "live delivery order is id order");
    }

    #[test]
    fn frame_id_reads_the_leading_id_line() {
        assert_eq!(frame_id("id: 42\nevent: status\ndata: {}\n\n"), Some(42));
        assert_eq!(frame_id("event: gap\ndata: {}\n\n"), None);
        assert_eq!(frame_id(":heartbeat\n\n"), None);
    }

    #[test]
    fn live_frames_already_replayed_are_dropped() {
        let frame = |id| format_sse_event(id, StreamFrame::Status, "{}");
        let mut last = Some(3);
        assert!(!admit_live_frame(&mut last, &frame(2)));
        assert!(!admit_live_frame(&mut last, &frame(3)));
        assert!(admit_live_frame(&mut last, &frame(4)));
        assert_eq!(last, Some(4));
        assert!(
            !admit_live_frame(&mut last, &frame(4)),
            "an id is written once"
        );
        assert!(admit_live_frame(&mut last, &format_gap_event(1)));
        assert_eq!(last, Some(4), "a frame without an id leaves the cursor");

        let mut fresh = None;
        assert!(admit_live_frame(&mut fresh, &frame(1)));
        assert_eq!(fresh, Some(1));
    }

    #[test]
    fn unnumbered_frame_has_no_id_line() {
        let out = format_unnumbered_sse_event(StreamFrame::Status, r#"{"id":"x"}"#);
        assert_eq!(out, "event: status\ndata: {\"id\":\"x\"}\n\n");
        assert_eq!(frame_id(&out), None);
    }

    #[test]
    fn format_gap_event_has_correct_shape() {
        let s = format_gap_event(42);
        assert!(s.contains("event: gap\n"), "missing event field");
        assert!(s.contains("\"first_available_id\":42"), "missing id field");
        assert!(s.ends_with("\n\n"), "missing double newline terminator");
    }

    #[test]
    fn format_lagged_event_has_exact_wire_form() {
        let s = format_lagged_event(7);
        assert_eq!(s, "event: lagged\ndata: {\"missed\":7}\n\n");
        let data = s
            .lines()
            .find_map(|line| line.strip_prefix("data: "))
            .expect("data line");
        let parsed: serde_json::Value = serde_json::from_str(data).expect("data is JSON");
        assert_eq!(parsed, serde_json::json!({"missed": 7}));
    }

    /// One non-comment production line of a file under `src/`, other than a `*_tests.rs` file.
    struct CodeLine {
        /// The path relative to `src/`.
        file: String,
        /// The `fn`, or the `impl` header, the line most recently entered.
        item: String,
        number: usize,
        code: String,
    }

    fn production_code() -> Vec<CodeLine> {
        let mut out = Vec::new();
        for (file, source) in crate::source_scan::crate_sources() {
            if file.ends_with("_tests.rs") {
                continue;
            }
            let mut item = String::new();
            for (index, line) in crate::source_scan::production_part(&source)
                .lines()
                .enumerate()
            {
                let code = line.trim();
                if code.starts_with("//") {
                    continue;
                }
                if let Some((_, rest)) = code.split_once("fn ") {
                    item = rest.split('(').next().unwrap_or(rest).to_string();
                } else if code.starts_with("impl ") {
                    item = code.trim_end_matches('{').trim().to_string();
                }
                out.push(CodeLine {
                    file: file.clone(),
                    item: item.clone(),
                    number: index + 1,
                    code: code.to_string(),
                });
            }
        }
        out
    }

    /// Whether a source line writes an `event:` line itself: a string literal that opens with one,
    /// or one that carries one after a newline escape.
    fn writes_event_line(line: &str) -> bool {
        line.contains("\"event: ") || line.contains("\\nevent: ")
    }

    #[test]
    fn stream_frame_writers_all_go_through_the_enum() {
        let mut bypasses = Vec::new();
        let mut formatters = Vec::new();
        for line in production_code() {
            if !writes_event_line(&line.code) {
                continue;
            }
            let is_formatter = line.file == "streaming.rs"
                && matches!(
                    line.item.as_str(),
                    "format_sse_event" | "format_unnumbered_sse_event"
                );
            if is_formatter {
                formatters.push(format!("{}:{}", line.file, line.number));
            } else {
                bypasses.push(format!("{}:{}: {}", line.file, line.number, line.code));
            }
        }
        assert!(
            bypasses.is_empty(),
            "frames written without StreamFrame, outside format_sse_event and \
             format_unnumbered_sse_event:\n{}",
            bypasses.join("\n")
        );
        assert_eq!(
            formatters.len(),
            2,
            "one event: line in each formatter: {formatters:?}"
        );
    }

    /// A frame written straight to one streaming connection must be one `writes_frame` lets that
    /// method carry, or the card of a door that serves only the other method omits a frame it
    /// sends. Any other write goes to the session's broadcast, which both methods relay.
    #[test]
    fn stream_frame_every_write_is_carried_by_its_method() {
        use crate::identity::{writes_frame, DoorMethod, TransportKind};
        let mut uncarried = Vec::new();
        for line in production_code() {
            let methods: &[DoorMethod] = match (line.file.as_str(), line.item.as_str()) {
                ("streaming.rs", "impl StreamFrame" | "wire_name")
                | ("identity.rs", "writes_frame") => continue,
                ("identity.rs", "handle_message_stream" | "format_rejected_event") => {
                    &[DoorMethod::MessageStream]
                }
                ("identity.rs", "handle_stream_watch") => &[DoorMethod::StreamWatch],
                _ => &[DoorMethod::MessageStream, DoorMethod::StreamWatch],
            };
            let named = line.code.split("StreamFrame::").skip(1).map(|rest| {
                rest.split(|c: char| !c.is_alphanumeric())
                    .next()
                    .unwrap_or("")
            });
            for name in named {
                let Some(frame) = StreamFrame::ALL
                    .into_iter()
                    .find(|f| format!("{f:?}") == name)
                else {
                    continue;
                };
                for &method in methods {
                    if !TransportKind::ALL
                        .into_iter()
                        .any(|transport| writes_frame(method, frame, transport))
                    {
                        uncarried.push(format!(
                            "{}:{}: {frame:?} on {method:?}: {}",
                            line.file, line.number, line.code
                        ));
                    }
                }
            }
        }
        assert!(
            uncarried.is_empty(),
            "frames written on a method writes_frame does not carry them on:\n{}",
            uncarried.join("\n")
        );
    }

    #[test]
    fn stream_frame_wire_names_are_distinct() {
        let names: std::collections::BTreeSet<&str> =
            StreamFrame::ALL.iter().map(|f| f.wire_name()).collect();
        assert_eq!(names.len(), StreamFrame::ALL.len());
    }

    #[test]
    fn is_final_status_for_matches_only_this_tasks_final_status() {
        let status_final = "id: 5\nevent: status\ndata: {\"id\":\"x\",\"final\":true}\n\n";
        let other_final = "id: 6\nevent: status\ndata: {\"id\":\"xy\",\"final\":true}\n\n";
        let text_final =
            "id: 3\nevent: text\ndata: {\"id\":\"x\",\"text\":\"\",\"final\":true}\n\n";
        let status_nonfinal = "id: 1\nevent: status\ndata: {\"id\":\"x\",\"final\":false}\n\n";

        assert!(
            is_final_status_for(status_final, "x"),
            "this task's final status is terminal"
        );
        assert!(
            !is_final_status_for(other_final, "x"),
            "another task's final status is not this task's end, even when its id contains this one"
        );
        assert!(
            !is_final_status_for(status_final, "xy"),
            "an id that only contains the frame's id does not match it"
        );
        assert!(
            !is_final_status_for(text_final, "x"),
            "text:final:true is not terminal"
        );
        assert!(
            !is_final_status_for(status_nonfinal, "x"),
            "status:final:false is not terminal"
        );
    }
}