client-core: SSE framing, REST client and event stream

Adds sse.rs (a pure port of Sse.kt's frame parser), api.rs (a Transport
trait plus a ureq-backed implementation and an ApiClient covering the
session lifecycle: list/read, message/unqueue/answer,
interrupt/stop/start, title/cwd/model/permission-mode/notify,
command/compact, delete, and a transcript page), and event_stream.rs
(follow_session_events, mirroring EventStream.kt's reset/event split).

ureq rather than reqwest: server/ already depends on it for its own
outbound HTTPS, this stays blocking like Api.kt's HttpURLConnection
calls with no async runtime to carry, and its own PEM cert support
means no extra rustls/rustls-pemfile dependency to pin. Network I/O
sits behind Transport so ApiClient and follow_session_events are
tested with fakes, no server involved.

Not yet covered, tracked in CLIENT_CORE.md: setups, the file explorer,
usage, models, and attachments/import.

cargo test (78 passed), clippy --all-targets and fmt clean.

Co-Authored-By: Claude Sonnet <noreply@anthropic.com>

client-core/Cargo.lock

@@ -47,8 +47,6 @@ name = "client-core"
 version = "0.1.0"
 dependencies = [
  "event-model",
- "rustls",
- "rustls-pemfile",
  "serde",
  "serde_json",
  "tempfile",
@@ -484,15 +482,6 @@ dependencies = [
  "zeroize",
 ]
 
-[[package]]
-name = "rustls-pemfile"
-version = "2.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50"
-dependencies = [
- "rustls-pki-types",
-]
-
 [[package]]
 name = "rustls-pki-types"
 version = "1.15.1"

client-core/Cargo.toml

@@ -27,12 +27,6 @@ serde_json = { version = "1", features = ["float_roundtrip"] }
 # no need of an async runtime, and RUST.md's brief for this port is
 # "lightweight" throughout.
 ureq = { version = "3", features = ["json"] }
-# Verifying the server's pinned self-signed leaf against the CA `wg-app-link`
-# mints, the same way `ServerConfig.kt`'s `applyPinnedTls` does. rustls
-# rather than native-tls because ureq is already rustls-backed here and the
-# rest of this project's TLS goes through rustls too.
-rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] }
-rustls-pemfile = "2"
 
 [dev-dependencies]
 tempfile = "3"

client-core/src/api.rs

