Merge branch 'wt/review-002-client-policy-wire-tests'

src/adapters/forward.rs

@@ -2892,4 +2892,516 @@ mod tests {
             "one byte past a full buffer still trips before extend"
         );
     }
+
+    /// FWD-07 Once-path decode arm: a `200` with `application/json` and
+    /// a body that does not parse surfaces a single `INTERNAL` decode
+    /// error envelope — never a partial or fabricated success.
+    #[tokio::test]
+    async fn malformed_json_body_200_decodes_to_an_internal_error_envelope() {
+        let base = spawn_responder(TestArc::new(|_parts| {
+            http_response(200, "application/json", b"{not json".to_vec())
+        }))
+        .await;
+        let envelope = call_forward(&base, noop_context()).await;
+        match envelope.result {
+            Err(err) => {
+                assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
+                assert!(
+                    err.message.contains("failed to decode response body"),
+                    "message was: {}",
+                    err.message
+                );
+            }
+            other => panic!("expected INTERNAL decode envelope, got {other:?}"),
+        }
+    }
+
+    /// FWD-07 Once-path binary arm: a `200` with
+    /// `application/octet-stream` surfaces the body as a JSON byte
+    /// array (bounded by the response cap like every other read path).
+    #[tokio::test]
+    async fn binary_octet_stream_200_surfaces_as_a_byte_array_envelope() {
+        let base = spawn_responder(TestArc::new(|_parts| {
+            http_response(
+                200,
+                "application/octet-stream",
+                vec![0x00, 0xFF, 0x10, 0x42],
+            )
+        }))
+        .await;
+        let envelope = call_forward(&base, noop_context()).await;
+        match envelope.result {
+            Ok(value) => assert_eq!(value, json!([0, 255, 16, 66])),
+            other => panic!("expected a byte-array envelope, got {other:?}"),
+        }
+    }
+
+    /// COV-10 (a): an upstream SSE line longer than the 1 MiB line cap
+    /// with no newline terminates `forward_stream` with a single
+    /// `INTERNAL` terminal envelope, then the stream ends.
+    #[tokio::test]
+    async fn oversized_upstream_sse_line_terminates_with_one_internal_envelope() {
+        let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
+        let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
+            use tokio::io::AsyncWriteExt;
+            let line = vec![b'a'; SSE_EVENT_BUFFER_CAP + 1];
+            let _ = sock.write_all(b"data: ").await;
+            let _ = sock.write_all(&line).await;
+            let _ = sock.write_all(b"\n\n").await;
+            let _ = sock.flush().await;
+        })
+        .await;
+        let stream = forward_stream(
+            &minimal_client(),
+            &base,
+            "/x",
+            "GET",
+            &None,
+            &TestHashMap::new(),
+            "svc",
+            &serde_json::json!({"type": "object"}),
+            &[],
+            json!({}),
+            noop_context(),
+        );
+        let envelopes = collect_stream(stream).await;
+        assert_eq!(
+            envelopes.len(),
+            1,
+            "the parse-overflow terminal envelope is the only output"
+        );
+        match &envelopes[0].result {
+            Err(err) => {
+                assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
+                assert!(
+                    err.message.contains("SSE parse error"),
+                    "message was: {}",
+                    err.message
+                );
+            }
+            other => panic!("expected one INTERNAL terminal 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
+    /// abort, not a graceful EOF. The delivered event passes, a single
+    /// `INTERNAL` terminal error envelope follows, the stream ends.
+    #[tokio::test]
+    async fn aborted_socket_mid_stream_emits_a_terminal_error_envelope() {
+        let head =
+            "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ncontent-length: 512\r\n\r\n";
+        let gate = TestArc::new(tokio::sync::Notify::new());
+        let notify = TestArc::clone(&gate);
+        let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
+            use tokio::io::AsyncWriteExt;
+            let _ = sock.write_all(b"data: {\"n\":1}\n\n").await;
+            let _ = sock.flush().await;
+            notify.notified().await;
+            drop(sock);
+        })
+        .await;
+        let stream = forward_stream(
+            &minimal_client(),
+            &base,
+            "/x",
+            "GET",
+            &None,
+            &TestHashMap::new(),
+            "svc",
+            &serde_json::json!({"type": "object"}),
+            &[],
+            json!({}),
+            noop_context(),
+        );
+        tokio::pin!(stream);
+        let first = tokio::time::timeout(Duration::from_secs(5), stream.next())
+            .await
+            .expect("the complete frame must deliver within the test budget")
+            .expect("stream must deliver the frame before the abort");
+        assert_eq!(first.result.clone().unwrap(), json!({"n": 1}));
+        gate.notify_one();
+        let terminal = tokio::time::timeout(Duration::from_secs(5), stream.next())
+            .await
+            .expect("the terminal envelope must follow within the test budget")
+            .expect("the stream must end with the terminal envelope");
+        match terminal.result {
+            Err(err) => {
+                assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
+                assert!(
+                    err.message.contains("SSE stream error"),
+                    "message was: {}",
+                    err.message
+                );
+            }
+            other => panic!("expected the terminal error envelope, got {other:?}"),
+        }
+        let ended = tokio::time::timeout(Duration::from_secs(1), stream.next()).await;
+        assert!(
+            matches!(ended, Ok(None) | Err(_)),
+            "the stream ends after the terminal envelope"
+        );
+    }
+
+    /// COV-10 (c): the responder ends the stream with a pending event
+    /// (data lines received, no trailing blank line) — the WHATWG
+    /// EOF-flush dispatches that pending event as a final success
+    /// envelope at EOF.
+    #[tokio::test]
+    async fn pending_event_flushes_at_eof_without_a_trailing_blank_line() {
+        let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\r\n";
+        let base = spawn_sse_responder_with_writer(head, move |mut sock| async move {
+            use tokio::io::AsyncWriteExt;
+            let _ = sock
+                .write_all(b"data: {\"n\":1}\n\ndata: {\"final\":true}\n")
+                .await;
+            let _ = sock.flush().await;
+            let _ = sock.shutdown().await;
+        })
+        .await;
+        let stream = forward_stream(
+            &minimal_client(),
+            &base,
+            "/x",
+            "GET",
+            &None,
+            &TestHashMap::new(),
+            "svc",
+            &serde_json::json!({"type": "object"}),
+            &[],
+            json!({}),
+            noop_context(),
+        );
+        let envelopes = collect_stream(stream).await;
+        assert_eq!(envelopes.len(), 2);
+        assert_eq!(envelopes[0].result.clone().unwrap(), json!({"n": 1}));
+        assert_eq!(
+            envelopes[1].result.clone().unwrap(),
+            json!({"final": true}),
+            "EOF-flush dispatches the pending event"
+        );
+    }
+
+    /// COV-10 (d), Once-path transport arm: a connect-refused upstream
+    /// surfaces a single `INTERNAL` error envelope through `forward`.
+    #[tokio::test]
+    async fn dead_port_forward_yields_an_internal_envelope() {
+        let envelope = forward(
+            &minimal_client(),
+            "http://127.0.0.1:9",
+            "/x",
+            "GET",
+            &None,
+            &TestHashMap::new(),
+            "svc",
+            &serde_json::json!({"type": "object"}),
+            &[],
+            json!({}),
+            noop_context(),
+        )
+        .await;
+        match envelope.result {
+            Err(err) => {
+                assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
+                assert!(
+                    err.message.contains("HTTP request failed"),
+                    "message was: {}",
+                    err.message
+                );
+            }
+            other => panic!("expected INTERNAL transport envelope, got {other:?}"),
+        }
+    }
+
+    /// COV-10 (d), streaming-path transport arm: the same dead upstream
+    /// through `forward_stream` surfaces a single terminal `INTERNAL`
+    /// error envelope (the streaming build/send-failure terminal arm).
+    #[tokio::test]
+    async fn dead_port_forward_stream_yields_a_terminal_internal_envelope() {
+        let stream = forward_stream(
+            &minimal_client(),
+            "http://127.0.0.1:9",
+            "/x",
+            "GET",
+            &None,
+            &TestHashMap::new(),
+            "svc",
+            &serde_json::json!({"type": "object"}),
+            &[],
+            json!({}),
+            noop_context(),
+        );
+        let envelopes = collect_stream(stream).await;
+        assert_eq!(
+            envelopes.len(),
+            1,
+            "exactly one terminal envelope, then the stream ends"
+        );
+        match &envelopes[0].result {
+            Err(err) => {
+                assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
+                assert!(
+                    err.message.contains("HTTP request failed"),
+                    "message was: {}",
+                    err.message
+                );
+            }
+            other => panic!("expected terminal INTERNAL envelope, got {other:?}"),
+        }
+    }
+
+    /// COV-12, loud-credential arm (ApiKey): an ApiKey credential value
+    /// carrying invalid HTTP header bytes fails loudly and never echoes
+    /// the credential material (FWD-08 family).
+    #[tokio::test]
+    async fn api_key_with_invalid_value_fails_loudly_without_echoing_secrets() {
+        let base = spawn_responder(TestArc::new(|_parts| {
+            http_response(200, "application/json", b"{}".to_vec())
+        }))
+        .await;
+        let ctx = ctx_with_capability("svc", "key\u{0003}-secret-marker".to_string());
+        let envelope = call_forward_authed(
+            &base,
+            ctx,
+            &Some(HttpAuthScheme::ApiKey {
+                header_name: "x-api-key".to_string(),
+            }),
+        )
+        .await;
+        match envelope.result {
+            Err(err) => {
+                assert!(
+                    err.message
+                        .contains("refusing to send the request unauthenticated"),
+                    "message was: {}",
+                    err.message
+                );
+                assert!(
+                    !err.message.contains("secret-marker") && !err.message.contains("key\u{0003}"),
+                    "error must not echo credential material: {}",
+                    err.message
+                );
+            }
+            other => panic!("expected loud credential error, got {other:?}"),
+        }
+    }
+
+    /// COV-12, loud-credential arm (Basic): a Basic credential pair
+    /// with invalid HTTP header bytes fails loudly like Bearer/ApiKey
+    /// (FWD-08 family).
+    #[tokio::test]
+    async fn basic_credential_with_invalid_value_fails_loudly_without_echoing_secrets() {
+        let base = spawn_responder(TestArc::new(|_parts| {
+            http_response(200, "application/json", b"{}".to_vec())
+        }))
+        .await;
+        let ctx = ctx_with_capability("svc", "basic\u{0001}-secret-marker".to_string());
+        let envelope = call_forward_authed(&base, ctx, &Some(HttpAuthScheme::Basic)).await;
+        match envelope.result {
+            Err(err) => {
+                assert!(
+                    err.message
+                        .contains("refusing to send the request unauthenticated"),
+                    "message was: {}",
+                    err.message
+                );
+                assert!(
+                    !err.message.contains("secret-marker")
+                        && !err.message.contains("basic\u{0001}"),
+                    "error must not echo credential material: {}",
+                    err.message
+                );
+            }
+            other => panic!("expected loud credential error, got {other:?}"),
+        }
+    }
+
+    /// COV-12: a declared header parameter with an invalid *value*
+    /// (control bytes) is rejected loudly at build time rather than
+    /// silently dropped from the request (FWD-08 family).
+    #[test]
+    fn declared_header_param_with_invalid_value_is_rejected() {
+        let ctx = noop_context();
+        let schema = json!({
+            "type": "object",
+            "properties": {"X-Trace": {"type": "string", "wire": "header"}},
+        });
+        let err = build_request(
+            "https://api.example.com",
+            "/x",
+            "GET",
+            &None,
+            &TestHashMap::new(),
+            "svc",
+            &schema,
+            &json!({"X-Trace": "bad\u{0000}value"}),
+            &ctx,
+        )
+        .expect_err("invalid header-param value must fail loudly");
+        assert!(
+            err.message.contains("X-Trace") && err.message.contains("invalid value"),
+            "message was: {}",
+            err.message
+        );
+    }
+
+    /// COV-12: a declared header parameter whose name is not a valid
+    /// HTTP header name is rejected loudly at build time (FWD-08
+    /// family, the header-param sibling of the default-header arm).
+    #[test]
+    fn declared_header_param_with_invalid_name_is_rejected() {
+        let ctx = noop_context();
+        let schema = json!({
+            "type": "object",
+            "properties": {"bad header": {"type": "string", "wire": "header"}},
+        });
+        let err = build_request(
+            "https://api.example.com",
+            "/x",
+            "GET",
+            &None,
+            &TestHashMap::new(),
+            "svc",
+            &schema,
+            &json!({"bad header": "v"}),
+            &ctx,
+        )
+        .expect_err("invalid header-param name must fail loudly");
+        assert!(
+            err.message.contains("valid HTTP header name"),
+            "message was: {}",
+            err.message
+        );
+    }
+
+    /// CLI-02 (a): a same-host 302 is followed, and the credential
+    /// header is forwarded to the followed hop — the FWD-03 policy only
+    /// scrubs headers across *cross-host* hops, so a same-origin hop
+    /// must stay authenticated end to end.
+    #[tokio::test]
+    async fn same_host_redirect_is_followed_with_credential_forwarding() {
+        let hits = TestArc::new(std::sync::atomic::AtomicU32::new(0));
+        let hit_counter = TestArc::clone(&hits);
+        let base = spawn_responder(TestArc::new(move |parts| {
+            hit_counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            if parts.target == "/x" {
+                http::Response::builder()
+                    .status(302)
+                    .header("location", "/final")
+                    .body(Vec::new())
+                    .expect("redirect response builds")
+            } else {
+                let auth = parts
+                    .headers
+                    .get("authorization")
+                    .cloned()
+                    .unwrap_or_default();
+                let body = format!(r#"{{"auth":"{auth}"}}"#);
+                http_response(200, "application/json", body.into_bytes())
+            }
+        }))
+        .await;
+        let ctx = ctx_with_capability("svc", "tok-secret-marker".to_string());
+        let envelope = call_forward_authed(&base, ctx, &Some(HttpAuthScheme::Bearer)).await;
+        match envelope.result {
+            Ok(value) => assert_eq!(
+                value,
+                json!({"auth": "Bearer tok-secret-marker"}),
+                "the followed hop must receive the credential header"
+            ),
+            other => panic!("same-host redirect must be followed to success, got {other:?}"),
+        }
+        assert_eq!(
+            hits.load(std::sync::atomic::Ordering::SeqCst),
+            2,
+            "origin request plus exactly one followed hop"
+        );
+    }
+
+    /// CLI-02 (b), the FWD-03 load-bearing property: a cross-host 302 is
+    /// surfaced to the caller as `HTTP_302` and the redirected-to host
+    /// receives zero requests — credential-bearing or not, no second
+    /// request ever leaves for another host.
+    #[tokio::test]
+    async fn cross_host_redirect_is_surfaced_and_the_target_receives_zero_requests() {
+        let attacker_hits = TestArc::new(std::sync::atomic::AtomicU32::new(0));
+        let attacker_counter = TestArc::clone(&attacker_hits);
+        let attacker = spawn_responder(TestArc::new(move |_| {
+            attacker_counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            http_response(200, "text/plain", b"stolen".to_vec())
+        }))
+        .await;
+        let base = spawn_responder(TestArc::new(move |_| {
+            http::Response::builder()
+                .status(302)
+                .header("location", format!("{attacker}/steal"))
+                .body(Vec::new())
+                .expect("redirect response builds")
+        }))
+        .await;
+        let ctx = ctx_with_capability("svc", "tok-secret-marker".to_string());
+        let envelope = call_forward_authed(&base, ctx, &Some(HttpAuthScheme::Bearer)).await;
+        match envelope.result {
+            Err(err) => {
+                assert_eq!(err.code, "HTTP_302", "message was: {}", err.message);
+                assert!(
+                    !err.message.contains("tok-secret-marker"),
+                    "error must carry no credential material: {}",
+                    err.message
+                );
+            }
+            other => panic!("cross-host redirect must surface the 3xx, got {other:?}"),
+        }
+        assert_eq!(
+            attacker_hits.load(std::sync::atomic::Ordering::SeqCst),
+            0,
+            "the cross-host target must receive zero requests"
+        );
+    }
+
+    /// CLI-02 (c): a same-host redirect loop runs into the hop cap and
+    /// fails loudly as an `INTERNAL` transport error naming the
+    /// redirect machinery — never silently looping or surfacing a
+    /// partial body.
+    #[tokio::test]
+    async fn redirect_hop_cap_errors_loudly() {
+        let hits = TestArc::new(std::sync::atomic::AtomicU32::new(0));
+        let hit_counter = TestArc::clone(&hits);
+        let base = spawn_responder(TestArc::new(move |_| {
+            hit_counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            http::Response::builder()
+                .status(302)
+                .header("location", "/next")
+                .body(Vec::new())
+                .expect("redirect response builds")
+        }))
+        .await;
+        let envelope = call_forward(&base, noop_context()).await;
+        match envelope.result {
+            Err(err) => {
+                assert_eq!(err.code, "INTERNAL", "message was: {}", err.message);
+                assert!(
+                    err.message.contains("HTTP request failed"),
+                    "message was: {}",
+                    err.message
+                );
+                assert!(
+                    err.message.contains("redirect"),
+                    "message was: {}",
+                    err.message
+                );
+            }
+            other => panic!("hop-cap redirect must error loudly, got {other:?}"),
+        }
+        let hit_count = hits.load(std::sync::atomic::Ordering::SeqCst);
+        assert!(
+            hit_count >= 2,
+            "the chain followed before capping: {hit_count}"
+        );
+        assert!(
+            hit_count <= 12,
+            "the hop cap bounded the redirect chain: {hit_count}"
+        );
+    }
 }

