lean-ctx-proxy 3.12.2

The LeanCTX model-API proxy: provider adapters, compression and accounting on top of LeanCTX | OSS.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
// SPDX-License-Identifier: Apache-2.0

//! Upstream URL construction and request/response header relay.

#[allow(clippy::wildcard_imports)]
use super::*;

#[test]
fn upstream_url_preserves_subpath() {
    let base = "https://api.anthropic.com";
    let parts = parts_for("/v1/messages/count_tokens");
    assert_eq!(
        crate::codec::build_upstream_url(&parts, base, "/v1/messages"),
        "https://api.anthropic.com/v1/messages/count_tokens"
    );
}

#[test]
fn upstream_url_preserves_batches_subpath() {
    let base = "https://api.anthropic.com";
    let parts = parts_for("/v1/messages/batches/batch_123/results");
    assert_eq!(
        crate::codec::build_upstream_url(&parts, base, "/v1/messages"),
        "https://api.anthropic.com/v1/messages/batches/batch_123/results"
    );
}

#[test]
fn upstream_url_exact_path() {
    let base = "https://api.anthropic.com";
    let parts = parts_for("/v1/messages");
    assert_eq!(
        crate::codec::build_upstream_url(&parts, base, "/v1/messages"),
        "https://api.anthropic.com/v1/messages"
    );
}

#[test]
fn upstream_url_preserves_query_params() {
    let base = "https://api.anthropic.com";
    let parts = parts_for("/v1/messages/count_tokens?model=claude-4");
    assert_eq!(
        crate::codec::build_upstream_url(&parts, base, "/v1/messages"),
        "https://api.anthropic.com/v1/messages/count_tokens?model=claude-4"
    );
}

#[test]
fn forwards_openai_project_and_auth_headers() {
    // #366: project-scoped OpenAI keys carry the scope via `OpenAI-Project`.
    // It must be forwarded upstream, otherwise the Responses API rejects the
    // call with `Missing scopes: api.responses.write`.
    for required in ["authorization", "openai-project", "openai-organization"] {
        assert!(
            ALLOWED_REQUEST_HEADERS.contains(&required),
            "request header `{required}` must be forwarded upstream"
        );
    }
}

#[test]
fn forwards_chatgpt_codex_oauth_headers() {
    for required in [
        "authorization",
        "chatgpt-account-id",
        "x-openai-fedramp",
        "x-openai-internal-codex-residency",
        "x-openai-product-sku",
        "oai-product-sku",
        "x-client-request-id",
        "x-codex-installation-id",
        "x-codex-turn-metadata",
        "x-openai-subagent",
        "x-codex-turn-state",
        "originator",
    ] {
        assert!(
            is_allowed_request_header(required),
            "request header `{required}` must be forwarded upstream"
        );
    }
}

#[test]
fn forwards_streamable_http_mcp_headers() {
    for required in ["mcp-session-id", "last-event-id"] {
        assert!(
            ALLOWED_REQUEST_HEADERS.contains(&required),
            "request header `{required}` must be forwarded upstream"
        );
    }
    assert!(
        is_forwarded_response_header("mcp-session-id"),
        "MCP session id response header must be forwarded downstream"
    );
}