@@ -0,0 +1,543 @@
+//! The REST half of the backend's surface (see `server/src/routes.rs`'s
+//! module doc for the table); the SSE half is [`crate::event_stream`].
+//! Ported from `app/.../Api.kt`, but **not at full parity yet** -- see
+//! `CLIENT_CORE.md` for exactly which routes have a typed method here and
+//! which do not.
+//!
+//! Network I/O sits behind the [`Transport`] trait so the rest of this
+//! crate, and anything built on it, can be tested against a fake one with
+//! no server involved. [`UreqTransport`] is the only real implementation.
+
+use std::io::Read;
+
+use serde::Deserialize;
+use serde_json::Value;
+
+/// A request that did not produce what it asked for, carrying the server's
+/// own wording where it sent some.
+///
+/// `status` is the HTTP status where there was a response at all, and
+/// `None` where the server was never reached -- mirroring `ApiException` in
+/// `Api.kt`.
+#[derive(Debug, Clone)]
+pub struct ApiError {
+    pub message: String,
+    pub status: Option<u16>,
+}
+
+impl std::fmt::Display for ApiError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.message)
+    }
+}
+impl std::error::Error for ApiError {}
+
+/// A request body to send, in whichever of the two shapes the surface
+/// takes: `Api.kt`'s `jsonBody` and `streamBody`.
+pub enum Body {
+    Json(Value),
+    Bytes {
+        content_type: String,
+        bytes: Vec<u8>,
+    },
+}
+
+/// What a transport hands back for a REST call: the status and the body
+/// read whole. A streamed body ([`Transport::stream`]) is a different
+/// method because its whole point is not reading it whole.
+pub struct RawResponse {
+    pub status: u16,
+    pub body: Vec<u8>,
+}
+
+/// The network boundary this crate's pure logic is kept out from behind.
+/// `server/src/routes.rs`'s module doc is the surface this drives.
+pub trait Transport: Send + Sync {
+    /// One request/response call -- everything but the long-lived SSE GETs.
+    fn request(
+        &self,
+        method: &str,
+        path: &str,
+        body: Option<Body>,
+    ) -> Result<RawResponse, ApiError>;
+
+    /// Opens `path` and answers a reader over the response body, for a
+    /// caller that reads it as a stream rather than all at once (the SSE
+    /// connections in [`crate::event_stream`]). Fails the same way
+    /// [`Transport::request`] does for a non-2xx response.
+    fn stream(&self, path: &str) -> Result<Box<dyn Read + Send>, ApiError>;
+}
+
+/// One session as `GET /sessions` and `GET /sessions/{id}` report it.
+/// Mirrors `Api.kt`'s `SessionSummary`; see that type's doc for what each
+/// field means and why `setup` is never shown.
+#[derive(Debug, Clone, PartialEq, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SessionSummary {
+    pub id: String,
+    pub setup: String,
+    #[serde(default)]
+    pub keeps_own_transcript: bool,
+    pub setup_name: String,
+    pub provider: String,
+    pub title: String,
+    #[serde(default)]
+    pub model: Option<String>,
+    #[serde(default)]
+    pub permission_mode: Option<String>,
+    #[serde(default)]
+    pub imported: bool,
+    #[serde(default = "default_true")]
+    pub notify: bool,
+    #[serde(default)]
+    pub cwd: Option<String>,
+    #[serde(default)]
+    pub context_tokens: Option<u64>,
+    #[serde(default)]
+    pub max_image_edge: Option<u32>,
+    pub status: String,
+    pub last_activity: f64,
+}
+
+fn default_true() -> bool {
+    true
+}
+
+/// A client-core equivalent of `requestFromServer` plus the typed calls
+/// built on it. Holds no state of its own beyond the transport -- the
+/// session id or setup id a call is about is a parameter, per this
+/// project's "ask for the least you need".
+pub struct ApiClient<T: Transport> {
+    transport: T,
+}
+
+impl<T: Transport> ApiClient<T> {
+    pub fn new(transport: T) -> Self {
+        Self { transport }
+    }
+
+    fn json_request<R: for<'de> Deserialize<'de>>(
+        &self,
+        method: &str,
+        path: &str,
+        body: Option<Value>,
+    ) -> Result<R, ApiError> {
+        let raw = self.transport.request(method, path, body.map(Body::Json))?;
+        serde_json::from_slice(&raw.body).map_err(|e| ApiError {
+            message: format!("Reached the server but couldn't read its response ({e})"),
+            status: Some(raw.status),
+        })
+    }
+
+    fn empty_request(&self, method: &str, path: &str, body: Option<Value>) -> Result<(), ApiError> {
+        self.transport.request(method, path, body.map(Body::Json))?;
+        Ok(())
+    }
+
+    pub fn fetch_sessions(&self) -> Result<Vec<SessionSummary>, ApiError> {
+        self.json_request("GET", "/sessions", None)
+    }
+
+    pub fn fetch_session(&self, session_id: &str) -> Result<SessionSummary, ApiError> {
+        self.json_request("GET", &format!("/sessions/{session_id}"), None)
+    }
+
+    pub fn send_message(
+        &self,
+        session_id: &str,
+        text: &str,
+        attachment_ids: &[String],
+    ) -> Result<(), ApiError> {
+        self.empty_request(
+            "POST",
+            &format!("/sessions/{session_id}/message"),
+            Some(serde_json::json!({ "text": text, "attachmentIds": attachment_ids })),
+        )
+    }
+
+    pub fn unqueue_message(&self, session_id: &str, message_id: &str) -> Result<(), ApiError> {
+        self.empty_request(
+            "POST",
+            &format!("/sessions/{session_id}/unqueue"),
+            Some(serde_json::json!({ "messageId": message_id })),
+        )
+    }
+
+    pub fn answer_question(
+        &self,
+        session_id: &str,
+        question_id: &str,
+        answers: &[String],
+    ) -> Result<(), ApiError> {
+        self.empty_request(
+            "POST",
+            &format!("/sessions/{session_id}/answer"),
+            Some(serde_json::json!({ "questionId": question_id, "answers": answers })),
+        )
+    }
+
+    pub fn interrupt_session(&self, session_id: &str) -> Result<(), ApiError> {
+        self.empty_request("POST", &format!("/sessions/{session_id}/interrupt"), None)
+    }
+
+    pub fn stop_session(&self, session_id: &str) -> Result<(), ApiError> {
+        self.empty_request("POST", &format!("/sessions/{session_id}/stop"), None)
+    }
+
+    pub fn start_session(&self, session_id: &str) -> Result<(), ApiError> {
+        self.empty_request("POST", &format!("/sessions/{session_id}/start"), None)
+    }
+
+    pub fn rename_session(&self, session_id: &str, title: &str) -> Result<(), ApiError> {
+        self.empty_request(
+            "POST",
+            &format!("/sessions/{session_id}/title"),
+            Some(serde_json::json!({ "title": title })),
+        )
+    }
+
+    pub fn set_session_cwd(&self, session_id: &str, cwd: &str) -> Result<(), ApiError> {
+        self.empty_request(
+            "POST",
+            &format!("/sessions/{session_id}/cwd"),
+            Some(serde_json::json!({ "cwd": cwd })),
+        )
+    }
+
+    pub fn set_session_model(&self, session_id: &str, model: &str) -> Result<(), ApiError> {
+        self.empty_request(
+            "POST",
+            &format!("/sessions/{session_id}/model"),
+            Some(serde_json::json!({ "model": model })),
+        )
+    }
+
+    pub fn set_session_permission_mode(
+        &self,
+        session_id: &str,
+        mode: &str,
+    ) -> Result<(), ApiError> {
+        self.empty_request(
+            "POST",
+            &format!("/sessions/{session_id}/permission-mode"),
+            Some(serde_json::json!({ "permissionMode": mode })),
+        )
+    }
+
+    pub fn set_session_notify(&self, session_id: &str, notify: bool) -> Result<(), ApiError> {
+        self.empty_request(
+            "POST",
+            &format!("/sessions/{session_id}/notify"),
+            Some(serde_json::json!({ "notify": notify })),
+        )
+    }
+
+    pub fn run_command(&self, session_id: &str, text: &str) -> Result<(), ApiError> {
+        self.empty_request(
+            "POST",
+            &format!("/sessions/{session_id}/command"),
+            Some(serde_json::json!({ "text": text })),
+        )
+    }
+
+    pub fn compact_session(&self, session_id: &str) -> Result<(), ApiError> {
+        self.empty_request("POST", &format!("/sessions/{session_id}/compact"), None)
+    }
+
+    pub fn delete_session(&self, session_id: &str, delete_foreign: bool) -> Result<(), ApiError> {
+        let path = if delete_foreign {
+            format!("/sessions/{session_id}?deleteForeign=true")
+        } else {
+            format!("/sessions/{session_id}")
+        };
+        self.empty_request("DELETE", &path, None)
+    }
+
+    /// A page of transcript history. `before` is the newest-first cursor
+    /// (server default is "the newest page" when absent, which a caller
+    /// gets by passing `None`); the events themselves are handed back as
+    /// [`event_model::SeqEvent`] via `crate::event_stream`'s parsing, kept
+    /// out of this method's signature so a caller that only wants the raw
+    /// lines (for the transcript cache) is not forced to parse them.
+    pub fn fetch_transcript_page(
+        &self,
+        session_id: &str,
+        before: Option<u64>,
+        limit: u32,
+        coalesce: bool,
+    ) -> Result<Vec<Value>, ApiError> {
+        let mut path = format!("/sessions/{session_id}/transcript?limit={limit}");
+        if let Some(before) = before {
+            path.push_str(&format!("&before={before}"));
+        }
+        if coalesce {
+            path.push_str("&coalesce=true");
+        }
+        self.json_request("GET", &path, None)
+    }
+}
+
+/// The blocking [`Transport`] backed by `ureq`, the same crate `server/`
+/// already depends on for its own outbound HTTPS (`usage.rs`'s Anthropic
+/// poll). Verifies the server's leaf against a single pinned CA, the way
+/// `ServerConfig.kt`'s `applyPinnedTls` does, rather than the system trust
+/// store -- the server's certificate is self-signed on purpose (see
+/// `wg-app-link`).
+pub struct UreqTransport {
+    agent: ureq::Agent,
+    base_url: String,
+    token: String,
+}
+
+impl UreqTransport {
+    /// `ca_pem` is the CA certificate `wg-app-link`'s `enroll` minted,
+    /// exactly as read from `certs/ca.pem`.
+    pub fn new(
+        base_url: impl Into<String>,
+        token: impl Into<String>,
+        ca_pem: &[u8],
+    ) -> Result<Self, ApiError> {
+        let cert = ureq::tls::Certificate::from_pem(ca_pem).map_err(|e| ApiError {
+            message: format!("The pinned CA certificate could not be read: {e}"),
+            status: None,
+        })?;
+        let tls_config = ureq::tls::TlsConfig::builder()
+            .root_certs(ureq::tls::RootCerts::new_with_certs(&[cert]))
+            .build();
+        let agent: ureq::Agent = ureq::Agent::config_builder()
+            .tls_config(tls_config)
+            // Read the body ourselves on every status, the way
+            // `requestFromServer` does: the server's own error wording is
+            // in the body of a 4xx/5xx, and the default behaviour throws
+            // it away before this code can read it.
+            .http_status_as_error(false)
+            .timeout_connect(Some(std::time::Duration::from_secs(5)))
+            .build()
+            .into();
+        Ok(Self {
+            agent,
+            base_url: base_url.into(),
+            token: token.into(),
+        })
+    }
+
+    fn url(&self, path: &str) -> String {
+        format!("{}{}", self.base_url, path)
+    }
+}
+
+impl Transport for UreqTransport {
+    fn request(
+        &self,
+        method: &str,
+        path: &str,
+        body: Option<Body>,
+    ) -> Result<RawResponse, ApiError> {
+        let url = self.url(path);
+        let auth = format!("Bearer {}", self.token);
+        let mut builder = ureq::http::Request::builder()
+            .method(method)
+            .uri(&url)
+            .header("Authorization", &auth);
+        let response = match body {
+            None => builder
+                .body(())
+                .map_err(ureq::Error::from)
+                .and_then(|req| self.agent.run(req)),
+            Some(Body::Json(value)) => {
+                builder = builder.header("Content-Type", "application/json");
+                builder
+                    .body(serde_json::to_vec(&value).unwrap_or_default())
+                    .map_err(ureq::Error::from)
+                    .and_then(|req| self.agent.run(req))
+            }
+            Some(Body::Bytes {
+                content_type,
+                bytes,
+            }) => {
+                builder = builder.header("Content-Type", content_type);
+                builder
+                    .body(bytes)
+                    .map_err(ureq::Error::from)
+                    .and_then(|req| self.agent.run(req))
+            }
+        };
+        let mut response = response.map_err(|e| transport_error(&self.base_url, path, e))?;
+        let status = response.status().as_u16();
+        let mut body = Vec::new();
+        response
+            .body_mut()
+            .as_reader()
+            .read_to_end(&mut body)
+            .map_err(|e| ApiError {
+                message: format!("Reached {url} but couldn't read its response ({e})"),
+                status: Some(status),
+            })?;
+        if !(200..300).contains(&status) {
+            return Err(response_error(status, &body, path));
+        }
+        Ok(RawResponse { status, body })
+    }
+
+    fn stream(&self, path: &str) -> Result<Box<dyn Read + Send>, ApiError> {
+        let url = self.url(path);
+        let auth = format!("Bearer {}", self.token);
+        let response = self
+            .agent
+            .get(&url)
+            .header("Authorization", &auth)
+            .header("Accept", "text/event-stream")
+            // No read timeout: between events there is nothing to read for
+            // as long as the thing being followed is idle, mirroring
+            // `EventStream.kt`'s `readTimeout = 0`.
+            .config()
+            .timeout_recv_response(None)
+            .build()
+            .call();
+        let mut response = response.map_err(|e| transport_error(&self.base_url, path, e))?;
+        let status = response.status().as_u16();
+        if status != 200 {
+            let mut body = Vec::new();
+            let _ = response.body_mut().as_reader().read_to_end(&mut body);
+            return Err(response_error(status, &body, path));
+        }
+        Ok(Box::new(response.into_body().into_reader()))
+    }
+}
+
+fn transport_error(base_url: &str, path: &str, e: ureq::Error) -> ApiError {
+    ApiError {
+        message: format!(
+            "Couldn't reach the server at {base_url} ({e}) -- is ai-server running, and is this \
+             device able to reach that address (WireGuard up)? [{path}]"
+        ),
+        status: None,
+    }
+}
+
+/// The 401 wording matches `Api.kt`'s, since that message is instructions
+/// for the reader rather than a diagnostic -- see this project's UI rule
+/// about shortening a failure in one place rather than at each display site.
+fn response_error(status: u16, body: &[u8], path: &str) -> ApiError {
+    let detail = String::from_utf8_lossy(body).trim().to_string();
+    let message = if status == 401 {
+        "The server rejected this device's token. Re-enroll by scanning the server's QR (or \
+         rotate with --rotate-token and scan the new one)."
+            .to_string()
+    } else if detail.is_empty() {
+        format!("Server returned HTTP {status} for {path}")
+    } else {
+        detail
+    };
+    ApiError {
+        message,
+        status: Some(status),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Cursor;
+    use std::sync::Mutex;
+
+    /// A transport with no network at all, for the pure-logic tests this
+    /// module can run without a server.
+    #[derive(Default)]
+    struct FakeTransport {
+        responses: Mutex<Vec<(String, String, RawResponse)>>,
+    }
+
+    impl FakeTransport {
+        fn respond(&self, method: &str, path: &str, status: u16, body: &str) {
+            self.responses.lock().unwrap().push((
+                method.to_string(),
+                path.to_string(),
+                RawResponse {
+                    status,
+                    body: body.as_bytes().to_vec(),
+                },
+            ));
+        }
+    }
+
+    impl Transport for FakeTransport {
+        fn request(
+            &self,
+            method: &str,
+            path: &str,
+            _body: Option<Body>,
+        ) -> Result<RawResponse, ApiError> {
+            let mut responses = self.responses.lock().unwrap();
+            let index = responses
+                .iter()
+                .position(|(m, p, _)| m == method && p == path)
+                .ok_or_else(|| ApiError {
+                    message: format!("no fake response for {method} {path}"),
+                    status: None,
+                })?;
+            let (_, _, response) = responses.remove(index);
+            if !(200..300).contains(&response.status) {
+                return Err(response_error(response.status, &response.body, path));
+            }
+            Ok(response)
+        }
+
+        fn stream(&self, _path: &str) -> Result<Box<dyn Read + Send>, ApiError> {
+            Ok(Box::new(Cursor::new(Vec::new())))
+        }
+    }
+
+    #[test]
+    fn fetch_sessions_parses_the_list() {
+        let transport = FakeTransport::default();
+        transport.respond(
+            "GET",
+            "/sessions",
+            200,
+            r#"[{"id":"s1","setup":"m1","setupName":"desktop","provider":"claude_cli",
+               "title":"hi","status":"idle","lastActivity":1.0}]"#,
+        );
+        let client = ApiClient::new(transport);
+        let sessions = client.fetch_sessions().unwrap();
+        assert_eq!(sessions.len(), 1);
+        assert_eq!(sessions[0].id, "s1");
+        assert_eq!(sessions[0].setup_name, "desktop");
+        // Defaults for fields the server omits.
+        assert!(sessions[0].notify);
+        assert_eq!(sessions[0].model, None);
+    }
+
+    #[test]
+    fn a_401_gets_the_enrollment_message_regardless_of_the_bare_body() {
+        let transport = FakeTransport::default();
+        transport.respond("POST", "/sessions/s1/interrupt", 401, "unauthorized");
+        let client = ApiClient::new(transport);
+        let err = client.interrupt_session("s1").unwrap_err();
+        assert!(err.message.contains("Re-enroll"));
+        assert_eq!(err.status, Some(401));
+    }
+
+    #[test]
+    fn a_bare_error_status_with_no_body_falls_back_to_a_generic_message() {
+        let transport = FakeTransport::default();
+        transport.respond("POST", "/sessions/s1/stop", 500, "");
+        let client = ApiClient::new(transport);
+        let err = client.stop_session("s1").unwrap_err();
+        assert!(err.message.contains("500"));
+    }
+
+    #[test]
+    fn a_server_explanation_in_the_body_is_surfaced_verbatim() {
+        let transport = FakeTransport::default();
+        transport.respond(
+            "POST",
+            "/sessions/s1/cwd",
+            409,
+            "that path does not exist on this machine",
+        );
+        let client = ApiClient::new(transport);
+        let err = client.set_session_cwd("s1", "/nope").unwrap_err();
+        assert_eq!(err.message, "that path does not exist on this machine");
+    }
+}