tests/client_config_reload.rs

@@ -0,0 +1,33 @@
+//! COV-12/FWD-12: `SharedHttpClient::config()` reflects the config the
+//! running clients were built from, and a `reload` swaps the config
+//! together with both clients in one atomic store — after a reload,
+//! `config()` shows the new generation and the old one is unreachable
+//! for every late reader (FWD-12's atomicity assertion).
+
+use std::time::Duration;
+
+use alkhttp::client::{HttpClientConfig, SharedHttpClient};
+
+#[test]
+fn config_accessor_tracks_a_reloaded_config() {
+    let initial = HttpClientConfig {
+        retry_after_ceiling: Duration::from_secs(11),
+        ..HttpClientConfig::default()
+    };
+    let http = SharedHttpClient::new(initial).expect("initial client builds");
+    assert_eq!(
+        http.config().retry_after_ceiling,
+        Duration::from_secs(11),
+        "config() starts as the config the client was built from"
+    );
+    let reloaded = HttpClientConfig {
+        retry_after_ceiling: Duration::from_secs(22),
+        ..HttpClientConfig::default()
+    };
+    futures::executor::block_on(http.reload(reloaded)).expect("reload succeeds");
+    assert_eq!(
+        http.config().retry_after_ceiling,
+        Duration::from_secs(22),
+        "config() reflects the reloaded generation, not the initial one (FWD-12)"
+    );
+}