#[test]
fn forwards_grok_cli_chat_proxy_headers() {
    // Grok CLI → cli-chat-proxy.grok.com. Stripping these used to yield
    // HTTP 426 Upgrade Required with client-version "(none)" on /responses.
    for required in [
        "x-xai-token-auth",
        "x-models-etag",
        "x-grok-client-version",
        "x-grok-client-identifier",
        "x-grok-client-mode",
        "x-grok-client-surface",
        "x-grok-model-override",
        "x-grok-agent-id",
        "x-grok-session-id",
        "x-grok-turn-id",
        "x-grok-conv-id",
        "x-grok-req-id",
        "x-grok-deployment-id",
        "x-grok-user-id",
        "x-grok-context-window",
        "x-grok-max-completion-tokens",
        "x-grok-doom-loop-check",
        "x-grok-managed-gateway",
    ] {
        assert!(
            ALLOWED_REQUEST_HEADERS.contains(&required),
            "request header `{required}` must be on the allowlist"
        );
    }
    // Internal gateway tag must stay off the allowlist (enterprise#11).
    assert!(!is_allowed_request_header("x-leanctx-project"));
    // The presenter capability is a host-internal secret: never on the
    // allowlist, in no spelling the relay would accept.
    assert!(!is_allowed_request_header(
        crate::request_project::PRESENTER_HEADER
    ));
    assert!(!should_forward_request_header(
        crate::request_project::PRESENTER_HEADER,
        true
    ));
}

#[test]
fn forwards_codex_state_response_headers() {
    for required in [
        "x-codex-turn-state",
        "x-codex-primary-used-percent",
        "openai-model",
        "x-models-etag",
        "x-reasoning-included",
        "x-oai-request-id",
        "cf-ray",
        "x-openai-authorization-error",
        "x-error-json",
    ] {
        assert!(
            is_forwarded_response_header(required),
            "response header `{required}` must be forwarded downstream"
        );
    }
}

/// #1638: Claude Code fills its plan-usage windows *only* from the
/// `anthropic-ratelimit-unified-*` response headers. The allowlist enumerated
/// the four legacy `requests-`/`tokens-` names, so when the `unified-*` family
/// arrived upstream the proxy silently dropped all of it: `rate_limits`
/// disappeared from the statusline payload, and — the part that actually
/// matters — the near-limit warning machinery reads the same state, so a user
/// behind the proxy was never warned before hitting a limit.
///
/// The names below are the full family as parsed by Claude Code 2.1.236,
/// quoted from the report. They are asserted individually rather than through
/// the prefix, so narrowing the rule back to an enumeration fails here.
#[test]
fn forwards_the_whole_anthropic_ratelimit_family() {
    for required in [
        "anthropic-ratelimit-unified-status",
        "anthropic-ratelimit-unified-reset",
        "anthropic-ratelimit-unified-5h-utilization",
        "anthropic-ratelimit-unified-5h-reset",
        "anthropic-ratelimit-unified-5h-surpassed-threshold",
        "anthropic-ratelimit-unified-7d-utilization",
        "anthropic-ratelimit-unified-7d-reset",
        "anthropic-ratelimit-unified-7d-surpassed-threshold",
        "anthropic-ratelimit-unified-grace-status",
        "anthropic-ratelimit-unified-grace-5h-utilization",
        "anthropic-ratelimit-unified-grace-7d-utilization",
        "anthropic-ratelimit-unified-overage-status",
        "anthropic-ratelimit-unified-overage-utilization",
        "anthropic-ratelimit-unified-overage-reset",
        "anthropic-ratelimit-unified-overage-period",
        "anthropic-ratelimit-unified-representative-claim",
        "anthropic-ratelimit-unified-fallback",
        "anthropic-ratelimit-unified-upgrade-paths",
        // The legacy names the allowlist used to carry explicitly must keep
        // working — the prefix replaced the enumeration, not the coverage.
        "anthropic-ratelimit-requests-limit",
        "anthropic-ratelimit-requests-remaining",
        "anthropic-ratelimit-tokens-limit",
        "anthropic-ratelimit-tokens-remaining",
    ] {
        assert!(
            is_forwarded_response_header(required),
            "response header `{required}` carries the client's own quota state \
             and must reach it"
        );
    }
}

/// Also reported in #1638: without `request-id` a user cannot correlate a
/// failed request with Anthropic support, and `x-should-retry` is what tells
/// an SDK whether a failure is worth retrying at all.
#[test]
fn forwards_request_correlation_and_retry_headers() {
    for required in ["request-id", "x-should-retry", "retry-after"] {
        assert!(
            is_forwarded_response_header(required),
            "response header `{required}` must reach the client"
        );
    }
}

