test(infra): cover review-002 stream/PEM/cap error arms + drop dead WsTimeouts Default

- forward_stream build-error arm (forward.rs): wire-level test asserts
  one INVALID_INPUT envelope then stream end with zero upstream
  contact (a panicking responder counts as the contact guard), plus
  the from_jsonschema integration mirror (undeclared key + non-scalar
  placeholder, each naming its rejection source)
- PEM read-failure arms (http_client.rs): nonexistent CA path →
  CaBundleRead (sync new), nonexistent client-cert path → ClientCertRead
  (async reload, prior generation retained)
- Over-cap poll_write rejection leg (byte_adapter.rs): cap+1 write →
  InvalidData naming the cap; stream stays usable for an at-cap write
  afterwards
- SSE parser edges: CRLF split across feed chunks frames one line;
  invalid-UTF8 data lines drop without killing the frame stream
- from_value structural rejects: non-object doc, missing `info`,
  missing `paths`, non-object `paths` each name the member
- Connection-failure arms: accept-path ConnectionClosed →
  HandlerError::ConnectionClosed via stream_error_to_handler; read-pump
  demux-gone break ends the pump when the byte-stream side is dropped
- Delete the caller-less `impl Default for WsTimeouts` (the extension
  is constructed explicitly)

cargo llvm-cov --all-features: all named arms covered; TOTAL regions
94.18% (was 93.86%), lines 96.04% (was 95.77%); http_client.rs
86.56% lines (was 81.72%).

docs(tasks): mark review-002-fu-stream-error-coverage done

src/adapters/forward.rs

@@ -2361,6 +2361,54 @@ mod tests {
         assert_eq!(envelopes[3].result.clone().unwrap(), json!({"n": 1}));
     }
 
+    /// SSE framing edge: a CRLF split across TCP chunks (chunk 1 ends
+    /// with the `\r`, chunk 2 opens with the `\n`) still frames as one
+    /// line — the reassembly window waits for the `\n` before parsing,
+    /// not treating the `\r` as a terminator (so the bare `\r` holds
+    /// until the `\n` arrives, then frames a single line).
+    #[test]
+    fn sse_parser_crlf_split_across_chunks_frames_one_line() {
+        let mut parser = SseParser::new();
+        let first = parser.feed(b"data: {\"n\":1}\r", false).expect("chunk 1");
+        assert!(first.is_empty(), "the bare \\r does not dispatch");
+        let second = parser
+            .feed(b"\ndata: {\"n\":2}\r\n\r\n", false)
+            .expect("chunk 2");
+        assert_eq!(second.len(), 1, "the joined CRLF framed exactly one event");
+        assert_eq!(
+            second[0].data, "{\"n\":1}\n{\"n\":2}",
+            "the \\r held as line content until the arriving \\n framed it, \
+             joining both data lines per WHATWG accumulation"
+        );
+        let rest = parser.feed(b"data: {\"n\":2}\n\n", true).expect("tail");
+        assert_eq!(rest.len(), 1);
+        assert_eq!(rest[0].data, "{\"n\":2}", "the second frame is unaffected");
+    }
+
+    /// A line that is not valid UTF-8 is dropped (WHATWG decode-failure
+    /// semantics for this parser): no event, no panic, and the framing
+    /// continues — the following valid frame dispatches normally.
+    #[test]
+    fn sse_parser_drops_invalid_utf8_lines_and_keeps_framing() {
+        let mut parser = SseParser::new();
+        let events = parser
+            .feed(b"data: \xff\xfe\xfd\n\ndata: {\"ok\":true}\n\n", false)
+            .expect("ascii framing is valid");
+        assert_eq!(events.len(), 1, "the invalid-UTF8 data line was dropped");
+        assert_eq!(events[0].data, "{\"ok\":true}");
+        assert_eq!(events[0].event, None);
+
+        let mut parser = SseParser::new();
+        let events = parser
+            .feed(b"data: good\r\ndata: \xf0\x28\x8c\x28\r\n\r\n", false)
+            .expect("CRLF framing is valid");
+        assert_eq!(events.len(), 1);
+        assert_eq!(
+            events[0].data, "good",
+            "the invalid line contributes nothing to the joined data"
+        );
+    }
+
     /// FWD-17: an upstream `event:` field on a non-JSON payload is
     /// carried in the wrapper; JSON payloads surface as themselves even
     /// under a named event, and the name is reset after dispatch (the