client-core/src/event_stream.rs

@@ -0,0 +1,142 @@
+//! The SSE half of the API: one long-lived GET per open session screen,
+//! replaying the transcript after a cursor and then following it live.
+//! Ported from `app/.../EventStream.kt`; the framing itself is
+//! [`crate::sse`].
+
+use std::io::{BufRead, BufReader};
+
+use event_model::SeqEvent;
+
+use crate::api::{ApiError, Transport};
+use crate::sse::SseReader;
+
+/// The frame name the server uses to say a cursor was too far behind to
+/// continue from. Must match `send_backlog` in `server/src/routes.rs`.
+const RESET_EVENT: &str = "reset";
+
+/// One frame of a session's event stream, folded from the wire shape the
+/// caller needs to act on -- mirroring what `EventStream.kt`'s three
+/// callbacks were for, as a single enum instead, since Rust has no
+/// equivalent of handing three closures to one blocking call.
+pub enum StreamItem {
+    /// The connection was accepted; the measured moment the stream is live
+    /// (see `EventStream.kt`'s doc on `onOpen` for why this, not the first
+    /// event, is what clears a previous failure on screen).
+    Open,
+    /// The cursor was too far behind to continue from: everything already
+    /// displayed is stale, and the events that follow are a fresh window.
+    /// Arrives before those events, so a caller that clears on it stays in
+    /// order.
+    Reset,
+    /// One event, as both the raw line the transcript cache stores and the
+    /// parsed [`SeqEvent`] the fold works from -- they have to be the same
+    /// line, so both travel together rather than being parsed twice from
+    /// two call sites.
+    Event { raw: String, event: SeqEvent },
+}
+
+/// Follows `/sessions/{id}/events?after={after}`, calling `on_item` for
+/// each [`StreamItem`] until the connection drops or `on_item` asks to
+/// stop (by returning `false`). Reconnecting -- with the last seq seen as
+/// the new cursor -- is the caller's job, same as in the Kotlin version.
+pub fn follow_session_events(
+    transport: &dyn Transport,
+    session_id: &str,
+    after: u64,
+    mut on_item: impl FnMut(StreamItem) -> bool,
+) -> Result<(), ApiError> {
+    let path = format!("/sessions/{session_id}/events?after={after}");
+    let body = transport.stream(&path)?;
+    if !on_item(StreamItem::Open) {
+        return Ok(());
+    }
+    let mut lines = BufReader::new(body).lines();
+    let mut reader = SseReader::new();
+    while let Some(line) = lines.next().transpose().map_err(|e| ApiError {
+        message: format!("Can't reach the server -- retrying. ({e})"),
+        status: None,
+    })? {
+        let Some(frame) = reader.feed_line(&line) else {
+            continue;
+        };
+        // A named frame carries no payload and a data frame has no name.
+        if frame.name.as_deref() == Some(RESET_EVENT) {
+            if !on_item(StreamItem::Reset) {
+                return Ok(());
+            }
+        } else if !frame.data.is_empty() {
+            let event: SeqEvent = serde_json::from_str(&frame.data).map_err(|e| ApiError {
+                message: format!("The server sent an event this build couldn't parse: {e}"),
+                status: None,
+            })?;
+            if !on_item(StreamItem::Event {
+                raw: frame.data,
+                event,
+            }) {
+                return Ok(());
+            }
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::api::{Body, RawResponse};
+    use std::io::Cursor;
+
+    struct FixtureTransport {
+        body: &'static str,
+    }
+
+    impl Transport for FixtureTransport {
+        fn request(
+            &self,
+            _method: &str,
+            _path: &str,
+            _body: Option<Body>,
+        ) -> Result<RawResponse, ApiError> {
+            unimplemented!("this fixture only serves a stream")
+        }
+
+        fn stream(&self, _path: &str) -> Result<Box<dyn std::io::Read + Send>, ApiError> {
+            Ok(Box::new(Cursor::new(self.body.as_bytes().to_vec())))
+        }
+    }
+
+    #[test]
+    fn events_and_a_reset_frame_are_told_apart() {
+        let transport = FixtureTransport {
+            body: "event:reset\n\ndata:{\"seq\":1,\"ts\":1.0,\"type\":\"status\",\"state\":\"idle\"}\n\n",
+        };
+        let mut items = Vec::new();
+        follow_session_events(&transport, "s1", 0, |item| {
+            items.push(match item {
+                StreamItem::Open => "open".to_string(),
+                StreamItem::Reset => "reset".to_string(),
+                StreamItem::Event { event, .. } => format!("event:{}", event.seq),
+            });
+            true
+        })
+        .unwrap();
+        assert_eq!(items, vec!["open", "reset", "event:1"]);
+    }
+
+    #[test]
+    fn the_caller_can_stop_early() {
+        let transport = FixtureTransport {
+            body: "data:{\"seq\":1,\"ts\":1.0,\"type\":\"status\",\"state\":\"idle\"}\n\n\
+                   data:{\"seq\":2,\"ts\":1.0,\"type\":\"status\",\"state\":\"idle\"}\n\n",
+        };
+        let mut count = 0;
+        follow_session_events(&transport, "s1", 0, |item| {
+            if matches!(item, StreamItem::Event { .. }) {
+                count += 1;
+            }
+            count < 1
+        })
+        .unwrap();
+        assert_eq!(count, 1);
+    }
+}

client-core/src/lib.rs

@@ -3,7 +3,10 @@
 //! not yet.
 
 pub mod ansi;
+pub mod api;
+pub mod event_stream;
 pub mod highlight;
+pub mod sse;
 pub mod transcript_cache;
 
 pub use event_model::*;

client-core/src/sse.rs

@@ -0,0 +1,121 @@
+//! Server-sent-events framing, ported from `app/.../Sse.kt`: `data:` and
+//! `event:` lines accumulate until a blank line ends the frame, comments
+//! start with `:`, and a frame is either named with no payload or a payload
+//! with no name.
+//!
+//! Pure and line-at-a-time, unlike the Kotlin original which also owned the
+//! socket: `server/routes.rs`'s SSE bodies are one event per line, so a
+//! caller here feeds lines from wherever they came from (a real connection,
+//! a test fixture) and gets frames back with no I/O of its own -- which is
+//! what lets this be tested with no server, per RUST.md's "pure logic
+//! first" for this crate.
+
+/// One SSE frame: its name (`None` for an ordinary data frame) and its payload.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Frame {
+    pub name: Option<String>,
+    pub data: String,
+}
+
+/// Accumulates lines into [`Frame`]s. One instance per connection --
+/// `feed_line` is called for every line the transport reads (with line
+/// endings already stripped), and answers a frame when a blank line closes
+/// one.
+#[derive(Debug, Default)]
+pub struct SseReader {
+    data: String,
+    name: Option<String>,
+}
+
+impl SseReader {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Feeds one line (no trailing `\n`). Answers the frame this line
+    /// completed, if any.
+    pub fn feed_line(&mut self, line: &str) -> Option<Frame> {
+        if line.is_empty() {
+            if self.name.is_some() || !self.data.is_empty() {
+                let frame = Frame {
+                    name: self.name.take(),
+                    data: std::mem::take(&mut self.data),
+                };
+                return Some(frame);
+            }
+            return None;
+        }
+        if let Some(rest) = line.strip_prefix("data:") {
+            self.data.push_str(rest.trim());
+        } else if let Some(rest) = line.strip_prefix("event:") {
+            self.name = Some(rest.trim().to_string());
+        }
+        // `id:`, comments -- nothing to do.
+        None
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn frames(lines: &[&str]) -> Vec<Frame> {
+        let mut reader = SseReader::new();
+        lines.iter().filter_map(|l| reader.feed_line(l)).collect()
+    }
+
+    #[test]
+    fn a_data_only_frame_has_no_name() {
+        assert_eq!(
+            frames(&["data:hello", ""]),
+            vec![Frame {
+                name: None,
+                data: "hello".to_string()
+            }]
+        );
+    }
+
+    #[test]
+    fn a_named_frame_with_no_payload_still_completes() {
+        assert_eq!(
+            frames(&["event:reset", ""]),
+            vec![Frame {
+                name: Some("reset".to_string()),
+                data: String::new()
+            }]
+        );
+    }
+
+    #[test]
+    fn a_blank_line_with_nothing_pending_yields_no_frame() {
+        assert_eq!(frames(&[""]), vec![]);
+    }
+
+    #[test]
+    fn a_comment_and_an_id_line_are_ignored() {
+        assert_eq!(
+            frames(&[":keepalive", "id:5", "data:hi", ""]),
+            vec![Frame {
+                name: None,
+                data: "hi".to_string()
+            }]
+        );
+    }
+
+    #[test]
+    fn two_frames_in_a_row_are_both_reported() {
+        assert_eq!(
+            frames(&["data:one", "", "data:two", ""]),
+            vec![
+                Frame {
+                    name: None,
+                    data: "one".to_string()
+                },
+                Frame {
+                    name: None,
+                    data: "two".to_string()
+                },
+            ]
+        );
+    }
+}