From 7f8b8be978c81065be01a44ac14aa62edecb6b96 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 1 Aug 2026 23:53:20 +0000 Subject: [PATCH 1/3] feat(protocol): add a shared subscription-open handshake MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Subscription-style requests (/events/subscribe, /logs/stream) answer on the bidi stream before opening the data stream, and answer with an error in routine cases: server_busy from the stream semaphore, requirements_invalid or not_found from /logs/stream validation, server_busy from a journal-open failure. Every consumer hand-rolled the client half of that handshake and each got a different subset wrong. open_subscription reads the envelope to FIN (the framing i[stream.control] actually specifies, not read_line), classifies it through the same parse_response as request(), and only then awaits the data stream — under a timeout, because the server's own open_uni failure path merely logs. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HtxQdsF6YhLzz9RrDmHEDv --- crates/protocol/src/client.rs | 353 +++++++++++++++++++++++++++++++--- docs/spec/interface.md | 6 + 2 files changed, 335 insertions(+), 24 deletions(-) diff --git a/crates/protocol/src/client.rs b/crates/protocol/src/client.rs index 8765aa53..80ab5629 100644 --- a/crates/protocol/src/client.rs +++ b/crates/protocol/src/client.rs @@ -13,7 +13,6 @@ use rustls_pki_types::{CertificateDer, ServerName, SubjectPublicKeyInfoDer, Unix use serde_json::Value; use sha2::{Digest, Sha256}; use subtle::ConstantTimeEq; -use tokio::io::{AsyncBufReadExt as _, BufReader}; // --------------------------------------------------------------------------- // Error type @@ -253,6 +252,18 @@ impl ServerCertVerifier for RecordingVerifier { // OiClient // --------------------------------------------------------------------------- +/// How long a client waits for the data stream after the server has accepted a +/// subscription. +const SUBSCRIBE_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10); + +/// Upper bound on a subscription's response envelope. Subscription responses +/// are `{"result":{}}` or a short error, so this is generous. +const RESPONSE_LIMIT: usize = 64 * 1024; + +/// Upper bound on an ordinary request's response body, which may carry +/// listings and script text. +const REQUEST_RESPONSE_LIMIT: usize = 4 * 1024 * 1024; + pub struct OiClient { conn: Connection, actor: Actor, @@ -265,6 +276,20 @@ impl OiClient { auth: ClientAuth, identity: &ClientIdentity, actor: Actor, + ) -> Result { + Self::connect_from(addr, "[::]:0".parse().unwrap(), auth, identity, actor).await + } + + /// [`Self::connect`] with the local socket address made explicit. + /// + /// Production always binds the dual-stack wildcard; tests bind an IPv4 + /// wildcard so they run on hosts without IPv6. + async fn connect_from( + addr: SocketAddr, + bind: SocketAddr, + auth: ClientAuth, + identity: &ClientIdentity, + actor: Actor, ) -> Result { let verifier: Arc = match auth { ClientAuth::Fingerprint(fp) => Arc::new(FingerprintVerifier { expected: fp }), @@ -287,8 +312,7 @@ impl OiClient { let mut client_cfg = ClientConfig::new(Arc::new(quic_config)); client_cfg.transport_config(Arc::new(transport)); - let mut endpoint = Endpoint::client("[::]:0".parse().unwrap()) - .map_err(|e| ClientError::Connect(Box::new(e)))?; + let mut endpoint = Endpoint::client(bind).map_err(|e| ClientError::Connect(Box::new(e)))?; endpoint.set_default_client_config(client_cfg); let conn = tokio::time::timeout( @@ -334,39 +358,89 @@ impl OiClient { .map_err(|e| ClientError::Transport(Box::new(e))) } - /// Subscribe to the server's event stream. + /// Open a subscription-style request and return the server-initiated + /// unidirectional stream carrying its data. /// - /// Sends `/events/subscribe`, discards the initial `{"result":{}}` line, - /// then accepts the server-initiated unidirectional stream the daemon opens - /// to push newline-delimited JSON events. Returns that stream. - pub async fn subscribe_events(&self) -> Result { - let (mut send, recv) = self + /// This is the only correct way to drive `/events/subscribe` and + /// `/logs/stream`: the response envelope on the bidirectional stream is + /// read to FIN and classified before the unidirectional stream is awaited, + /// so an error response surfaces as [`ClientError::Api`] instead of + /// parking the caller on a stream the server will never open. + // i[impl stream.subscribe] + pub async fn open_subscription( + &self, + method: &str, + params: Value, + ) -> Result { + let req = serde_json::to_vec(&serde_json::json!({ + "method": method, + "actor": &self.actor, + "params": params, + })) + .expect("request serialisation never fails"); + self.open_subscription_raw(&req).await + } + + /// As [`Self::open_subscription`], but sending a request whose envelope the + /// caller has already serialised. + /// + /// Only for callers that must preserve an envelope built elsewhere — the + /// web gateway relays the browser session's actor rather than its own. + // i[impl stream.subscribe] + pub async fn open_subscription_raw(&self, request: &[u8]) -> Result { + self.open_subscription_within(request, SUBSCRIBE_HANDSHAKE_TIMEOUT) + .await + } + + /// The handshake itself, with the data-stream wait bounded by `timeout`. + /// + /// Only the timeout is parameterised, so tests can exercise the expiry + /// without waiting out the production budget. + async fn open_subscription_within( + &self, + request: &[u8], + timeout: Duration, + ) -> Result { + let (mut send, mut recv) = self .conn .open_bi() .await .map_err(|e| ClientError::Transport(Box::new(e)))?; - let req = serde_json::to_vec(&serde_json::json!({ - "method": "/events/subscribe", - "actor": &self.actor, - "params": {} - })) - .expect("serialise never fails"); - send.write_all(&req) + send.write_all(request) .await .map_err(|e| ClientError::Transport(Box::new(e)))?; send.finish() .map_err(|e| ClientError::Transport(Box::new(e)))?; - // Consume the initial {"result":{}} line. - let mut buf = BufReader::new(recv); - let mut line = String::new(); - buf.read_line(&mut line) + // i[stream.control] — the FIN is the message boundary, so the envelope + // is read to end rather than to a newline it does not carry. + let body = recv + .read_to_end(RESPONSE_LIMIT) .await .map_err(|e| ClientError::Transport(Box::new(e)))?; + Self::parse_response(&body)?; - // The daemon now opens a server-initiated uni stream carrying events. - self.accept_uni().await + // A server that answered successfully can still fail to open the data + // stream (its own failure path only logs), so the wait is bounded. + tokio::time::timeout(timeout, self.conn.accept_uni()) + .await + .map_err(|_| { + ClientError::Protocol( + "server accepted the subscription but never opened the data stream".into(), + ) + })? + .map_err(|e| ClientError::Transport(Box::new(e))) + } + + /// Subscribe to the server's event stream. + /// + /// Sends `/events/subscribe` and returns the server-initiated + /// unidirectional stream the daemon opens to push newline-delimited JSON + /// events. + pub async fn subscribe_events(&self) -> Result { + self.open_subscription("/events/subscribe", serde_json::json!({})) + .await } /// Send a QUIC datagram to the server. @@ -414,10 +488,25 @@ impl OiClient { .map_err(|e| ClientError::Transport(Box::new(e)))?; let resp_bytes = recv - .read_to_end(4 * 1024 * 1024) + .read_to_end(REQUEST_RESPONSE_LIMIT) .await .map_err(|e| ClientError::Transport(Box::new(e)))?; + Self::parse_response(&resp_bytes) + } + + /// Classify a response envelope read from a bidirectional stream. + /// + /// Shared by [`Self::request`] and [`Self::open_subscription_raw`] so that + /// every consumer of the operator interface agrees on what an error, an + /// unparseable body, and a stream that closed without a response mean. + fn parse_response(bytes: &[u8]) -> Result { + if bytes.is_empty() { + return Err(ClientError::Protocol( + "server closed the stream without a response".into(), + )); + } + #[derive(serde::Deserialize)] #[serde(untagged)] enum Response { @@ -430,7 +519,7 @@ impl OiClient { message: String, } - match serde_json::from_slice::(&resp_bytes) + match serde_json::from_slice::(bytes) .map_err(|e| ClientError::Protocol(format!("invalid response: {e}")))? { Response::Ok { result } => Ok(result), @@ -520,3 +609,219 @@ fn build_client_cert_resolver( rustls::client::AlwaysResolvesClientRawPublicKeys::new(ck), )) } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use quinn::{Endpoint, ServerConfig}; + use rustls::{ServerConfig as TlsServerConfig, server::AlwaysResolvesServerRawPublicKeys}; + + use super::*; + use crate::keys::ClientIdentity; + + /// What the stub server does once it has read a subscription request. + #[derive(Clone, Copy)] + enum StubBehaviour { + /// Answer with an error envelope and open no data stream — the + /// `server_busy` / `requirements_invalid` / `not_found` branches. + Error, + /// Finish the response stream without writing anything. + EmptyResponse, + /// Answer `{"result":{}}` and then never open the uni stream — the + /// server's own `open_uni` failure path, which only logs. + OkThenNoUni, + /// The full, correct handshake. + OkThenUni, + } + + /// Accept any client key, but negotiate the raw-public-key certificate + /// type the real daemon does — without it the handshake fails before the + /// behaviour under test is reached. + #[derive(Debug)] + struct AcceptAnyClientKey; + + impl rustls::server::danger::ClientCertVerifier for AcceptAnyClientKey { + fn root_hint_subjects(&self) -> &[rustls::DistinguishedName] { + &[] + } + + fn verify_client_cert( + &self, + _end_entity: &CertificateDer<'_>, + _intermediates: &[CertificateDer<'_>], + _now: UnixTime, + ) -> Result { + Ok(rustls::server::danger::ClientCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &DigitallySignedStruct, + ) -> Result { + ring_verify_tls12(message, cert, dss) + } + + fn verify_tls13_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &DigitallySignedStruct, + ) -> Result { + ring_verify_tls13_rpk(message, cert, dss) + } + + fn supported_verify_schemes(&self) -> Vec { + ring_schemes() + } + + fn requires_raw_public_keys(&self) -> bool { + true + } + } + + /// Stand up a QUIC endpoint speaking the OI wire protocol that answers one + /// subscription request per connection according to `behaviour`. + fn spawn_stub_server(behaviour: StubBehaviour) -> SocketAddr { + let identity = ClientIdentity::ephemeral(); + let resolver = Arc::new(AlwaysResolvesServerRawPublicKeys::new( + identity.to_certified_key().expect("certified key"), + )); + let mut tls = TlsServerConfig::builder() + .with_client_cert_verifier(Arc::new(AcceptAnyClientKey)) + .with_cert_resolver(resolver); + tls.alpn_protocols = vec![OI_ALPN.to_vec()]; + let quic = quinn::crypto::rustls::QuicServerConfig::try_from(tls).expect("quic config"); + let endpoint = Endpoint::server( + ServerConfig::with_crypto(Arc::new(quic)), + "127.0.0.1:0".parse().unwrap(), + ) + .expect("endpoint"); + let addr = endpoint.local_addr().expect("local addr"); + + tokio::spawn(async move { + while let Some(incoming) = endpoint.accept().await { + tokio::spawn(async move { + let Ok(conn) = incoming.await else { return }; + let Ok((mut send, mut recv)) = conn.accept_bi().await else { + return; + }; + let _ = recv.read_to_end(64 * 1024).await; + + match behaviour { + StubBehaviour::Error => { + let body = br#"{"error":{"code":"server_busy","message":"stream concurrency limit reached; retry after a delay"}}"#; + let _ = send.write_all(body).await; + let _ = send.finish(); + } + StubBehaviour::EmptyResponse => { + let _ = send.finish(); + } + StubBehaviour::OkThenNoUni => { + let _ = send.write_all(br#"{"result":{}}"#).await; + let _ = send.finish(); + } + StubBehaviour::OkThenUni => { + let _ = send.write_all(br#"{"result":{}}"#).await; + let _ = send.finish(); + if let Ok(mut uni) = conn.open_uni().await { + let _ = uni.write_all(b"{\"event\":\"hello\"}\n").await; + let _ = uni.finish(); + } + } + } + // Hold the connection open so the client sees the + // behaviour under test rather than a connection close. + let _ = conn.closed().await; + }); + } + }); + addr + } + + /// Short enough to keep the timeout case fast, long enough that a healthy + /// loopback handshake never trips it. + const TEST_HANDSHAKE_TIMEOUT: Duration = Duration::from_millis(500); + + async fn subscribe_against(behaviour: StubBehaviour) -> Result { + let _ = rustls::crypto::ring::default_provider().install_default(); + let addr = spawn_stub_server(behaviour); + let identity = ClientIdentity::ephemeral(); + let client = OiClient::connect_from( + addr, + "0.0.0.0:0".parse().unwrap(), + ClientAuth::TrustAny, + &identity, + Actor::default(), + ) + .await + .expect("connect to stub"); + let request = serde_json::to_vec(&serde_json::json!({ + "method": "/events/subscribe", + "actor": client.actor(), + "params": {}, + })) + .expect("serialisation"); + // Every outcome must be reached promptly: a regression that parks the + // caller fails here instead of hanging the test run. + tokio::time::timeout( + Duration::from_secs(20), + client.open_subscription_within(&request, TEST_HANDSHAKE_TIMEOUT), + ) + .await + .expect("subscribe must not block indefinitely") + } + + // i[verify stream.subscribe] + // An error response terminates the request: the client must surface the + // server's code and message rather than waiting for a stream that the + // server has already decided not to open. + #[tokio::test] + async fn error_response_surfaces_instead_of_hanging() { + match subscribe_against(StubBehaviour::Error).await { + Err(ClientError::Api { code, message }) => { + assert_eq!(code, "server_busy"); + assert!(message.contains("concurrency"), "message preserved"); + } + other => panic!("expected an Api error, got {other:?}"), + } + } + + // i[verify stream.subscribe] + // The server drops malformed requests by finishing the bidi stream with no + // response at all; that is an error, not a successful handshake. + #[tokio::test] + async fn empty_response_is_a_protocol_error() { + match subscribe_against(StubBehaviour::EmptyResponse).await { + Err(ClientError::Protocol(msg)) => { + assert!(msg.contains("without a response"), "got {msg}"); + } + other => panic!("expected a Protocol error, got {other:?}"), + } + } + + // i[verify stream.subscribe] + // The server's `open_uni` failure path only logs, so a client that waits + // unbounded on a confirmed-OK handshake still hangs forever. + #[tokio::test] + async fn missing_data_stream_times_out() { + match subscribe_against(StubBehaviour::OkThenNoUni).await { + Err(ClientError::Protocol(msg)) => { + assert!(msg.contains("never opened the data stream"), "got {msg}"); + } + other => panic!("expected a Protocol error, got {other:?}"), + } + } + + // i[verify stream.subscribe] + #[tokio::test] + async fn successful_handshake_returns_the_data_stream() { + let mut stream = subscribe_against(StubBehaviour::OkThenUni) + .await + .expect("handshake should succeed"); + let body = stream.read_to_end(4096).await.expect("read events"); + assert_eq!(body, b"{\"event\":\"hello\"}\n"); + } +} diff --git a/docs/spec/interface.md b/docs/spec/interface.md index b195f200..1916dd19 100644 --- a/docs/spec/interface.md +++ b/docs/spec/interface.md @@ -89,6 +89,12 @@ Absent specification bugs, anything that is not defined here is either defined i > If that object contains a `"method"` key it is dispatched as a control request per [stream.control](#i--stream.control). > If it contains a `"forward"` key it is dispatched as a port forward data stream per [stream.forward](#i--stream.forward). +> i[stream.subscribe] +> A subscription-style request — one whose success causes the server to open a server-initiated unidirectional stream carrying the subscribed data — is answered on the bidirectional stream exactly as any other control request, before any unidirectional stream is opened. +> A response carrying an error terminates the request: the server opens no unidirectional stream, and the connection remains usable for further requests. +> Clients must therefore read and classify the response envelope before waiting for the unidirectional stream, must treat closure of the bidirectional stream without a response envelope as an error, must surface the error envelope's `code` and `message` to their caller rather than discarding them, and must not wait indefinitely for a unidirectional stream that a server which answered successfully may still fail to open. +> The subscription-style requests are [events.subscribe](#i--event.subscribe) and [logs.stream](#i--logs.stream). + > i[stream.events] > After a client sends a `/events/subscribe` request, the server opens one server-initiated unidirectional QUIC stream per connection and pushes events as newline-delimited JSON objects for the duration of the connection. From 67a982e856a3c7027b2f354fe05dbb3ccdec011e Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 1 Aug 2026 23:57:23 +0000 Subject: [PATCH 2/3] refactor: drive every subscription through the shared handshake Deletes the three hand-rolled copies. Each had its own defect: subscribe_events and the web gateway's start_log_stream read a line that the envelope does not terminate, discarded it, and then parked on accept_uni forever when the response was an error; ctl's subscribe path read it correctly but reported a rejection as a graceful close, exiting 0. The web gateway now relays the daemon's own code and message to the browser instead of flattening every failure to daemon_unavailable, and ctl events exits 1 on a rejection. A CI guard keeps accept_uni to the helper and the shell paths, whose framing genuinely differs. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HtxQdsF6YhLzz9RrDmHEDv --- .github/workflows/rust.yml | 2 ++ crates/ctl/src/logs.rs | 39 ++--------------------------- crates/ctl/src/subscribe.rs | 49 ++++++++++++------------------------- crates/web/src/daemon.rs | 32 +++--------------------- crates/web/src/wt.rs | 15 +++++++++--- etc/ci/check-accept-uni.sh | 46 ++++++++++++++++++++++++++++++++++ 6 files changed, 81 insertions(+), 102 deletions(-) create mode 100755 etc/ci/check-accept-uni.sh diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 38cdc460..023944d3 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -37,6 +37,8 @@ jobs: run: cargo fmt --all --check - name: cargo clippy run: cargo clippy --workspace --all-targets -- -D warnings + - name: No hand-rolled subscription handshakes + run: etc/ci/check-accept-uni.sh test: name: Test diff --git a/crates/ctl/src/logs.rs b/crates/ctl/src/logs.rs index b02c6ff1..48131a46 100644 --- a/crates/ctl/src/logs.rs +++ b/crates/ctl/src/logs.rs @@ -31,45 +31,10 @@ async fn run_log_session( json_mode: bool, follow: bool, ) -> Result<(), String> { - let req_bytes = serde_json::to_vec(&serde_json::json!({ - "method": "/logs/stream", - "params": params, - })) - .expect("serialisation"); - - let (mut send, mut recv) = client - .open_bi() - .await - .map_err(|e| format!("open_bi: {e}"))?; - - send.write_all(&req_bytes) - .await - .map_err(|e| format!("write: {e}"))?; - let _ = send.finish(); - - let resp = recv - .read_to_end(64 * 1024) - .await - .map_err(|e| format!("read response: {e}"))?; - - if let Ok(v) = serde_json::from_slice::(&resp) - && let Some(err) = v.get("error") - { - let code = err - .get("code") - .and_then(|c| c.as_str()) - .unwrap_or("unknown"); - let msg = err - .get("message") - .and_then(|m| m.as_str()) - .unwrap_or("unknown error"); - return Err(format!("[{code}] {msg}")); - } - let mut log_stream = client - .accept_uni() + .open_subscription("/logs/stream", params) .await - .map_err(|e| format!("accept_uni: {e}"))?; + .map_err(|e| e.to_string())?; let mut buf = Vec::new(); let mut tmp = [0u8; 4096]; diff --git a/crates/ctl/src/subscribe.rs b/crates/ctl/src/subscribe.rs index c1e028b5..63d255fe 100644 --- a/crates/ctl/src/subscribe.rs +++ b/crates/ctl/src/subscribe.rs @@ -2,7 +2,7 @@ use std::{net::SocketAddr, time::Duration}; use seedling_protocol::{ actor::Actor, - client::{ClientAuth, OiClient}, + client::{ClientAuth, ClientError, OiClient}, keys::ClientIdentity, }; @@ -45,7 +45,13 @@ pub async fn subscribe( backoff = Duration::from_secs(1); match run_subscribe_session(&client).await { - SessionOutcome::GracefulClose => return, + // The server refused the subscription — retrying will not change + // its mind, and exiting 0 would tell a script the feed was + // consumed to its end. + SessionOutcome::Rejected(e) => { + eprintln!("subscription rejected: {e}"); + std::process::exit(1); + } SessionOutcome::Error(e) => { // Reset the deadline when we start reconnecting, not when the // session began — otherwise a long-lived session causes the @@ -61,45 +67,20 @@ pub async fn subscribe( } enum SessionOutcome { - GracefulClose, + /// The server answered the subscription request with an error. + Rejected(String), Error(String), Interrupted, } // i[impl ctl.graceful-shutdown] async fn run_subscribe_session(client: &OiClient) -> SessionOutcome { - let req_bytes = serde_json::to_vec(&serde_json::json!({ - "method": "/events/subscribe", - "actor": client.actor(), - "params": {}, - })) - .expect("serialisation"); - - let (mut send, mut recv) = match client.open_bi().await { - Ok(s) => s, - Err(e) => return SessionOutcome::Error(format!("open_bi: {e}")), - }; - - if let Err(e) = send.write_all(&req_bytes).await { - return SessionOutcome::Error(format!("write: {e}")); - } - let _ = send.finish(); - - let resp = match recv.read_to_end(64 * 1024).await { - Ok(r) => r, - Err(e) => return SessionOutcome::Error(format!("read response: {e}")), - }; - - if let Ok(v) = serde_json::from_slice::(&resp) - && v.get("error").is_some() - { - eprintln!("{}", serde_json::to_string_pretty(&v).unwrap_or_default()); - return SessionOutcome::GracefulClose; - } - - let mut event_stream = match client.accept_uni().await { + let mut event_stream = match client.subscribe_events().await { Ok(s) => s, - Err(e) => return SessionOutcome::Error(format!("accept_uni: {e}")), + Err(ClientError::Api { code, message }) => { + return SessionOutcome::Rejected(format!("[{code}] {message}")); + } + Err(e) => return SessionOutcome::Error(e.to_string()), }; let mut buf = Vec::new(); diff --git a/crates/web/src/daemon.rs b/crates/web/src/daemon.rs index 55db82af..c73493d9 100644 --- a/crates/web/src/daemon.rs +++ b/crates/web/src/daemon.rs @@ -7,7 +7,6 @@ use seedling_protocol::actor::Actor; use seedling_protocol::client::{ClientAuth, ClientError, OiClient}; use seedling_protocol::keys::ClientIdentity; use serde_json::json; -use tokio::io::{AsyncBufReadExt as _, BufReader}; use tokio::sync::{Mutex, oneshot}; /// Routes incoming daemon uni streams to registered handlers by QUIC stream ID. @@ -235,33 +234,10 @@ impl DaemonConn { request_bytes: &[u8], ) -> Result<(OiClient, quinn::RecvStream), ClientError> { let client = self.new_events_client().await?; - let conn = client.connection().clone(); - - let (mut send, recv) = conn - .open_bi() - .await - .map_err(|e| ClientError::Transport(Box::new(e)))?; - - send.write_all(request_bytes) - .await - .map_err(|e| ClientError::Transport(Box::new(e)))?; - send.write_all(b"\n") - .await - .map_err(|e| ClientError::Transport(Box::new(e)))?; - send.finish() - .map_err(|e| ClientError::Transport(Box::new(e)))?; - - let mut buf = BufReader::new(recv); - let mut line = String::new(); - buf.read_line(&mut line) - .await - .map_err(|e| ClientError::Transport(Box::new(e)))?; - - let log_recv = conn - .accept_uni() - .await - .map_err(|e| ClientError::Transport(Box::new(e)))?; - + // The raw variant, not `open_subscription`: the request the gateway + // relays already carries the browser session's actor, which must not + // be replaced by the web service's own. + let log_recv = client.open_subscription_raw(request_bytes).await?; Ok((client, log_recv)) } diff --git a/crates/web/src/wt.rs b/crates/web/src/wt.rs index 5e725b40..e0a058bc 100644 --- a/crates/web/src/wt.rs +++ b/crates/web/src/wt.rs @@ -1,6 +1,7 @@ use std::net::SocketAddr; use std::sync::Arc; +use seedling_protocol::client::ClientError; use serde_json::json; use tokio::io::AsyncWriteExt as _; use tokio::sync::watch; @@ -250,9 +251,17 @@ async fn handle_incoming(incoming: wtransport::endpoint::IncomingSession, state: } Err(e) => { tracing::error!("log stream setup failed: {e}"); - let msg = serde_json::json!({ - "error": { "code": "daemon_unavailable", "message": e.to_string() } - }); + // A rejection by the daemon is the daemon's answer, not + // a gateway failure: relay its code and message so the + // browser can tell "no such app" from "daemon down". + let msg = match &e { + ClientError::Api { code, message } => serde_json::json!({ + "error": { "code": code, "message": message } + }), + _ => serde_json::json!({ + "error": { "code": "daemon_unavailable", "message": e.to_string() } + }), + }; let _ = wt_send.write_all((msg.to_string() + "\n").as_bytes()).await; let _ = wt_send.shutdown().await; } diff --git a/etc/ci/check-accept-uni.sh b/etc/ci/check-accept-uni.sh new file mode 100755 index 00000000..0ebfee3c --- /dev/null +++ b/etc/ci/check-accept-uni.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +# Fail if a subscription-style request drives the handshake by hand. +# +# Reading the response envelope, classifying it, and only then awaiting the +# server-initiated data stream is one contract (i[stream.subscribe]) that four +# call sites used to re-implement, each getting a different part wrong: two +# blocked forever on an error response, one exited 0 on a rejection. It lives +# in OiClient::open_subscription now, and a fifth copy should not appear. +# +# The allowlist is for stream kinds whose framing genuinely differs: the shell +# protocol announces its uni stream IDs in the handshake, and the web gateway +# demuxes those streams for the browser. + +set -euo pipefail + +allowed=( + 'crates/protocol/src/client.rs' # the shared helper itself + 'crates/ctl/src/shell.rs' # i[stream.shell] — different framing + 'crates/web/src/daemon.rs' # the shell uni-stream dispatcher +) + +pattern='accept_uni' +mapfile -t hits < <(grep -rln --include='*.rs' "$pattern" crates/ | sort) + +violations=() +for hit in "${hits[@]}"; do + skip=false + for ok in "${allowed[@]}"; do + [[ "$hit" == "$ok" ]] && skip=true && break + done + $skip || violations+=("$hit") +done + +if ((${#violations[@]})); then + echo "error: accept_uni outside the allowlist:" >&2 + printf ' %s\n' "${violations[@]}" >&2 + cat >&2 <<'EOF' + +Subscription-style requests must go through OiClient::open_subscription, which +reads and classifies the response envelope before awaiting the data stream. See +docs/logic-bug-audit-2026-07/theme-1-stream-error-handling.md. If this stream +kind really does frame differently (as the shell protocol does), add it to the +allowlist in this script with a comment saying why. +EOF + exit 1 +fi From ab724d785cdb854a9d4f78ad5933422e5e368027 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 03:10:32 +0000 Subject: [PATCH 3/3] docs(spec): say that a subscription response is FIN-delimited MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The section carries two framings — newline-terminated dispatch headers and stream-boundary messages — so which one bounds the response envelope was left to inference by anyone implementing a client from the spec alone. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HtxQdsF6YhLzz9RrDmHEDv --- docs/spec/interface.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/spec/interface.md b/docs/spec/interface.md index c5f7423e..6515044d 100644 --- a/docs/spec/interface.md +++ b/docs/spec/interface.md @@ -104,6 +104,7 @@ Absent specification bugs, anything that is not defined here is either defined i > i[stream.subscribe] > A subscription-style request — one whose success causes the server to open a server-initiated unidirectional stream carrying the subscribed data — is answered on the bidirectional stream exactly as any other control request, before any unidirectional stream is opened. > A response carrying an error terminates the request: the server opens no unidirectional stream, and the connection remains usable for further requests. +> The response is bounded by the close of the bidirectional stream, as for any other control request per [stream.control](#i--stream.control), and not by a newline: a client must read the bidirectional stream to its end before parsing, and the subsequent data arrives on the unidirectional stream rather than on this one. > Clients must therefore read and classify the response envelope before waiting for the unidirectional stream, must treat closure of the bidirectional stream without a response envelope as an error, must surface the error envelope's `code` and `message` to their caller rather than discarding them, and must not wait indefinitely for a unidirectional stream that a server which answered successfully may still fail to open. > The subscription-style requests are [events.subscribe](#i--event.subscribe) and [logs.stream](#i--logs.stream).