@@ -2983,6 +3031,50 @@ mod tests {
         }
     }
 
+    /// The streaming analog of the Once-path build-error envelope: a
+    /// `forward_stream` invocation with an invalid input yields exactly
+    /// one `INVALID_INPUT` envelope and the stream ends, before any
+    /// upstream contact — the responder counts requests and the test
+    /// asserts the count stays zero (a regression returning an empty
+    /// stream, silently swallowing the error, would be the worst
+    /// failure shape for a subscriptions consumer).
+    #[tokio::test]
+    async fn stream_build_error_yields_one_invalid_input_envelope_with_zero_upstream_contact() {
+        let base = spawn_responder(TestArc::new(|_parts| {
+            panic!("the rejected invocation must never reach the upstream");
+        }))
+        .await;
+        for input in [
+            json!({"debug": true}),
+            json!({"id": {"deeply": {"nested": "object"}}}),
+        ] {
+            let stream = forward_stream(
+                &minimal_client(),
+                &base,
+                "/x/{id}",
+                "GET",
+                &None,
+                &TestHashMap::new(),
+                "svc",
+                &serde_json::json!({
+                    "type": "object",
+                    "properties": {"id": {"type": "string"}}
+                }),
+                &[],
+                input,
+                noop_context(),
+            );
+            let envelopes = collect_stream(stream).await;
+            assert_eq!(envelopes.len(), 1, "one envelope, then the stream ends");
+            match &envelopes[0].result {
+                Err(err) => {
+                    assert_eq!(err.code, "INVALID_INPUT", "message was: {}", err.message);
+                }
+                other => panic!("expected one INVALID_INPUT envelope, got {other:?}"),
+            }
+        }
+    }
+
     /// COV-10 (b): the responder emits one complete `data:` frame, lets
     /// the client consume it, then kills the connection before the
     /// declared body length is met — a genuine mid-stream transport

src/adapters/from_jsonschema.rs

@@ -591,6 +591,63 @@ mod tests {
         assert_eq!(collected[1].request_id, "req-13");
     }
 