tests/client_tls.rs

@@ -21,7 +21,7 @@ use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
-use alkhttp::client::{ClientCertConfig, HttpClientConfig, SharedHttpClient};
+use alkhttp::client::{ClientCertConfig, HttpClientBuildError, HttpClientConfig, SharedHttpClient};
 
 /// A throwaway private PKI: CA, server leaf for `127.0.0.1`/`localhost`,
 /// and a client leaf, freshly minted per test.
@@ -373,3 +373,105 @@ async fn reload_to_a_ca_bundle_backed_client_succeeds() {
     cleanup_dir(&dir);
     tokio::time::sleep(Duration::from_millis(1)).await;
 }
+
+/// 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
+/// yields zero sections and leaves the trust store empty — reqwest
+/// accepts it — so the failure arm needs a structurally broken PEM.)
+#[test]
+fn corrupt_ca_bundle_fails_ca_bundle_parse_with_path() {
+    let dir = std::env::temp_dir().join(format!(
+        "alkhttp-pem-parse-{}-{}",
+        std::process::id(),
+        uuid::Uuid::new_v4()
+    ));
+    std::fs::create_dir_all(&dir).expect("temp dir");
+    let ca_path = dir.join("ca.pem");
+    std::fs::write(
+        &ca_path,
+        b"-----BEGIN CERTIFICATE-----\n!!not-base64!!\n-----END CERTIFICATE-----\n",
+    )
+    .expect("write corrupt pem");
+    let error = SharedHttpClient::new(client_config(Some(ca_path.clone()), None))
+        .expect_err("a corrupt CA bundle must fail the build");
+    match error {
+        HttpClientBuildError::CaBundleParse { path, .. } => {
+            assert_eq!(path, ca_path, "the error names the unparseable path");
+        }
+        other => panic!("expected CaBundleParse, got {other:?}"),
+    }
+    cleanup_dir(&dir);
+}
+
+/// COV-11/CLI-03 parse-failure arm: client cert files that exist but do
+/// not form a valid reqwest `Identity` fail the build with
+/// `ClientCertParse` carrying the cert path — and the message never
+/// carries key material (the PEM bytes are never echoed).
+#[test]
+fn garbage_client_cert_fails_client_cert_parse_with_path_and_no_key_material() {
+    let dir = std::env::temp_dir().join(format!(
+        "alkhttp-pem-parse-{}-{}",
+        std::process::id(),
+        uuid::Uuid::new_v4()
+    ));
+    std::fs::create_dir_all(&dir).expect("temp dir");
+    let cert_path = dir.join("client-cert.pem");
+    let key_path = dir.join("client-key.pem");
+    std::fs::write(
+        &cert_path,
+        b"-----BEGIN GARBAGE-----\nnope\n-----END GARBAGE-----\n",
+    )
+    .expect("write garbage cert");
+    std::fs::write(
+        &key_path,
+        b"-----BEGIN PRIVATE KEY-----\nnot-a-key\n-----END PRIVATE KEY-----\n",
+    )
+    .expect("write garbage key");
+    let key_marker = "not-a-key";
+    let error = SharedHttpClient::new(client_config(
+        None,
+        Some(ClientCertConfig {
+            cert_pem: cert_path.clone(),
+            key_pem: key_path,
+        }),
+    ))
+    .expect_err("a non-parseable client identity must fail the build");
+    let rendered = format!("{error}");
+    match error {
+        HttpClientBuildError::ClientCertParse { path, .. } => {
+            assert_eq!(path, cert_path, "the error names the identity's cert path");
+        }
+        other => panic!("expected ClientCertParse, got {other:?}"),
+    }
+    assert!(
+        !rendered.contains(key_marker),
+        "the error must never echo key material: {rendered}"
+    );
+    cleanup_dir(&dir);
+}
+
+/// FWD-12, config() half: after a `reload`, `config()` reflects the
+/// reloaded generation (the companion wire test lives in
+/// tests/client_config_reload.rs); here it is pinned on the TLS-config
+/// path where reload rebuilds from PEM files.
+#[tokio::test]
+async fn config_accessor_reflects_a_tls_config_reload() {
+    let pki = TestPki::generate();
+    let (ca, _cert, dir) = pki.write_config_files(false);
+
+    let http = SharedHttpClient::new(HttpClientConfig::default()).expect("initial client");
+    assert!(
+        http.config().ca_bundle.is_none(),
+        "the initial config has no CA bundle"
+    );
+    http.reload(client_config(ca, None))
+        .await
+        .expect("reload with the CA bundle succeeds");
+    let visible = http.config();
+    assert!(
+        visible.ca_bundle.is_some(),
+        "config() reflects the reloaded generation's CA bundle (FWD-12)"
+    );
+    cleanup_dir(&dir);
+}