/// The fix widens one family; it must not turn the allowlist into a
/// pass-through. Hop-by-hop and framing headers stay out — the proxy rewrites
/// the body (decompression, SSE re-framing), so relaying the upstream's
/// framing would describe a response that no longer exists.
#[test]
fn the_allowlist_stays_an_allowlist() {
    for denied in [
        "transfer-encoding",
        "connection",
        "content-length",
        "server",
        "set-cookie",
    ] {
        assert!(
            !is_forwarded_response_header(denied),
            "`{denied}` must not be relayed"
        );
    }
}

#[test]
fn chatgpt_responses_use_openai_responses_holdout_key() {
    let _iso = lean_ctx::core::data_dir::isolated_data_dir();
    lean_ctx::core::config::Config::update_global(|c| {
        c.proxy.output_holdout = Some(1.0);
    })
    .unwrap();

    let body = serde_json::json!({
        "model": "gpt-5",
        "input": "same conversation",
    });

    assert_eq!(
        cohort_arm(&body, "ChatGPT", "/backend-api/codex/responses"),
        cohort_arm(&body, "OpenAI", "/v1/responses")
    );
}

#[test]
fn cache_prompt_hash_is_content_sensitive() {
    let hash = super::super::super::ocla_cache_bridge::prompt_hash;
    assert_ne!(hash(b"one"), hash(b"two"));
}

/// Loopback upstream that answers exactly one request and hands back its raw,
/// lowercased request header block. Lets a test assert what actually leaves the
/// proxy, not only what the allowlist constant claims.
async fn upstream_capturing_headers() -> (String, tokio::task::JoinHandle<String>) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    let server = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let mut request = Vec::new();
        let mut buffer = [0_u8; 1024];
        let header_end = loop {
            let read = stream.read(&mut buffer).await.unwrap();
            if read == 0 {
                break request.len();
            }
            request.extend_from_slice(&buffer[..read]);
            if let Some(end) = request.windows(4).position(|window| window == b"\r\n\r\n") {
                break end + 4;
            }
        };
        stream
            .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\n{}")
            .await
            .unwrap();
        String::from_utf8_lossy(&request[..header_end]).to_lowercase()
    });
    (format!("http://{address}"), server)
}

/// Minimal state for the transport layer. `send_upstream` reads only
/// `state.client`; the watch receiver keeps serving its last value after the
/// sender drops, so no keep-alive is needed.
fn relay_test_state() -> ProxyState {
    let (_upstreams, state_upstreams) =
        tokio::sync::watch::channel(Arc::new(lean_ctx::core::config::Upstreams {
            anthropic: "https://api.anthropic.com".into(),
            openai: "https://api.openai.com".into(),
            chatgpt: "https://chatgpt.com".into(),
            gemini: "https://generativelanguage.googleapis.com".into(),
            providers: Vec::new(),
        }));
    ProxyState {
        client: reqwest::Client::new(),
        port: 0,
        stats: Arc::new(crate::ProxyStats::default()),
        break_even: Arc::new(crate::break_even::BreakEvenCalculator::new(1500)),
        introspect: Arc::new(crate::introspect::IntrospectState::default()),
        ocla_cache: None,
        upstreams: state_upstreams,
        chatgpt_cookies: crate::chatgpt_cookies::shared_chatgpt_cloudflare_cookie_store(),
        mcp_servers: Arc::new(Vec::new()),
    }
}