+    /// The streaming analog of the Once-path build-error envelope: a
+    /// Sub op invoked with an invalid input (undeclared key, or a
+    /// non-scalar value bound to a path placeholder) yields exactly one
+    /// `INVALID_INPUT` envelope, then the stream ends — pre-send, so
+    /// the upstream receives zero requests. An empty stream (a silently
+    /// swallowed error) would be the worst failure shape for a
+    /// subscriptions consumer; the non-empty, single-error shape is the
+    /// assertion.
+    #[tokio::test]
+    async fn integration_sse_subscription_invalid_input_yields_one_error_envelope_and_ends() {
+        for (input, expected_fragment) in [
+            (
+                serde_json::json!({"debug": true}),
+                "not declared by the operation's input schema",
+            ),
+            (
+                serde_json::json!({"id": {"nested": "object"}}),
+                "must be a scalar",
+            ),
+        ] {
+            let base = spawn_echo_server(200, "data: {\"n\":1}\n\n", "text/event-stream").await;
+            let adapter = FromJsonSchema::new(
+                test_spec("svc/stream", OperationType::Sub),
+                test_config("svc", &base),
+                "/stream/{id}".to_string(),
+                "POST".to_string(),
+                test_http_client(),
+            )
+            .unwrap();
+            let bundles = adapter.import().await.unwrap();
+            let registration = &bundles[0];
+            let ctx = noop_context("req-13b", Capabilities::new());
+            let stream = match &registration.handler {
+                HandlerKind::Stream(h) => h(input.clone(), ctx),
+                _ => panic!("expected Stream handler"),
+            };
+            let collected: Vec<ResponseEnvelope> = stream.collect().await;
+            assert_eq!(
+                collected.len(),
+                1,
+                "exactly one error envelope, then the stream ends (never an empty-200-subscribe)"
+            );
+            match &collected[0].result {
+                Err(e) => {
+                    assert_eq!(e.code, "INVALID_INPUT", "message was: {}", e.message);
+                    assert!(
+                        e.message.contains(expected_fragment),
+                        "message `{}` lacks `{expected_fragment}`",
+                        e.message
+                    );
+                }
+                other => panic!("expected INVALID_INPUT, got {other:?}"),
+            }
+            assert_eq!(collected[0].request_id, "req-13b");
+        }
+    }
+
     #[test]
     fn no_env_vars_read_in_build_request() {
         std::env::set_var("OPENAI_API_KEY", "should-not-be-used");

src/adapters/openapi_spec.rs

@@ -1204,6 +1204,41 @@ mod tests {
         assert!(props.get("name").is_some());
     }
 
+    /// The structural-reject arms at the very top of `from_value`
+    /// (before any path walking): a non-object document, a
+    /// missing-`info` object, and a non-object `paths` each fail with
+    /// `SchemaParse` naming the missing/mistyped member.
+    #[test]
+    fn from_value_structural_rejects_name_the_missing_member() {
+        for (raw, expected_fragment) in [
+            (json!("just a string"), "must be a JSON object"),
+            (json!([1, 2, 3]), "must be a JSON object"),
+            (json!({"paths": {}}), "missing `info`"),
+            (
+                json!({"info": {"title": "T", "version": "1"}}),
+                "missing `paths`",
+            ),
+            (
+                json!({
+                    "info": {"title": "T", "version": "1"},
+                    "paths": ["/x"]
+                }),
+                "`paths` must be a JSON object",
+            ),
+        ] {
+            match OpenAPISpec::from_value(raw) {
+                Err(AdapterError::SchemaParse { message }) => {
+                    assert!(
+                        message.contains(expected_fragment),
+                        "message `{message}` lacks `{expected_fragment}`"
+                    );
+                }
+                Ok(_) => panic!("the malformed document must be rejected"),
+                other => panic!("expected SchemaParse, got {other:?}"),
+            }
+        }
+    }
+
     #[test]
     fn request_body_self_ref_fails_import_not_silent_bodyless_op() {
         let doc = r##"{

src/server/adapter.rs

@@ -765,6 +765,29 @@ mod tests {
         let _ = server_task.await;
     }
 
+    /// The accept-path connection-failure arm: a `Connection` whose
+    /// single underlying stream is already gone (closed by the dial
+    /// side) yields `StreamError::ConnectionClosed` from `accept_bi`,
+    /// which `stream_error_to_handler` maps to
+    /// `HandlerError::ConnectionClosed` — the error the consumer's
+    /// accept loop observes for a peer that died before the first
+    /// stream arrived.
+    #[tokio::test]
+    async fn accept_bi_failure_maps_to_handler_error_connection_closed() {
+        let adapter = HttpAdapter::new(provider(), empty_registry());
+        let (client, server) = tokio::io::duplex(64 * 1024);
+        let conn = Connection::from_bidi(server, b"http/1.1".to_vec(), None);
+        let auth = AuthContext::anonymous(b"http/1.1");
+
+        drop(client);
+        conn.close(0, "gone before the first stream");
+        let result = ProtocolHandler::handle(&adapter, conn, &auth).await;
+        match result {
+            Err(HandlerError::ConnectionClosed) => {}
+            other => panic!("expected HandlerError::ConnectionClosed, got {other:?}"),
+        }
+    }
+
     #[tokio::test]
     async fn healthz_served_by_the_adapter_over_duplex() {
         let adapter = HttpAdapter::new(provider(), empty_registry());

src/websocket/byte_adapter.rs

@@ -1188,6 +1188,54 @@ mod tests {
         .expect("pump emitted the cap-sized chunk");
     }
 
+    /// WS-14's rejection leg: a single `AsyncWrite` call one byte *over*
+    /// `PENDING_BUFFER_CAP` fails synchronously with `InvalidData`,
+    /// naming the cap, before any byte enters the write queue — the mux
+    /// sees the stream error directly instead of the fault reaching the
+    /// wire (the = cap acceptance is the test above; this is the
+    /// over-cap rejection the WS-14 move introduced).
+    #[tokio::test]
+    async fn tungstenite_write_one_byte_over_the_cap_is_rejected_with_invalid_data() {
+        let (client_io, mut server_io) = tokio::io::duplex(64);
+        let ws = tokio_tungstenite::WebSocketStream::from_raw_socket(
+            client_io,
+            tokio_tungstenite::tungstenite::protocol::Role::Client,
+            None,
+        )
+        .await;
+        let (mut stream, _pumps) = split_tungstenite_to_bytes(ws);
+
+        let err = stream
+            .write_all(&vec![0u8; PENDING_BUFFER_CAP + 1])
+            .await
+            .expect_err("the over-cap write must be rejected");
+        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+        assert!(
+            err.to_string().contains("pending buffer cap"),
+            "error names the cap: {err}"
+        );
+        assert!(
+            err.to_string().contains(&PENDING_BUFFER_CAP.to_string()),
+            "error carries the cap value: {err}"
+        );
+
+        // The rejection must not poison the stream: an exactly-at-cap
+        // write still flows through the pump afterwards.
+        let mut chunk = vec![0u8; PENDING_BUFFER_CAP];
+        chunk[4..8].copy_from_slice(&(PENDING_BUFFER_CAP as u32 - 8).to_be_bytes());
+        stream
+            .write_all(&chunk)
+            .await
+            .expect("at-cap write after the rejection is accepted");
+        stream.flush().await.expect("flush");
+        tokio::time::timeout(std::time::Duration::from_secs(30), async {
+            let mut buf = [0u8; 2];
+            server_io.read_exact(&mut buf).await.expect("chunk read")
+        })
+        .await
+        .expect("pump still emits after a rejected over-cap write");
+    }
+
     /// A valid chunk (length field ≤ `MAX_CHUNK_LEN`) still parses and
     /// flushes through the pump after the caps landed — the validation
     /// must not false-positive on well-framed traffic. (The pump-side
@@ -1268,6 +1316,54 @@ mod tests {
         assert_eq!(&buf, b"from-peer");
     }
 
+    /// The read-pump's demux-gone break: with the byte-stream side
+    /// (the `read_rx` holder) dropped while the WS is still open, the
+    /// next inbound binary message fails its `read_tx.send` and ends
+    /// the pump — it must not park on the now-unreceivable channel.
+    /// The peer holds the socket open (no EOF, idle window far beyond
+    /// the test budget), so the only exit for the pump is the failed
+    /// send itself; a regression that parks here hangs until the test
+    /// deadline rather than ending the task.
+    #[tokio::test]
+    async fn read_pump_breaks_when_the_byte_stream_side_is_dropped() {
+        let (client_io, server_io) = tokio::io::duplex(1 << 16);
+        let ws = tokio_tungstenite::WebSocketStream::from_raw_socket(
+            client_io,
+            tokio_tungstenite::tungstenite::protocol::Role::Client,
+            None,
+        )
+        .await;
+        let (_stream, pumps) = split_tungstenite_to_bytes(ws);
+        drop(_stream);
+
+        let peer = tokio::spawn(async move {
+            let peer = tokio_tungstenite::WebSocketStream::from_raw_socket(
+                server_io,
+                tokio_tungstenite::tungstenite::protocol::Role::Server,
+                None,
+            )
+            .await;
+            let (mut sink, mut reader) = peer.split();
+            use futures::{SinkExt, StreamExt};
+            sink.send(tokio_tungstenite::tungstenite::Message::Binary(
+                b"after-drop".to_vec().into(),
+            ))
+            .await
+            .expect("peer send");
+            let _ = tokio::time::timeout(std::time::Duration::from_secs(10), reader.next()).await;
+        });
+
+        let ended = tokio::time::timeout(std::time::Duration::from_secs(5), pumps.read_task)
+            .await
+            .expect("the pump must end on the demux-gone break, not park");
+        assert!(
+            ended.is_ok(),
+            "the pump ends by itself (the break), not by abort: {ended:?}"
+        );
+
+        peer.abort();
+    }
+
     /// The lossless read-EOF signal on the tungstenite path (COV-03,
     /// the WS-11 constraint): the peer disappearing surfaces on
     /// `pumps.read_eof()` as `watch` = true — observable both before

src/websocket/upgrade.rs

@@ -341,15 +341,6 @@ pub struct WsTimeouts {
     pub write: Option<std::time::Duration>,
 }
 
-impl Default for WsTimeouts {
-    fn default() -> Self {
-        Self {
-            idle: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT),
-            write: None,
-        }
-    }
-}
-
 /// The upgrade handler. Requires the resolved identity in request
 /// extensions (stashed by [`ws_bearer_auth`]) — a WS session without
 /// an identity cannot run `AccessControl::check`.

tasks/infra/review-002-fu-stream-error-coverage.md

@@ -1,7 +1,7 @@
 ---
 id: review-002-fu-stream-error-coverage
 name: Streaming error-arm coverage — forward_stream invalid input, PEM read-failure, over-cap poll_write mirror (post-bulk coverage gap)
-status: pending
+status: done
 depends_on: []
 scope: narrow
 risk: low

tests/client_tls.rs

@@ -374,6 +374,73 @@ async fn reload_to_a_ca_bundle_backed_client_succeeds() {
     tokio::time::sleep(Duration::from_millis(1)).await;
 }
 
+/// COV-11/CLI-03 read-failure arms (the sync `SharedHttpClient::new`
+/// path): a `ca_bundle` path that does not exist fails the build with
+/// `CaBundleRead` carrying the path — the unreadable-file complement to
+/// the parse-failure tests above.
+#[test]
+fn nonexistent_ca_bundle_path_fails_ca_bundle_read_with_path() {
+    let path = std::env::temp_dir().join(format!(
+        "alkhttp-pem-read-{}-{}-missing.pem",
+        std::process::id(),
+        uuid::Uuid::new_v4()
+    ));
+    let error = SharedHttpClient::new(client_config(Some(path.clone()), None))
+        .expect_err("a nonexistent CA path must fail the build");
+    match error {
+        HttpClientBuildError::CaBundleRead { path: p, .. } => {
+            assert_eq!(p, path, "the error names the unreadable path");
+        }
+        other => panic!("expected CaBundleRead, got {other:?}"),
+    }
+}
+
+/// COV-11/CLI-03 read-failure arm (the async `reload` path): a
+/// `client_cert` key path that does not exist fails the reload with
+/// `ClientCertRead` carrying the unreadable path, and the held clients
+/// stay on the previous generation.
+#[tokio::test]
+async fn reload_with_nonexistent_client_cert_path_fails_client_cert_read() {
+    let pki = TestPki::generate();
+    let (_ca, _cert, dir) = pki.write_config_files(false);
+
+    let http = SharedHttpClient::new(HttpClientConfig::default()).expect("initial client");
+    let missing_key = std::env::temp_dir().join(format!(
+        "alkhttp-pem-read-{}-{}-missing-key.pem",
+        std::process::id(),
+        uuid::Uuid::new_v4()
+    ));
+    let missing_cert = std::env::temp_dir().join(format!(
+        "alkhttp-pem-read-{}-{}-missing-cert.pem",
+        std::process::id(),
+        uuid::Uuid::new_v4()
+    ));
+    let error = http
+        .reload(client_config(
+            None,
+            Some(ClientCertConfig {
+                cert_pem: missing_cert.clone(),
+                key_pem: missing_key.clone(),
+            }),
+        ))
+        .await
+        .expect_err("a nonexistent client-cert key path must fail the reload");
+    match error {
+        HttpClientBuildError::ClientCertRead { path: p, .. } => {
+            assert_eq!(
+                p, missing_cert,
+                "the error names the unreadable cert path (read before the key)"
+            );
+        }
+        other => panic!("expected ClientCertRead, got {other:?}"),
+    }
+    assert!(
+        http.config().client_cert.is_none(),
+        "the reload failure keeps the previous generation's clients+config (FWD-12)"
+    );
+    cleanup_dir(&dir);
+}
+
 /// COV-11/CLI-03 parse-failure arms: a `ca_bundle` file that parses as
 /// PEM framing but carries a corrupt section fails the build with
 /// `CaBundleParse`, carrying the offending path. (Purely non-PEM text