lean-ctx-proxy 3.12.3

// SPDX-License-Identifier: Apache-2.0
use axum::{
    body::Body,
    extract::State,
    http::{Request, StatusCode},
    response::Response,
};
use serde_json::Value;

use super::ProxyState;
use super::compress_shared::{self, ToolKind};
use super::forward;
use super::sticky_tools::{self, Shape};
use super::tool_kind::{self, ToolResultKind};
use super::{cache_safety, prose};
use lean_ctx::core::config::{HistoryMode, ProseRole};

/// Proxy handler for OpenAI's Responses API (`POST /v1/responses`).
///
/// The Responses API superseded Chat Completions for clients such as opencode
/// and the OpenAI Agents SDK. Its conversation turns live in `input` rather than
/// `messages`, so the Chat Completions handler never saw — and never compressed —
/// them. This handler reuses the same upstream, auth and streaming path but
/// understands the Responses-API request shape.
///
/// Retrieve / cancel / delete / input_items sub-paths
/// (`/v1/responses/{id}/...`) are routed here as well and pass through untouched:
/// they carry no `input` array, so `compress_request_body` is a no-op for them.
///
/// Handles `POST /v1/responses` (and the bare `/responses`) over HTTP/SSE.
pub async fn handler(
    State(state): State<ProxyState>,
    mut req: Request<Body>,
) -> Result<Response, StatusCode> {
    // #1685: a Codex ChatGPT-subscription login that reaches this rail is on the
    // wrong one, and `api.openai.com` says so with `401 … Missing scopes:
    // api.responses.write` — a message about organization roles that names
    // nothing real. Send it where it authenticates instead of forwarding it to a
    // certain failure and explaining the failure afterwards.
    if let Some(uri) = chatgpt_rail_uri(&state, &req) {
        *req.uri_mut() = uri;
        return super::chatgpt::codex_responses_handler(State(state), req).await;
    }
    let upstream = state.openai_upstream();
    forward::forward_request(
        State(state),
        req,
        &upstream,
        "/v1/responses",
        compress_request_body,
        "OpenAI",
        &[],
    )
    .await
}

/// The ChatGPT-rail URI this request belongs on, or `None` to forward it to the
/// OpenAI platform rail unchanged.
///
/// Re-routing is deliberately confined to the one configuration where the
/// current route cannot succeed: a **stock** `api.openai.com` upstream reached
/// with a bearer that is not an OpenAI API key but is a JWT. `api.openai.com`
/// accepts `sk-` keys only, so such a request is rejected no matter what we do —
/// which is what makes moving it safe. No working setup changes route: an Azure,
/// Copilot, LiteLLM or other gateway upstream is configured explicitly and fails
/// condition one, and an API key fails condition two.
///
/// This does not contradict #366. That issue is about the URL the *client* is
/// given — `OPENAI_BASE_URL` keeps its `/v1` so OpenCode's ChatGPT-OAuth plugin
/// still matches `/v1/responses`. Which upstream the proxy then forwards to is a
/// separate decision, and the credential that arrived answers it.
fn chatgpt_rail_uri(state: &ProxyState, req: &Request<Body>) -> Option<axum::http::Uri> {
    if !is_stock_openai_upstream(&state.openai_upstream()) {
        return None;
    }
    if !is_chatgpt_subscription_bearer(req.headers()) {
        return None;
    }
    let path = req.uri().path();
    let rest = path
        .strip_prefix("/v1/responses")
        .or_else(|| path.strip_prefix("/responses"))?;
    let query = req
        .uri()
        .query()
        .map_or_else(String::new, |q| format!("?{q}"));
    axum::http::Uri::try_from(format!("/backend-api/codex/responses{rest}{query}")).ok()
}

/// True for the default OpenAI endpoint — the only upstream for which
/// `chatgpt.com` is the meaningful alternative. Any configured gateway
/// (Azure, Copilot, LiteLLM, a self-hosted relay) returns false and is never
/// re-routed.
fn is_stock_openai_upstream(upstream: &str) -> bool {
    let host = upstream
        .trim_end_matches('/')
        .rsplit_once("//")
        .map_or(upstream, |(_, rest)| rest);
    let host = host.split('/').next().unwrap_or(host);
    let host = host.split(':').next().unwrap_or(host);
    host.eq_ignore_ascii_case("api.openai.com")
}