tests/retry_policy_wire.rs

@@ -0,0 +1,164 @@
+//! Retry-policy wire coverage (review-002 CLI-03, FWD-04): counting
+//! responders against the real `SharedHttpClient` middleware stack.
+//!
+//! Pinned here:
+//!
+//! - the method gate — a POST gets exactly one upstream hit even when
+//!   the upstream answers 500 (non-idempotent requests never re-send);
+//! - the idempotent retry path — a GET that 500s twice then succeeds
+//!   results in exactly three upstream hits;
+//! - the wall-clock budget — with a generous attempt cap but a small
+//!   `max_total_retry_duration`, an always-500 upstream stops the
+//!   retry loop on budget exhaustion, bounding both the wall time and
+//!   the hit count.
+
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use alkhttp::client::{HttpClientConfig, RetryConfig, SharedHttpClient};
+use std::sync::Mutex;
+
+/// A raw-TCP HTTP/1.1 responder that counts accepted connections and
+/// answers each request from a per-attempt script: every entry is one
+/// response head; the last entry repeats when the script runs out.
+struct ScriptedResponder {
+    addr: std::net::SocketAddr,
+    hits: Arc<AtomicU32>,
+}
+
+impl ScriptedResponder {
+    async fn spawn(script: Vec<&'static str>) -> Self {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("responder binds");
+        let addr = listener.local_addr().expect("responder address");
+        let hits = Arc::new(AtomicU32::new(0));
+        let script = Arc::new(Mutex::new(script));
+        let hits_loop = Arc::clone(&hits);
+        let script_loop = Arc::clone(&script);
+        tokio::spawn(async move {
+            loop {
+                let Ok((mut sock, _)) = listener.accept().await else {
+                    return;
+                };
+                hits_loop.fetch_add(1, Ordering::SeqCst);
+                let head = {
+                    let mut queue = script_loop.lock().unwrap_or_else(|e| e.into_inner());
+                    if queue.len() <= 1 {
+                        queue
+                            .first()
+                            .copied()
+                            .unwrap_or("HTTP/1.1 500 Internal Server Error")
+                    } else {
+                        queue.remove(0)
+                    }
+                };
+                let body = "";
+                let response = format!(
+                    "{head}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n",
+                    body.len()
+                );
+                let _ = tokio::io::AsyncWriteExt::write_all(&mut sock, response.as_bytes()).await;
+                tokio::io::AsyncWriteExt::shutdown(&mut sock).await.ok();
+            }
+        });
+        Self { addr, hits }
+    }
+
+    fn hits(&self) -> u32 {
+        self.hits.load(Ordering::SeqCst)
+    }
+}
+
+fn retrying_config(max_total_retry_duration: Duration) -> HttpClientConfig {
+    HttpClientConfig {
+        request_timeout: Some(Duration::from_secs(5)),
+        connect_timeout: Some(Duration::from_secs(2)),
+        read_timeout: Some(Duration::from_secs(2)),
+        retry: RetryConfig {
+            max_retries: 50,
+            initial_backoff: Duration::from_millis(10),
+            max_retry_interval: Duration::from_millis(50),
+        },
+        max_total_retry_duration,
+        ..HttpClientConfig::default()
+    }
+}
+
+#[tokio::test]
+async fn post_500_receives_exactly_one_upstream_hit() {
+    let responder = ScriptedResponder::spawn(vec!["HTTP/1.1 500 Internal Server Error"]).await;
+    let http =
+        SharedHttpClient::new(retrying_config(Duration::from_secs(10))).expect("client builds");
+    let response = http
+        .client()
+        .post(format!("http://{}/mutate", responder.addr))
+        .send()
+        .await
+        .expect("500 is a delivered response, not a transport error");
+    assert_eq!(response.status(), 500);
+    assert_eq!(
+        responder.hits(),
+        1,
+        "the method gate must bypass the retry middleware for POST: a non-idempotent request is never re-sent"
+    );
+}
+
+#[tokio::test]
+async fn get_500_twice_then_success_is_exactly_three_hits() {
+    let responder = ScriptedResponder::spawn(vec![
+        "HTTP/1.1 500 Internal Server Error",
+        "HTTP/1.1 503 Service Unavailable",
+        "HTTP/1.1 200 OK",
+    ])
+    .await;
+    let http =
+        SharedHttpClient::new(retrying_config(Duration::from_secs(10))).expect("client builds");
+    let response = http
+        .client()
+        .get(format!("http://{}/flaky", responder.addr))
+        .send()
+        .await
+        .expect("retry-to-success delivers the final response");
+    assert_eq!(response.status(), 200);
+    assert_eq!(
+        responder.hits(),
+        3,
+        "two failed attempts retried, third attempt succeeded"
+    );
+}
+
+#[tokio::test]
+async fn budget_exhaustion_stops_retries_despite_a_generous_attempt_cap() {
+    let responder = ScriptedResponder::spawn(vec!["HTTP/1.1 500 Internal Server Error"]).await;
+    let http =
+        SharedHttpClient::new(retrying_config(Duration::from_millis(400))).expect("client builds");
+    let started = std::time::Instant::now();
+    let response = http
+        .client()
+        .get(format!("http://{}/stuck", responder.addr))
+        .send()
+        .await
+        .expect("500 is a delivered response, not a transport error");
+    assert_eq!(response.status(), 500);
+    let hits = responder.hits();
+    assert_eq!(
+        response.status(),
+        500,
+        "the surfaced status is the final upstream 500"
+    );
+    assert!(
+        hits >= 2,
+        "at least one retry ran before the budget closed, hits: {hits}"
+    );
+    assert!(
+        hits < 50,
+        "budget exhaustion must stop retries long before the 50-attempt cap, hits: {hits}"
+    );
+    assert!(
+        started.elapsed() < Duration::from_secs(3),
+        "wall time is bounded by max_total_retry_duration + one attempt, took {:?} over {hits} hits",
+        started.elapsed()
+    );
+}