/// The transport sends admitted bodies only: admit the empty-context `{}`
/// these header tests relay and bind it, with `head`, to the loopback
/// upstream.
async fn admitted_relay_body(
    upstream: &str,
    head: crate::admitted::RequestHead,
) -> crate::admitted::AdmittedBody {
    let target = lean_ctx::core::context_admission::egress::EgressTarget {
        provider: "Anthropic",
        model: None,
        upstream_base: upstream,
    };
    let admission = crate::admitted::admit(
        crate::admitted::Payload::Opaque(0),
        &target,
        lean_ctx::core::context_admission::egress::TrustedDestinationBinding::default(),
    )
    .expect("an empty payload is admitted");
    crate::admitted::AdmittedBody::bind(
        admission,
        b"{}".to_vec(),
        upstream,
        upstream,
        head,
        2,
        None,
    )
    .await
    .expect("the admitted body binds to its own upstream")
}

fn built_head(value: &'static str) -> crate::admitted::RequestHead {
    crate::admitted::RequestHead::new(
        reqwest::Method::POST,
        [(
            axum::http::HeaderName::from_static("x-built"),
            axum::http::HeaderValue::from_static(value),
        )],
    )
}

/// AP10-C2: the caller's code that finished a request could add, change or
/// drop headers after the body was bound, and they left with it. Now the
/// head is bound with the body: a request whose finished head is not the
/// bound one is refused before it is handed to the client.
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn a_header_the_bound_head_does_not_carry_is_not_sent() {
    let _data = lean_ctx::core::data_dir::isolated_data_dir();
    let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let upstream = format!("http://{}", listener.local_addr().unwrap());
    let body = admitted_relay_body(&upstream, built_head("bound")).await;
    let client = reqwest::Client::new();
    let later: [fn(reqwest::RequestBuilder) -> reqwest::RequestBuilder; 2] = [
        |request| request.header("authorization", "Bearer later"),
        |request| request.header("x-built", "changed"),
    ];
    for decorate in later {
        let refused = body.send(&client, 1, decorate).await;
        let Err(refusal) = refused else {
            panic!("a request with another head was handed to the client");
        };
        assert_eq!(
            refusal.response("Anthropic").headers()[crate::admitted::REASON_CODES_HEADER],
            "egress.head_unbound"
        );
    }
    assert!(
        tokio::time::timeout(std::time::Duration::from_millis(200), listener.accept())
            .await
            .is_err(),
        "the upstream sees no connection"
    );
}

/// The authority is proven on the request that is actually sent. A policy
/// that changes while the caller's own code builds the request stops it:
/// the first attempt of an admitted body goes out, the next one of the same
/// body does not, and the upstream sees exactly one connection.
#[tokio::test]
#[serial_test::serial(cache_telemetry)]
async fn a_policy_change_while_the_request_is_built_stops_the_send() {
    let _data = lean_ctx::core::data_dir::isolated_data_dir();
    let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
    write_gateway_config("[context_gateway]\nmode = \"developer\"\n");
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let upstream = format!("http://{}", listener.local_addr().unwrap());
    let body = admitted_relay_body(&upstream, built_head("live")).await;
    let client = reqwest::Client::new();
    let built = std::cell::Cell::new(0);

    // Live: the decorated request reaches the bound upstream. The answer
    // closes the connection, so a second dispatch would have to open one.
    let serve = async {
        let (mut stream, _) = listener.accept().await.unwrap();
        let mut request = Vec::new();
        let mut buffer = [0_u8; 1024];
        while !request.windows(4).any(|window| window == b"\r\n\r\n") {
            let read = stream.read(&mut buffer).await.unwrap();
            assert!(read > 0, "the request ended before its headers");
            request.extend_from_slice(&buffer[..read]);
        }
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nconnection: close\r\ncontent-length: 2\r\n\r\n{}")
            .await
            .unwrap();
        String::from_utf8_lossy(&request).to_lowercase()
    };
    let (first, seen) = tokio::join!(
        body.send(&client, 1, |request| {
            built.set(built.get() + 1);
            request
        }),
        serve
    );
    assert!(first.is_ok_and(|attempt| {
        attempt
            .response
            .is_ok_and(|response| response.status().is_success())
    }));
    assert!(seen.contains("x-built: live"), "{seen}");

    // The same admitted body again; the policy changes inside the caller's
    // decoration, after the first proof and before the request would leave.
    let refused = body
        .send(&client, 2, |request| {
            built.set(built.get() + 1);
            write_gateway_config("[context_gateway]\nmode = \"governed\"\n");
            request
        })
        .await;
    assert_eq!(built.get(), 2, "both requests were really built");
    let Err(refusal) = refused else {
        panic!("a request built under a changed policy was handed to the client");
    };
    assert_eq!(
        refusal.response("Anthropic").headers()[crate::admitted::REASON_CODES_HEADER],
        "policy.changed"
    );
    assert!(
        tokio::time::timeout(std::time::Duration::from_millis(200), listener.accept())
            .await
            .is_err(),
        "the ready upstream must see no second connection"
    );
}