/// True when the `Authorization` bearer is a ChatGPT subscription token rather
/// than an OpenAI platform API key.
///
/// Platform keys are opaque `sk-…` strings and are never JWTs; a ChatGPT login
/// carries the OAuth access token Codex stores in `auth.json`, which is one.
/// The token is only ever inspected for its shape — never decoded, stored or
/// logged.
fn is_chatgpt_subscription_bearer(headers: &axum::http::HeaderMap) -> bool {
    let Some(token) = headers
        .get(axum::http::header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .map(str::trim)
        .filter(|t| !t.is_empty())
    else {
        return false;
    };
    if token.starts_with("sk-") {
        return false;
    }
    // A compact JWS: three base64url segments. The header of every JWT whose
    // first claim set starts with `{"alg"` encodes to this prefix.
    token.starts_with("eyJ") && token.split('.').count() == 3
}

/// Handles the WebSocket Responses transport on `GET /v1/responses`.
///
/// Codex (and the OpenAI SDK) default to `ws://…/responses` with one
/// `response.create` event per turn. Bridging the upgrade here lets the proxy be
/// a drop-in for Codex without forcing `supports_websockets = false` (#440); the
/// actual WS↔HTTP/SSE bridging lives in `openai_responses_ws`.
pub async fn ws_handler(
    State(state): State<ProxyState>,
    headers: axum::http::HeaderMap,
    ws: axum::extract::ws::WebSocketUpgrade,
) -> Response {
    super::openai_responses_ws::upgrade(state, ws, &headers)
}

/// One pass over a Responses request, as on the Messages path: the unit in
/// which a block that leaves with a recovery stub and the conversation it
/// belongs to are brought together (`sticky_tools`).
pub(super) fn compress_request_body(
    parsed: Value,
    original_size: usize,
) -> (Vec<u8>, usize, usize) {
    sticky_tools::in_pass(|| compress_in_pass(parsed, original_size))
}

fn compress_in_pass(parsed: Value, original_size: usize) -> (Vec<u8>, usize, usize) {
    let mut doc = parsed;
    // Under protection a block leaves shortened only with a stub that names
    // the Engine's expand tool as the host listed it in this request's own
    // `tools[].name`. Whether the request lists it and lets its model call
    // it is settled before any block is touched; if not, blocks stay whole.
    // The proxy adds no tool, so a request that continues a history the
    // provider holds (`previous_response_id`) is judged like any other: by
    // its own list.
    sticky_tools::begin(Shape::Responses, &doc);
    let cfg = lean_ctx::core::config::Config::load();
    // #1905: the input-compression control arm forwards the body uncompressed.
    let compression_control = super::holdout::in_compression_control();
    let system_aggr = cfg
        .proxy
        .resolved_role_aggressiveness(ProseRole::System)
        .filter(|_| !compression_control);
    let user_aggr = cfg
        .proxy
        .resolved_role_aggressiveness(ProseRole::User)
        .filter(|_| !compression_control);
    let live_compress = cfg.proxy.live_compresses() && !compression_control;
    let mode = if compression_control {
        HistoryMode::Off
    } else {
        cfg.proxy.resolved_history_mode()
    };
    // #493: in-band CCR expansion (opt-in). Splice any <lc_expand:HASH> the model
    // echoed back into the verbatim original from the local tee store. A strict
    // no-op when no marker is present (byte-identical body → cache-safe). Runs
    // before the meter-only short-circuit so an explicit expand request is
    // honored even when the proxy is otherwise byte-passthrough.
    let mut modified = false;
    // #895 Track B: output-savings holdout arm, from the pristine body (before any
    // mutation below) so it matches the arm the response meter records. Control
    // conversations skip output-shaping but are still metered. Default 0 → Treatment.
    let arm = super::holdout::assign(
        &super::holdout::openai_responses_key(&doc),
        cfg.proxy.output_holdout_fraction(),
    );
    if cfg.proxy.ccr_inband_enabled() {
        modified |= super::ccr::splice_inband_in_place(&mut doc);
    }
    // #834: cache-safe cross-provider effort control. Default off → no-op. The
    // value is a constant, so it never perturbs the prompt-cache prefix; it sets
    // `reasoning.effort` only on reasoning models and never overrides a
    // client-set value.
    if arm == super::holdout::Arm::Treatment {
        if let Some(effort) = cfg.proxy.resolved_effort() {
            modified |= super::effort::apply_openai_responses(&mut doc, effort);
        }
        // #895: cache-safe wire verbosity steer; control arm skips it (measured).
        if cfg.proxy.verbosity_steer_enabled() {
            modified |= super::verbosity::apply_openai_responses(&mut doc);
        }
    }
    // Meter-only (#481): live compression off and history pruning off → forward
    // the body unchanged while upstream usage metering still runs. A pending
    // in-band splice (`modified`) opts out: the body did change this turn.
    if !live_compress
        && mode == HistoryMode::Off
        && system_aggr.is_none()
        && user_aggr.is_none()
        && !modified
    {
        let out = serde_json::to_vec(&doc).unwrap_or_default();
        return (out, original_size, original_size);
    }
    // K26: large tool descriptions leave shortened (see the Messages path).
    if live_compress {
        modified |= super::tool_descriptions::shorten(&mut doc);
    }
    let mut prose_segments: u64 = 0;
    if let Some(a) = system_aggr {
        prose_segments += u64::from(prose::compress_string_field(&mut doc, "instructions", a));
    }
    // Two-stage, like the Chat Completions path: (1) cache-aware prune of the
    // frozen OLD region — old file reads collapse to re-read stubs, old logs
    // head/tail summarize — then (2) compress whatever recent outputs remain.
    // Stage 1 runs first so a stubbed old output isn't needlessly re-compressed.
    modified |= prune_responses_input(&mut doc);
    modified |= compress_responses_input(&mut doc);
    if let Some(a) = user_aggr {
        prose_segments += u64::from(compress_responses_user_prose(&mut doc, mode, a));
    }
    if prose_segments > 0 {
        modified = true;
    }
    cache_safety::record(prose_segments, true);
    let out = serde_json::to_vec(&doc).unwrap_or_default();
    let compressed_size = if modified { out.len() } else { original_size };
    (out, original_size, compressed_size)
}

fn compress_responses_user_prose(doc: &mut Value, mode: HistoryMode, aggressiveness: f64) -> u32 {
    let Some(input) = doc.get_mut("input").and_then(|i| i.as_array_mut()) else {
        return 0;
    };
    let boundary = super::history_prune::prune_boundary(mode, input.len());
    if boundary == 0 {
        return 0;
    }

    let mut segments = 0;
    for item in input.iter_mut().take(boundary) {
        let item_type = item
            .get("type")
            .and_then(|t| t.as_str())
            .unwrap_or("message");
        let role = item.get("role").and_then(|r| r.as_str());
        if item_type == "message" && role == Some("user") {
            segments += prose::compress_message_content(item, aggressiveness);
        }
    }
    segments
}

/// Cache-aware history pruning for the Responses API.
///
/// Unlike the Chat Completions path we never *remove* an item: the Responses API
/// rejects a `function_call` whose matching `function_call_output` is absent (and
/// reasoning items must keep their originating call). Instead we rewrite the
/// `output` text of every `function_call_output` in the frozen OLD region
/// (`input[..boundary]`) — pairing and ordering are untouched, so there is no
/// risk of a 400.
///
/// The boundary is the same monotone staircase as every other rail
/// ([`history_prune::prune_boundary`]), so the request prefix stays byte-stable
/// for up to a full stride and OpenAI's automatic prompt cache keeps hitting.
///
/// Shared with the WebSocket bridge (#440) so Codex/WS turns prune identically.
pub(super) fn prune_responses_input(doc: &mut Value) -> bool {
    let mode = lean_ctx::core::config::Config::load()
        .proxy
        .resolved_history_mode();
    let Some(input) = doc.get_mut("input").and_then(|i| i.as_array_mut()) else {
        return false;
    };
    let boundary = super::history_prune::prune_boundary(mode, input.len());
    if boundary == 0 {
        return false;
    }
    let tool_names = tool_kind::responses_tool_names(input);
    let mut modified = false;
    for item in input.iter_mut().take(boundary) {
        if compress_shared::classify_tool_kind(item) != ToolKind::ToolResult {
            continue;
        }
        let call_id = item.get("call_id").and_then(|v| v.as_str());
        // One tool-result block of the call with this id (`sticky_tools::block`).
        let _block = sticky_tools::block(call_id);
        let tool_name = call_id
            .and_then(|id| tool_names.get(id))
            .map(String::as_str);
        let kind = compress_shared::tool_result_kind(tool_name);
        if let Some(output) = item.get_mut("output") {
            modified |= prune_output_field(output, kind);
        }
    }
    modified
}

/// Apply [`history_prune::prune_output_text`] to a `function_call_output.output`,
/// handling both the JSON-string and array-of-content-parts shapes — the
/// pruning analogue of [`compress_output_field`].
fn prune_output_field(output: &mut Value, kind: ToolResultKind) -> bool {
    super::tool_output::prune_value(output, kind)
}

/// Compresses the `function_call_output.output` entries of a Responses-API body
/// in place, returning whether anything changed. Shared by the HTTP handler and
/// the WebSocket bridge (#440) so both paths get identical, safe savings.
///
/// The only token sink we shrink is each `function_call_output.output` — the
/// Responses-API analogue of a Chat Completions `role:"tool"` message. We never
/// remove or reorder `input` items: the Responses API rejects a `function_call`
/// whose matching `function_call_output` is absent (and reasoning items must keep
/// their originating call), so all token reclamation happens *in place* on the
/// output text. Cache-aware pruning of the frozen OLD region lives in
/// [`prune_responses_input`]; this pass compresses whatever recent outputs remain.
pub(super) fn compress_responses_input(doc: &mut Value) -> bool {
    // #481: recent-region live compression respects the global toggle. Old-region
    // pruning stays governed by `history_mode` in `prune_responses_input`.
    let cfg = lean_ctx::core::config::Config::load();
    if !cfg.proxy.live_compresses() || super::holdout::in_compression_control() {
        return false;
    }
    let mut modified = false;
    if let Some(input) = doc.get_mut("input").and_then(|i| i.as_array_mut()) {
        let tool_names = tool_kind::responses_tool_names(input);
        for item in input.iter_mut() {
            if compress_shared::classify_tool_kind(item) != ToolKind::ToolResult {
                continue;
            }
            let call_id = item.get("call_id").and_then(|v| v.as_str());
            // One tool-result block of the call with this id (`sticky_tools::block`).
            let _block = sticky_tools::block(call_id);
            let name = call_id
                .and_then(|id| tool_names.get(id))
                .map(String::as_str);
            // #481: per-tool exclusion (Serena default) — skip live compression
            // for excluded tools; history pruning above still applies.
            if name.is_some_and(|n| cfg.proxy.is_tool_live_compress_excluded(n)) {
                continue;
            }
            let kind = compress_shared::tool_result_kind(name);
            if let Some(output) = item.get_mut("output") {
                modified |= compress_output_field(output, name, kind);
            }
        }
    }
    modified
}

/// Compress a `function_call_output.output`. OpenAI sends this as a JSON string,
/// but the API also accepts an array of content parts (`input_text` blocks) for
/// tools returning richer data, so both shapes are handled.
///
/// A protected file/source read (resolved from the matching `function_call`
/// name) is left intact so a mid-refactor model never loses the body it edits.
fn compress_output_field(
    output: &mut Value,
    tool_name: Option<&str>,
    kind: ToolResultKind,
) -> bool {
    super::tool_output::compress_value(output, tool_name, kind)
}

#[cfg(test)]
mod tests {
    use super::super::compress::compress_tool_result;
    use super::*;

    // --- #1685: a subscription token is moved to the rail it authenticates on ---

    /// A shape-valid compact JWS. Only the three-segment `eyJ…` shape is read;
    /// nothing here is decoded, so the payload is arbitrary.
    const JWT_BEARER: &str = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0In0.c2lnbmF0dXJl";

    fn headers_with_auth(value: &str) -> axum::http::HeaderMap {
        let mut headers = axum::http::HeaderMap::new();
        headers.insert(
            axum::http::header::AUTHORIZATION,
            axum::http::HeaderValue::from_str(value).unwrap(),
        );
        headers
    }

    #[test]
    fn a_jwt_bearer_is_recognized_and_an_api_key_is_not() {
        assert!(is_chatgpt_subscription_bearer(&headers_with_auth(
            &format!("Bearer {JWT_BEARER}")
        )));
        assert!(
            !is_chatgpt_subscription_bearer(&headers_with_auth("Bearer sk-proj-abc123")),
            "an OpenAI platform key belongs on the /v1 rail"
        );
        assert!(
            !is_chatgpt_subscription_bearer(&headers_with_auth("Bearer opaque-gateway-token")),
            "a credential whose shape says nothing must not be moved"
        );
        assert!(
            !is_chatgpt_subscription_bearer(&axum::http::HeaderMap::new()),
            "no credential, no re-route"
        );
    }

    #[test]
    fn only_the_stock_openai_endpoint_is_eligible() {
        assert!(is_stock_openai_upstream("https://api.openai.com"));
        assert!(is_stock_openai_upstream("https://api.openai.com/"));
        // A configured gateway is an explicit choice and is never second-guessed.
        for configured in [
            "https://my-resource.openai.azure.com",
            "https://api.githubcopilot.com",
            "http://127.0.0.1:4000",
            "https://api.openai.com.evil.test",
        ] {
            assert!(
                !is_stock_openai_upstream(configured),
                "{configured} must keep its configured upstream"
            );
        }
    }

    fn rail_for(upstream: &str, auth: &str, uri: &str) -> Option<String> {
        // `chatgpt_rail_uri` reads only the upstream string and the headers, so
        // the decision is exercised without standing up a ProxyState.
        if !is_stock_openai_upstream(upstream) {
            return None;
        }
        if !is_chatgpt_subscription_bearer(&headers_with_auth(auth)) {
            return None;
        }
        let uri: axum::http::Uri = uri.parse().unwrap();
        let rest = uri
            .path()
            .strip_prefix("/v1/responses")
            .or_else(|| uri.path().strip_prefix("/responses"))?;
        let query = uri.query().map_or_else(String::new, |q| format!("?{q}"));
        Some(format!("/backend-api/codex/responses{rest}{query}"))
    }

    #[test]
    fn a_subscription_token_moves_to_the_chatgpt_rail() {
        assert_eq!(
            rail_for(
                "https://api.openai.com",
                &format!("Bearer {JWT_BEARER}"),
                "/v1/responses"
            )
            .as_deref(),
            Some("/backend-api/codex/responses")
        );
    }

    #[test]
    fn the_rail_move_preserves_subpaths_and_query() {
        assert_eq!(
            rail_for(
                "https://api.openai.com",
                &format!("Bearer {JWT_BEARER}"),
                "/v1/responses/resp_abc/cancel?stream=true"
            )
            .as_deref(),
            Some("/backend-api/codex/responses/resp_abc/cancel?stream=true")
        );
    }

    #[test]
    fn an_api_key_and_a_configured_gateway_both_stay_put() {
        assert_eq!(
            rail_for(
                "https://api.openai.com",
                "Bearer sk-proj-abc123",
                "/v1/responses"
            ),
            None,
            "an API key is exactly what the /v1 rail wants"
        );
        assert_eq!(
            rail_for(
                "https://my-resource.openai.azure.com",
                &format!("Bearer {JWT_BEARER}"),
                "/v1/responses"
            ),
            None,
            "a configured gateway may legitimately accept a JWT"
        );
    }

    // --- the same decision, driven through the real handler ---
    //
    // The tests above pin the rule; these pin that the rule is actually wired
    // into `handler` and that the request arrives at the other end on the path
    // it was moved to. Both point every upstream the handler could pick at a
    // local sink, so no test reaches the network.

    fn proxy_state(openai_upstream: &str, chatgpt_upstream: &str) -> ProxyState {
        use std::sync::Arc;
        let (_tx, rx) = tokio::sync::watch::channel(Arc::new(lean_ctx::core::config::Upstreams {
            anthropic: "https://api.anthropic.com".into(),
            openai: openai_upstream.into(),
            chatgpt: chatgpt_upstream.into(),
            gemini: "https://generativelanguage.googleapis.com".into(),
            providers: Vec::new(),
        }));
        // The sender is dropped here; a `watch::Receiver` keeps serving the last
        // value it saw, which is all `state.openai_upstream()` needs.
        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: rx,
            chatgpt_cookies: crate::chatgpt_cookies::shared_chatgpt_cloudflare_cookie_store(),
            mcp_servers: Arc::new(Vec::new()),
        }
    }

    /// A one-shot upstream that reports the request line it was given.
    async fn spawn_sink() -> (String, tokio::sync::oneshot::Receiver<String>) {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let (tx, rx) = tokio::sync::oneshot::channel();
        tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut buf = Vec::new();
            let mut chunk = [0_u8; 2048];
            while let Ok(n) = socket.read(&mut chunk).await {
                if n == 0 {
                    break;
                }
                buf.extend_from_slice(&chunk[..n]);
                if buf.windows(4).any(|w| w == b"\r\n\r\n") {
                    break;
                }
            }
            let first_line = String::from_utf8_lossy(&buf)
                .lines()
                .next()
                .unwrap_or_default()
                .to_string();
            let _ = tx.send(first_line);
            let _ = socket
                .write_all(
                    b"HTTP/1.1 200 OK\r\n\
                      content-type: application/json\r\n\
                      content-length: 2\r\n\
                      \r\n\
                      {}",
                )
                .await;
        });
        (format!("http://{addr}"), rx)
    }

    fn responses_request(auth: &str) -> Request<Body> {
        Request::builder()
            .method("POST")
            .uri("/v1/responses")
            .header(axum::http::header::AUTHORIZATION, auth)
            .header(axum::http::header::CONTENT_TYPE, "application/json")
            .body(Body::from(
                serde_json::to_vec(&serde_json::json!({"model": "gpt-5.5", "input": []})).unwrap(),
            ))
            .unwrap()
    }

    #[tokio::test]
    async fn handler_sends_a_subscription_token_to_the_chatgpt_backend() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        let (sink, seen) = spawn_sink().await;
        // Stock OpenAI upstream — the condition under test. It is never
        // contacted, because the request is moved to the ChatGPT sink.
        let state = proxy_state("https://api.openai.com", &sink);

        let _ = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            handler(
                State(state),
                responses_request(&format!("Bearer {JWT_BEARER}")),
            ),
        )
        .await
        .expect("the handler must answer from the local sink");

        let line = seen.await.unwrap();
        assert!(
            line.starts_with("POST /backend-api/codex/responses "),
            "a subscription token must arrive on the ChatGPT rail, got: {line}"
        );
    }

    #[tokio::test]
    async fn handler_leaves_a_configured_gateway_on_the_platform_path() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        let (sink, seen) = spawn_sink().await;
        // The same JWT, but the operator configured this upstream explicitly.
        // It may well accept a JWT, so the path must be left exactly as sent.
        let state = proxy_state(&sink, "https://chatgpt.com");

        let _ = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            handler(
                State(state),
                responses_request(&format!("Bearer {JWT_BEARER}")),
            ),
        )
        .await
        .expect("the handler must answer from the local sink");

        let line = seen.await.unwrap();
        assert!(
            line.starts_with("POST /v1/responses "),
            "a configured gateway must keep the path it was sent, got: {line}"
        );
    }

    /// A long `git status` is a known-compressible fixture: `has_structural_output`
    /// is false for it, so it flows through the git-status pattern compressor.
    fn long_git_status() -> String {
        let mut s = String::from(
            "$ git status\nOn branch main\nYour branch is up to date with 'origin/main'.\n\nChanges not staged for commit:\n  (use \"git add <file>...\" to update what will be committed)\n",
        );
        for i in 0..80 {
            s.push_str(&format!("\tmodified:   src/module_{i}/file_{i}.rs\n"));
        }
        s.push_str("\nno changes added to commit (use \"git add\" and/or \"git commit -a\")\n");
        s
    }

    fn big_prose() -> String {
        let p = "You are a careful, senior software engineer. You always explain your \
                 reasoning before making changes, you prefer small reviewable diffs, and \
                 you never introduce mock data or placeholders into production code. ";
        [p; 6].join("\n")
    }

    fn long_tool_json() -> String {
        let rows = (0..32)
            .map(|i| {
                serde_json::json!({
                    "path": format!("/Users/alex/work/app/src/module_{i}.rs"),
                    "regex": r"src/[a-z_]+\.rs:\d+",
                    "error": format!("error[E0{i:03}]: expected exact diagnostic text"),
                })
            })
            .collect::<Vec<_>>();
        serde_json::to_string(&serde_json::json!({ "results": rows })).unwrap()
    }

    #[test]
    fn shell_json_envelope_text_is_compressed() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        let raw = long_git_status();
        let expected = compress_tool_result(&raw, Some("Bash"));
        let envelope = serde_json::to_string(&serde_json::json!({
            "content": [{"type": "text", "text": raw}],
            "isError": false,
        }))
        .unwrap();

        let body = serde_json::json!({
            "model": "gpt-5",
            "input": [
                {"type": "function_call", "call_id": "call_1", "name": "Bash", "arguments": "{}"},
                {"type": "function_call_output", "call_id": "call_1", "output": envelope}
            ]
        });
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body, bytes.len());

        assert!(comp < orig);
        let parsed: Value = serde_json::from_slice(&out).unwrap();
        let output = parsed["input"][1]["output"].as_str().unwrap();
        let envelope: Value = serde_json::from_str(output).unwrap();
        assert_eq!(envelope["content"][0]["text"].as_str().unwrap(), expected);
    }

    #[test]
    fn shell_json_envelope_non_text_field_is_compressed() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        let raw = long_git_status();
        let expected = compress_tool_result(&raw, Some("Bash"));
        // Big shell output in a non-"text" field: the Shell/Search all-strings
        // rewrite must still reach it, while small scalar fields stay intact.
        let envelope = serde_json::to_string(&serde_json::json!({
            "stdout": raw,
            "exit_code": 0,
        }))
        .unwrap();

        let body = serde_json::json!({
            "model": "gpt-5",
            "input": [
                {"type": "function_call", "call_id": "call_1", "name": "Bash", "arguments": "{}"},
                {"type": "function_call_output", "call_id": "call_1", "output": envelope}
            ]
        });
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body, bytes.len());

        assert!(comp < orig, "non-text shell field should be compressed");
        let parsed: Value = serde_json::from_slice(&out).unwrap();
        let output = parsed["input"][1]["output"].as_str().unwrap();
        let envelope: Value = serde_json::from_str(output).unwrap();
        assert_eq!(envelope["stdout"].as_str().unwrap(), expected);
        assert_eq!(envelope["exit_code"].as_i64().unwrap(), 0);
    }

    #[test]
    fn old_shell_json_envelope_text_is_pruned() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        let raw = long_git_status();
        let envelope = serde_json::to_string(&serde_json::json!({
            "content": [{"type": "text", "text": raw}],
            "isError": false,
        }))
        .unwrap();
        let mut output = Value::String(envelope);

        assert!(prune_output_field(&mut output, ToolResultKind::Shell));
        let envelope: Value = serde_json::from_str(output.as_str().unwrap()).unwrap();
        assert!(
            envelope["content"][0]["text"].as_str().unwrap().len() < raw.len(),
            "nested text payload should be pruned"
        );
    }

    #[test]
    fn string_output_mirrors_engine_and_shrinks() {
        // tee path depends on the data dir; serialize env access so a parallel
        // test never swaps LEAN_CTX_DATA_DIR between the two compressions (#498).
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        let raw = long_git_status();
        let expected = compress_tool_result(&raw, None);
        assert!(
            expected.len() < raw.len(),
            "fixture must be compressible by the shared engine"
        );

        let body = serde_json::json!({
            "model": "gpt-5",
            "input": [
                {"type": "function_call_output", "call_id": "call_1", "output": raw}
            ]
        });
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body, bytes.len());

        assert!(comp < orig, "compressed body must be smaller");
        let parsed: Value = serde_json::from_slice(&out).unwrap();
        assert_eq!(
            parsed["input"][0]["output"].as_str().unwrap(),
            expected,
            "output must be exactly what the shared compressor produces"
        );
    }

    #[test]
    fn array_output_text_is_compressed() {
        // tee path depends on the data dir; serialize env access so a parallel
        // test never swaps LEAN_CTX_DATA_DIR between the two compressions (#498).
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        let raw = long_git_status();
        let expected = compress_tool_result(&raw, None);

        let body = serde_json::json!({
            "input": [
                {
                    "type": "function_call_output",
                    "call_id": "call_1",
                    "output": [{"type": "input_text", "text": raw}]
                }
            ]
        });
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body, bytes.len());

        assert!(comp < orig);
        let parsed: Value = serde_json::from_slice(&out).unwrap();
        assert_eq!(
            parsed["input"][0]["output"][0]["text"].as_str().unwrap(),
            expected
        );
    }

    #[test]
    fn non_tool_output_items_are_untouched() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        lean_ctx::core::config::Config::update_global(|c| {
            c.proxy.proxy_mode = Some("cache".into());
        })
        .unwrap();
        let body = serde_json::json!({
            "input": [
                {"type": "message", "role": "user", "content": long_git_status()},
                {"type": "function_call", "call_id": "c", "name": "x", "arguments": "{}"}
            ]
        });
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body.clone(), bytes.len());

        assert_eq!(comp, orig, "no function_call_output → passthrough");
        let reparsed: Value = serde_json::from_slice(&out).unwrap();
        assert_eq!(reparsed, body);
    }

    #[test]
    fn plain_string_input_passthrough() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        lean_ctx::core::config::Config::update_global(|c| {
            c.proxy.proxy_mode = Some("cache".into());
        })
        .unwrap();
        let body = serde_json::json!({"model": "gpt-5", "input": "hello world"});
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body.clone(), bytes.len());
        assert_eq!(comp, orig);
        let reparsed: Value = serde_json::from_slice(&out).unwrap();
        assert_eq!(reparsed, body);
    }

    #[test]
    fn no_input_field_passthrough() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        let body = serde_json::json!({"model": "gpt-5", "previous_response_id": "resp_abc"});
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body.clone(), bytes.len());
        assert_eq!(comp, orig);
        let reparsed: Value = serde_json::from_slice(&out).unwrap();
        assert_eq!(reparsed, body);
    }

    #[test]
    fn chatgpt_responses_eval_fixture_keeps_exact_payloads_and_pairing() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        lean_ctx::core::config::Config::update_global(|c| {
            c.proxy.role_aggressiveness.user = Some(0.8);
            c.proxy.proxy_mode = Some("cache".into());
        })
        .unwrap();

        let command_input = "$ cargo test --lib proxy::openai_responses\nerror[E0425]: cannot find value `x` in this scope\nsrc/proxy/openai_responses.rs:12:9";
        let body = serde_json::json!({"model": "gpt-5", "input": command_input});
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body.clone(), bytes.len());
        assert_eq!(comp, orig, "top-level exact command string must stay raw");
        assert_eq!(serde_json::from_slice::<Value>(&out).unwrap(), body);

        let old_json = long_tool_json();
        let recent_json = long_tool_json();
        let old_shell = long_git_status();
        let input_text_block = "```rust\nfn main() { panic!(\"exact\"); }\n```\nRegex: src/[a-z_]+\\.rs:\\d+\nPath: /Users/alex/work/app/src/main.rs\nError: error[E0425]: cannot find value `x` in this scope";
        let mut input = vec![
            serde_json::json!({"type": "reasoning", "id": "rs_1", "summary": []}),
            serde_json::json!({"type": "function_call", "call_id": "json_old", "name": "submit_tool_json", "arguments": "{\"strict\":true}"}),
            serde_json::json!({"type": "function_call_output", "call_id": "json_old", "output": old_json}),
            serde_json::json!({"type": "function_call", "call_id": "shell_old", "name": "Bash", "arguments": "{\"cmd\":\"git status\"}"}),
            serde_json::json!({"type": "function_call_output", "call_id": "shell_old", "output": old_shell}),
            serde_json::json!({"type": "message", "role": "user", "content": [{"type": "input_text", "text": input_text_block}]}),
        ];
        while input.len() < 22 {
            input.push(serde_json::json!({
                "type": "message",
                "role": "user",
                "content": format!("filler {}", input.len()),
            }));
        }
        input.push(serde_json::json!({"type": "function_call", "call_id": "json_recent", "name": "submit_tool_json", "arguments": "{\"strict\":true}"}));
        input.push(serde_json::json!({"type": "function_call_output", "call_id": "json_recent", "output": recent_json}));

        let body = serde_json::json!({"model": "gpt-5", "input": input});
        let item_count = body["input"].as_array().unwrap().len();
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body, bytes.len());
        assert!(comp < orig, "old shell output should still provide savings");

        let parsed: Value = serde_json::from_slice(&out).unwrap();
        let input = parsed["input"].as_array().unwrap();
        assert_eq!(input.len(), item_count, "no Responses item may be dropped");
        assert_eq!(input[0]["type"], "reasoning");
        assert_eq!(input[1]["type"], "function_call");
        assert_eq!(input[2]["type"], "function_call_output");
        assert_eq!(input[2]["output"].as_str().unwrap(), old_json);
        assert_ne!(input[4]["output"].as_str().unwrap(), old_shell);
        assert_eq!(
            input[5]["content"][0]["text"].as_str().unwrap(),
            input_text_block
        );
        assert_eq!(input[22]["type"], "function_call");
        assert_eq!(input[23]["type"], "function_call_output");
        assert_eq!(input[23]["output"].as_str().unwrap(), recent_json);
        assert_eq!(input[1]["call_id"], input[2]["call_id"]);
        assert_eq!(input[22]["call_id"], input[23]["call_id"]);
    }

    #[test]
    fn responses_instructions_prose_compressed_and_assistant_untouched() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        lean_ctx::core::config::Config::update_global(|c| {
            c.proxy.role_aggressiveness.system = Some(0.6);
        })
        .unwrap();

        let prose = big_prose();
        let body = serde_json::json!({
            "model": "gpt-5",
            "instructions": prose,
            "input": [
                {"type": "message", "role": "user", "content": "hi"},
                {"type": "message", "role": "assistant", "content": prose},
            ]
        });
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body, bytes.len());
        assert!(comp < orig, "enabled instructions prose must save bytes");
        let parsed: Value = serde_json::from_slice(&out).unwrap();

        assert!(
            parsed["instructions"].as_str().unwrap().len() < prose.len(),
            "Responses instructions must be compressed when enabled"
        );
        assert_eq!(
            parsed["input"][1]["content"].as_str().unwrap(),
            prose,
            "assistant turns must pass through verbatim (#710)"
        );
    }

    #[test]
    fn responses_user_prose_compressed_only_in_frozen_region() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        lean_ctx::core::config::Config::update_global(|c| {
            c.proxy.role_aggressiveness.user = Some(0.7);
            c.proxy.proxy_mode = Some("cache".into());
        })
        .unwrap();

        let prose = big_prose();
        // 30 messages -> cache-aware boundary = ((30 - 8) / 16) * 16 = 16.
        let mut input = Vec::new();
        for i in 0..30 {
            let role = if i % 2 == 0 { "user" } else { "assistant" };
            input.push(serde_json::json!({
                "type": "message",
                "role": role,
                "content": prose,
            }));
        }
        let body = serde_json::json!({"model": "gpt-5", "input": input});
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body, bytes.len());
        assert!(comp < orig, "old user prose must save bytes");
        let parsed: Value = serde_json::from_slice(&out).unwrap();

        let frozen_user = parsed["input"][0]["content"].as_str().unwrap();
        assert!(
            frozen_user.len() < prose.len(),
            "old user prose should compress"
        );
        assert_eq!(
            parsed["input"][1]["content"].as_str().unwrap(),
            prose,
            "assistant prose must stay verbatim"
        );
        assert_eq!(
            parsed["input"][16]["content"].as_str().unwrap(),
            prose,
            "live-tail user prose must stay verbatim"
        );
    }

    #[test]
    fn responses_prose_compression_is_deterministic() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        lean_ctx::core::config::Config::update_global(|c| {
            c.proxy.role_aggressiveness.system = Some(0.6);
        })
        .unwrap();

        let prose = big_prose();
        let mk = || {
            serde_json::json!({
                "model": "gpt-5",
                "instructions": prose,
                "input": [{"type": "message", "role": "user", "content": "hi"}],
            })
        };
        let (a, b) = (mk(), mk());
        let (la, lb) = (
            serde_json::to_vec(&a).unwrap().len(),
            serde_json::to_vec(&b).unwrap().len(),
        );
        assert_eq!(
            compress_request_body(a, la).0,
            compress_request_body(b, lb).0,
            "identical input must yield byte-identical output (#498)"
        );
    }

    #[test]
    fn short_output_unchanged() {
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        let body = serde_json::json!({
            "input": [
                {"type": "function_call_output", "call_id": "c", "output": "ok"}
            ]
        });
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body.clone(), bytes.len());
        assert_eq!(comp, orig);
        let reparsed: Value = serde_json::from_slice(&out).unwrap();
        assert_eq!(reparsed, body);
    }

    /// `pairs` Responses turns: each is a `function_call` + its matching
    /// `function_call_output` carrying a long file read.
    fn responses_read_turns(pairs: usize) -> Vec<Value> {
        let code = (0..40)
            .map(|i| format!("    let v{i} = compute_{i}(ctx, opts);"))
            .collect::<Vec<_>>()
            .join("\n");
        let mut input = Vec::new();
        for t in 0..pairs {
            input.push(serde_json::json!({
                "type": "function_call", "call_id": format!("c{t}"),
                "name": "read_file", "arguments": "{}"
            }));
            input.push(serde_json::json!({
                "type": "function_call_output", "call_id": format!("c{t}"),
                "output": format!("{code}\n// turn {t}")
            }));
        }
        input
    }

    #[test]
    fn cache_aware_prune_stubs_old_reads_keeps_recent_and_pairing() {
        // Default (isolated) config = cache-aware history mode.
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        // 14 pairs = 28 items → staircase boundary 16.
        let body = serde_json::json!({"model": "gpt-5", "input": responses_read_turns(14)});
        let item_count = body["input"].as_array().unwrap().len();
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, orig, comp) = compress_request_body(body, bytes.len());
        assert!(comp < orig, "old reads must be pruned for savings");

        let parsed: Value = serde_json::from_slice(&out).unwrap();
        let input = parsed["input"].as_array().unwrap();
        // Pairing + ordering preserved: not a single item dropped or moved.
        assert_eq!(input.len(), item_count, "no items may be removed (pairing)");
        for (i, item) in input.iter().enumerate() {
            let expect = if i.is_multiple_of(2) {
                "function_call"
            } else {
                "function_call_output"
            };
            assert_eq!(item["type"], expect, "item {i} type/order changed");
        }
        // An OLD file read (output index 1, before boundary 16) is stubbed.
        let old = input[1]["output"].as_str().unwrap();
        assert!(
            old.contains("Re-read the file"),
            "old read should be stubbed, got: {old}"
        );
        // A RECENT file read (output index 27, after the boundary) keeps its body.
        let recent = input[27]["output"].as_str().unwrap();
        assert!(
            recent.contains("v39"),
            "recent read must be protected, got: {recent}"
        );
    }

    #[test]
    fn responses_compression_is_deterministic() {
        // #498: the same request must compress to byte-identical output so the
        // provider's prompt cache (and our regression diffs) stay stable.
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        let mk = || serde_json::json!({"model": "gpt-5", "input": responses_read_turns(14)});
        let (a, b) = (mk(), mk());
        let (la, lb) = (
            serde_json::to_vec(&a).unwrap().len(),
            serde_json::to_vec(&b).unwrap().len(),
        );
        let (out_a, _, _) = compress_request_body(a, la);
        let (out_b, _, _) = compress_request_body(b, lb);
        assert_eq!(out_a, out_b, "identical input must yield identical bytes");
    }

    #[test]
    fn cache_aware_responses_prefix_is_byte_stable_across_turns() {
        // THE cache invariant for the Responses rail: as `input` grows turn by
        // turn, every item before an already-passed boundary must stay
        // byte-identical, or OpenAI's automatic prompt cache stops hitting.
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        let _community = lean_ctx::core::policy::runtime::TestPolicyOverride::set(None);
        let mut prev: Vec<String> = Vec::new();
        let mut prev_boundary = 0;
        for pairs in 1..=20 {
            let input = responses_read_turns(pairs);
            let len = input.len();
            let body = serde_json::json!({"model": "gpt-5", "input": input});
            let bytes = serde_json::to_vec(&body).unwrap();
            let (out, _, _) = compress_request_body(body, bytes.len());
            let parsed: Value = serde_json::from_slice(&out).unwrap();
            let items: Vec<String> = parsed["input"]
                .as_array()
                .unwrap()
                .iter()
                .map(Value::to_string)
                .collect();
            for i in 0..prev_boundary {
                assert_eq!(
                    prev[i], items[i],
                    "Responses item {i} changed at turn {pairs} — prompt cache prefix broken"
                );
            }
            prev = items;
            prev_boundary = crate::history_prune::prune_boundary(
                lean_ctx::core::config::HistoryMode::CacheAware,
                len,
            );
        }
    }

    #[test]
    fn effort_control_sets_nested_reasoning_effort() {
        // #834 end-to-end through the Responses request path.
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();
        lean_ctx::test_env::remove_var("LEAN_CTX_PROXY_EFFORT");
        lean_ctx::core::config::Config::update_global(|c| {
            c.proxy.effort = Some("low".into());
        })
        .unwrap();
        let body = serde_json::json!({"model": "gpt-5.5", "input": []});
        let bytes = serde_json::to_vec(&body).unwrap();
        let (out, _o, _c) = compress_request_body(body, bytes.len());
        assert_eq!(
            serde_json::from_slice::<Value>(&out).unwrap()["reasoning"]["effort"],
            "low"
        );
    }
}