#[test]
fn forwards_opencode_session_header() {
    // #1752: OpenCode (and OpenCode zen) key session affinity off their own
    // header. #1730 enumerated only the Claude Code spellings, so this sibling
    // kept being stripped and every relayed request looked like a new session.
    assert!(ALLOWED_REQUEST_HEADERS.contains(&"x-opencode-session"));
    assert!(is_allowed_request_header("x-opencode-session"));
    assert!(should_forward_request_header("x-opencode-session", false));
}

#[tokio::test]
async fn opencode_session_reaches_the_upstream_verbatim() {
    let _data = lean_ctx::core::data_dir::isolated_data_dir();
    let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
    // #1752 end-to-end: assert what the upstream actually receives, including
    // the mixed-case spelling OpenCode puts on the wire.
    let (upstream, server) = upstream_capturing_headers().await;
    let parts = Request::builder()
        .method("POST")
        .uri("/v1/messages")
        .header("X-OpenCode-Session", "0f9c2b71-4d3a-4e58-9c11-7a6e5b2d8f40")
        .header("X-Leanctx-Project", "internal-only")
        .header("X-Leanctx-Presenter", format!("lcp1_{}", "c3".repeat(32)))
        .body(())
        .unwrap()
        .into_parts()
        .0;

    let body = admitted_relay_body(&upstream, transport::forwarded_head(&parts, false)).await;
    let response = transport::send_upstream(&relay_test_state(), &body, "Anthropic")
        .await
        .unwrap()
        .response;
    assert!(response.status().is_success());

    let seen = server.await.unwrap();
    assert!(
        seen.contains("x-opencode-session: 0f9c2b71-4d3a-4e58-9c11-7a6e5b2d8f40"),
        "session header must reach the upstream verbatim, got: {seen}"
    );
    // Widening the allowlist for #1752 must not widen it for anything else.
    assert!(
        !seen.contains("x-leanctx-project"),
        "internal header must stay stripped, got: {seen}"
    );
    // The relay drops a presenter capability even where the handler did not
    // take it out of the headers first (this call bypasses the handler).
    assert!(
        !seen.contains("x-leanctx-presenter") && !seen.contains(&"c3".repeat(32)),
        "a presenter capability must never reach the upstream"
    );

    // The handler's own step: read from the headers only, removed there, and
    // exactly one well-formed value counts.
    let header = crate::request_project::PRESENTER_HEADER;
    let mut headers = parts.headers.clone();
    assert!(crate::request_project::take_presenter(&mut headers).is_some());
    assert!(
        !headers.contains_key(header),
        "taken out of the request before anything else sees it"
    );
    for _ in 0..2 {
        headers.append(header, parts.headers[header].clone());
    }
    assert!(crate::request_project::take_presenter(&mut headers).is_none());
    assert!(!headers.contains_key(header));
}

#[test]
fn forwards_claude_code_session_id_header() {
    // #1730: Claude Code's per-conversation UUID must survive the relay so a
    // downstream proxy can key session affinity and prompt-cache reuse on it.
    assert!(ALLOWED_REQUEST_HEADERS.contains(&"x-claude-code-session-id"));
    assert!(is_allowed_request_header("x-claude-code-session-id"));
    assert!(should_forward_request_header(
        "x-claude-code-session-id",
        false
    ));
}

#[tokio::test]
async fn claude_code_session_id_reaches_the_upstream_verbatim() {
    let _data = lean_ctx::core::data_dir::isolated_data_dir();
    let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
    // #1730 end-to-end: assert what the upstream actually receives, including
    // the mixed-case spelling Claude Code puts on the wire.
    let (upstream, server) = upstream_capturing_headers().await;
    let parts = Request::builder()
        .method("POST")
        .uri("/v1/messages")
        .header(
            "X-Claude-Code-Session-Id",
            "11111111-2222-3333-4444-555555555555",
        )
        .header("X-Leanctx-Project", "internal-only")
        .body(())
        .unwrap()
        .into_parts()
        .0;

    let body = admitted_relay_body(&upstream, transport::forwarded_head(&parts, false)).await;
    let response = transport::send_upstream(&relay_test_state(), &body, "Anthropic")
        .await
        .unwrap()
        .response;
    assert!(response.status().is_success());

    let seen = server.await.unwrap();
    assert!(
        seen.contains("x-claude-code-session-id: 11111111-2222-3333-4444-555555555555"),
        "session id must reach the upstream verbatim, got: {seen}"
    );
    // Widening the allowlist for #1730 must not widen it for anything else:
    // the internal gateway tag stays stripped (enterprise#11).
    assert!(
        !seen.contains("x-leanctx-project"),
        "internal header must stay stripped, got: {seen}"
    );
}

#[test]
fn forwards_commandcode_cli_headers() {
    // Command Code (`cmd`) gates agent calls on `x-command-code-version`.
    // Stripping it yields 403 upgrade_required ("CLI is out of date").
    for required in [
        "x-command-code-version",
        "x-cli-environment",
        "x-oauth-token",
        "x-oauth-provider",
        "x-project-slug",
        "x-taste-learning",
        "x-taste-usage",
        "x-oss-primary-provider",
        "x-system-prompt-breakdown",
        "x-cmd-zdr",
        "x-session-id", // shared; pin so CC session stays wired
    ] {
        assert!(
            ALLOWED_REQUEST_HEADERS.contains(&required),
            "Command Code CLI header `{required}` must be on the request allowlist"
        );
    }
}

/// Stub upstream that answers `200 {}` and yields the exact request body
/// bytes it received, so a test can assert byte-identity on the wire.
async fn upstream_capturing_body() -> (String, tokio::task::JoinHandle<Vec<u8>>) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    let server = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let mut request = Vec::new();
        let mut buffer = [0_u8; 4096];
        let header_end = loop {
            let read = stream.read(&mut buffer).await.unwrap();
            request.extend_from_slice(&buffer[..read]);
            if let Some(end) = request.windows(4).position(|window| window == b"\r\n\r\n") {
                break end + 4;
            }
        };
        let header = String::from_utf8_lossy(&request[..header_end]).to_lowercase();
        let content_length = header
            .lines()
            .find_map(|line| line.strip_prefix("content-length: "))
            .unwrap()
            .trim()
            .parse::<usize>()
            .unwrap();
        while request.len() < header_end + content_length {
            let read = stream.read(&mut buffer).await.unwrap();
            request.extend_from_slice(&buffer[..read]);
        }
        stream
            .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\n{}")
            .await
            .unwrap();
        request[header_end..header_end + content_length].to_vec()
    });
    (format!("http://{address}"), server)
}

/// #1912: a request the proxy does not change must reach the provider as the
/// client's exact bytes. Re-serializing reorders keys and rewrites number and
/// whitespace formatting, which moves the prompt-cache prefix every turn.
#[tokio::test]
async fn unchanged_request_is_forwarded_byte_identical() {
    let _iso = lean_ctx::core::data_dir::isolated_data_dir();
    let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
    let (upstream, server) = upstream_capturing_body().await;
    // Key order, spacing and `1.0` all differ from serde_json's canonical form.
    let raw = br#"{ "temperature": 1.0,  "model": "claude-sonnet-4-5",
  "messages": [ {"role": "user", "content": "hi"} ] }"#
        .to_vec();
    let request = Request::builder()
        .method("POST")
        .uri("/v1/messages")
        .header(CONTENT_TYPE, "application/json")
        .body(Body::from(raw.clone()))
        .unwrap();

    let response = forward_request(
        State(relay_test_state()),
        request,
        &upstream,
        "/v1/messages",
        |value, original_size| {
            let out = serde_json::to_vec(&value).unwrap();
            let size = out.len();
            (out, original_size, size)
        },
        "Anthropic",
        &[],
    )
    .await
    .unwrap();
    assert!(response.status().is_success());

    assert_eq!(
        server.await.unwrap(),
        raw,
        "an unchanged request must be forwarded as the client's exact bytes"
    );
}

/// A model alias that targets a `[[proxy.providers]]` entry with `api_key_env`
/// moves a built-in route onto the operator's key. A request that only got in
/// through the provider-key fallback must not have that key injected.
#[tokio::test]
async fn routed_gateway_credential_requires_lean_ctx_token() {
    let _isolation = lean_ctx::core::data_dir::isolated_data_dir();
    let mut config = lean_ctx::core::config::Config::default();
    config.proxy.routing.enabled = Some(true);
    config
        .proxy
        .routing
        .aliases
        .insert("acme/fast".into(), "foundry:gpt-4o-mini".into());
    config.save().unwrap();
    lean_ctx::test_env::set_var("LC_TEST_GATEWAY_HELD_KEY", "sk-gateway-held");
    let body = serde_json::json!({
        "model": "acme/fast",
        "messages": [{"role": "user", "content": "hi"}]
    });

    for fallback_only in [true, false] {
        let (upstream, server) = upstream_capturing_headers().await;
        let mut state = proxy_test_state("http://127.0.0.1:9");
        let (_sender, upstreams) =
            tokio::sync::watch::channel(Arc::new(lean_ctx::core::config::Upstreams {
                anthropic: "http://127.0.0.1:9".into(),
                openai: "http://127.0.0.1:9".into(),
                chatgpt: "http://127.0.0.1:9".into(),
                gemini: "http://127.0.0.1:9".into(),
                providers: vec![lean_ctx::core::config::ResolvedProvider {
                    id: "foundry".into(),
                    shape: lean_ctx::core::config::WireShape::OpenAi,
                    base_url: upstream,
                    api_key_env: Some("LC_TEST_GATEWAY_HELD_KEY".into()),
                    aws_region: None,
                    account_ref: None,
                    region: None,
                    local: true,
                }],
            }));
        state.upstreams = upstreams;
        let mut driver = proxy_test_driver(state, &body, "http://127.0.0.1:9");
        let request = driver.request.as_mut().unwrap();
        request
            .headers_mut()
            .insert("x-api-key", "caller-junk".parse().unwrap());
        if fallback_only {
            request
                .extensions_mut()
                .insert(crate::server::ProviderKeyFallback);
        }
        let context = proxy_test_context();
        driver.apply_security_boundaries(&context).await.unwrap();
        let prepared = driver.gather_context_strategy(&context).await;
        if fallback_only {
            assert_eq!(prepared.err(), Some(StatusCode::UNAUTHORIZED));
            server.abort();
        } else {
            prepared.unwrap();
            driver.dispatch_primitive(&context).await.unwrap();
            let headers = server.await.unwrap();
            assert!(
                headers.contains("authorization: bearer sk-gateway-held"),
                "token-authenticated alias traffic still gets the gateway key: {headers}"
            );
        }
    }
    lean_ctx::test_env::remove_var("LC_TEST_GATEWAY_HELD_KEY");
}