lean_ctx/proxy/
openai_responses.rs

1use axum::{
2    body::Body,
3    extract::State,
4    http::{Request, StatusCode},
5    response::Response,
6};
7use serde_json::Value;
8
9use super::ProxyState;
10use super::compress::compress_tool_result;
11use super::forward;
12use super::tool_kind::{self, ToolResultKind, should_protect};
13
14/// Proxy handler for OpenAI's Responses API (`POST /v1/responses`).
15///
16/// The Responses API superseded Chat Completions for clients such as opencode
17/// and the OpenAI Agents SDK. Its conversation turns live in `input` rather than
18/// `messages`, so the Chat Completions handler never saw — and never compressed —
19/// them. This handler reuses the same upstream, auth and streaming path but
20/// understands the Responses-API request shape.
21///
22/// Retrieve / cancel / delete / input_items sub-paths
23/// (`/v1/responses/{id}/...`) are routed here as well and pass through untouched:
24/// they carry no `input` array, so `compress_request_body` is a no-op for them.
25///
26/// Handles `POST /v1/responses` (and the bare `/responses`) over HTTP/SSE.
27pub async fn handler(
28    State(state): State<ProxyState>,
29    req: Request<Body>,
30) -> Result<Response, StatusCode> {
31    let upstream = state.openai_upstream();
32    forward::forward_request(
33        State(state),
34        req,
35        &upstream,
36        "/v1/responses",
37        compress_request_body,
38        "OpenAI",
39        &[],
40    )
41    .await
42}
43
44/// Handles the WebSocket Responses transport on `GET /v1/responses`.
45///
46/// Codex (and the OpenAI SDK) default to `ws://…/responses` with one
47/// `response.create` event per turn. Bridging the upgrade here lets the proxy be
48/// a drop-in for Codex without forcing `supports_websockets = false` (#440); the
49/// actual WS↔HTTP/SSE bridging lives in `openai_responses_ws`.
50pub async fn ws_handler(
51    State(state): State<ProxyState>,
52    headers: axum::http::HeaderMap,
53    ws: axum::extract::ws::WebSocketUpgrade,
54) -> Response {
55    super::openai_responses_ws::upgrade(state, ws, &headers)
56}
57
58fn compress_request_body(parsed: Value, original_size: usize) -> (Vec<u8>, usize, usize) {
59    let mut doc = parsed;
60    let cfg = crate::core::config::Config::load();
61    // #493: in-band CCR expansion (opt-in). Splice any <lc_expand:HASH> the model
62    // echoed back into the verbatim original from the local tee store. A strict
63    // no-op when no marker is present (byte-identical body → cache-safe). Runs
64    // before the meter-only short-circuit so an explicit expand request is
65    // honored even when the proxy is otherwise byte-passthrough.
66    let mut modified = false;
67    if cfg.proxy.ccr_inband_enabled() {
68        modified |= super::ccr::splice_inband_in_place(&mut doc);
69    }
70    // #834: cache-safe cross-provider effort control. Default off → no-op. The
71    // value is a constant, so it never perturbs the prompt-cache prefix; it sets
72    // `reasoning.effort` only on reasoning models and never overrides a
73    // client-set value.
74    if let Some(effort) = cfg.proxy.resolved_effort() {
75        modified |= super::effort::apply_openai_responses(&mut doc, effort);
76    }
77    // Meter-only (#481): live compression off and history pruning off → forward
78    // the body unchanged while upstream usage metering still runs. A pending
79    // in-band splice (`modified`) opts out: the body did change this turn.
80    if !cfg.proxy.live_compresses()
81        && cfg.proxy.resolved_history_mode() == crate::core::config::HistoryMode::Off
82        && !modified
83    {
84        let out = serde_json::to_vec(&doc).unwrap_or_default();
85        return (out, original_size, original_size);
86    }
87    // Two-stage, like the Chat Completions path: (1) cache-aware prune of the
88    // frozen OLD region — old file reads collapse to re-read stubs, old logs
89    // head/tail summarize — then (2) compress whatever recent outputs remain.
90    // Stage 1 runs first so a stubbed old output isn't needlessly re-compressed.
91    modified |= prune_responses_input(&mut doc);
92    modified |= compress_responses_input(&mut doc);
93    let out = serde_json::to_vec(&doc).unwrap_or_default();
94    let compressed_size = if modified { out.len() } else { original_size };
95    (out, original_size, compressed_size)
96}
97
98/// Cache-aware history pruning for the Responses API.
99///
100/// Unlike the Chat Completions path we never *remove* an item: the Responses API
101/// rejects a `function_call` whose matching `function_call_output` is absent (and
102/// reasoning items must keep their originating call). Instead we rewrite the
103/// `output` text of every `function_call_output` in the frozen OLD region
104/// (`input[..boundary]`) — pairing and ordering are untouched, so there is no
105/// risk of a 400.
106///
107/// The boundary is the same monotone staircase as every other rail
108/// ([`history_prune::prune_boundary`]), so the request prefix stays byte-stable
109/// for up to a full stride and OpenAI's automatic prompt cache keeps hitting.
110///
111/// Shared with the WebSocket bridge (#440) so Codex/WS turns prune identically.
112pub(super) fn prune_responses_input(doc: &mut Value) -> bool {
113    let mode = crate::core::config::Config::load()
114        .proxy
115        .resolved_history_mode();
116    let Some(input) = doc.get_mut("input").and_then(|i| i.as_array_mut()) else {
117        return false;
118    };
119    let boundary = super::history_prune::prune_boundary(mode, input.len());
120    if boundary == 0 {
121        return false;
122    }
123    let tool_names = tool_kind::responses_tool_names(input);
124    let mut modified = false;
125    for item in input.iter_mut().take(boundary) {
126        if item.get("type").and_then(|t| t.as_str()) != Some("function_call_output") {
127            continue;
128        }
129        let kind = item
130            .get("call_id")
131            .and_then(|v| v.as_str())
132            .and_then(|id| tool_names.get(id))
133            .map_or(ToolResultKind::Other, |n| tool_kind::classify_tool_name(n));
134        if let Some(output) = item.get_mut("output") {
135            modified |= prune_output_field(output, kind);
136        }
137    }
138    modified
139}
140
141/// Apply [`history_prune::prune_output_text`] to a `function_call_output.output`,
142/// handling both the JSON-string and array-of-content-parts shapes — the
143/// pruning analogue of [`compress_output_field`].
144fn prune_output_field(output: &mut Value, kind: ToolResultKind) -> bool {
145    match output {
146        Value::String(s) => match super::history_prune::prune_output_text(s, kind) {
147            Some(pruned) => {
148                *s = pruned;
149                true
150            }
151            None => false,
152        },
153        Value::Array(parts) => {
154            let mut changed = false;
155            for part in parts.iter_mut() {
156                if let Some(Value::String(text)) = part.get_mut("text")
157                    && let Some(pruned) = super::history_prune::prune_output_text(text, kind)
158                {
159                    *text = pruned;
160                    changed = true;
161                }
162            }
163            changed
164        }
165        _ => false,
166    }
167}
168
169/// Compresses the `function_call_output.output` entries of a Responses-API body
170/// in place, returning whether anything changed. Shared by the HTTP handler and
171/// the WebSocket bridge (#440) so both paths get identical, safe savings.
172///
173/// The only token sink we shrink is each `function_call_output.output` — the
174/// Responses-API analogue of a Chat Completions `role:"tool"` message. We never
175/// remove or reorder `input` items: the Responses API rejects a `function_call`
176/// whose matching `function_call_output` is absent (and reasoning items must keep
177/// their originating call), so all token reclamation happens *in place* on the
178/// output text. Cache-aware pruning of the frozen OLD region lives in
179/// [`prune_responses_input`]; this pass compresses whatever recent outputs remain.
180pub(super) fn compress_responses_input(doc: &mut Value) -> bool {
181    // #481: recent-region live compression respects the global toggle. Old-region
182    // pruning stays governed by `history_mode` in `prune_responses_input`.
183    let cfg = crate::core::config::Config::load();
184    if !cfg.proxy.live_compresses() {
185        return false;
186    }
187    let mut modified = false;
188    if let Some(input) = doc.get_mut("input").and_then(|i| i.as_array_mut()) {
189        let tool_names = tool_kind::responses_tool_names(input);
190        for item in input.iter_mut() {
191            if item.get("type").and_then(|t| t.as_str()) != Some("function_call_output") {
192                continue;
193            }
194            let name = item
195                .get("call_id")
196                .and_then(|v| v.as_str())
197                .and_then(|id| tool_names.get(id))
198                .map(String::as_str);
199            // #481: per-tool exclusion (Serena default) — skip live compression
200            // for excluded tools; history pruning above still applies.
201            if name.is_some_and(|n| cfg.proxy.is_tool_live_compress_excluded(n)) {
202                continue;
203            }
204            let kind = name.map_or(ToolResultKind::Other, tool_kind::classify_tool_name);
205            if let Some(output) = item.get_mut("output") {
206                modified |= compress_output_field(output, name, kind);
207            }
208        }
209    }
210    modified
211}
212
213/// Compress a `function_call_output.output`. OpenAI sends this as a JSON string,
214/// but the API also accepts an array of content parts (`input_text` blocks) for
215/// tools returning richer data, so both shapes are handled.
216///
217/// A protected file/source read (resolved from the matching `function_call`
218/// name) is left intact so a mid-refactor model never loses the body it edits.
219fn compress_output_field(
220    output: &mut Value,
221    tool_name: Option<&str>,
222    kind: ToolResultKind,
223) -> bool {
224    match output {
225        Value::String(s) => {
226            if should_protect(kind, s) {
227                return false;
228            }
229            let compressed = compress_tool_result(s, tool_name);
230            if compressed.len() < s.len() {
231                *s = compressed;
232                return true;
233            }
234            false
235        }
236        Value::Array(parts) => {
237            let mut changed = false;
238            for part in parts.iter_mut() {
239                if let Some(Value::String(text)) = part.get_mut("text") {
240                    if should_protect(kind, text) {
241                        continue;
242                    }
243                    let compressed = compress_tool_result(text, tool_name);
244                    if compressed.len() < text.len() {
245                        *text = compressed;
246                        changed = true;
247                    }
248                }
249            }
250            changed
251        }
252        _ => false,
253    }
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259
260    /// A long `git status` is a known-compressible fixture: `has_structural_output`
261    /// is false for it, so it flows through the git-status pattern compressor.
262    fn long_git_status() -> String {
263        let mut s = String::from(
264            "$ 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",
265        );
266        for i in 0..80 {
267            s.push_str(&format!("\tmodified:   src/module_{i}/file_{i}.rs\n"));
268        }
269        s.push_str("\nno changes added to commit (use \"git add\" and/or \"git commit -a\")\n");
270        s
271    }
272
273    #[test]
274    fn string_output_mirrors_engine_and_shrinks() {
275        // tee path depends on the data dir; serialize env access so a parallel
276        // test never swaps LEAN_CTX_DATA_DIR between the two compressions (#498).
277        let _lock = crate::core::data_dir::test_env_lock();
278        let raw = long_git_status();
279        let expected = compress_tool_result(&raw, None);
280        assert!(
281            expected.len() < raw.len(),
282            "fixture must be compressible by the shared engine"
283        );
284
285        let body = serde_json::json!({
286            "model": "gpt-5",
287            "input": [
288                {"type": "function_call_output", "call_id": "call_1", "output": raw}
289            ]
290        });
291        let bytes = serde_json::to_vec(&body).unwrap();
292        let (out, orig, comp) = compress_request_body(body, bytes.len());
293
294        assert!(comp < orig, "compressed body must be smaller");
295        let parsed: Value = serde_json::from_slice(&out).unwrap();
296        assert_eq!(
297            parsed["input"][0]["output"].as_str().unwrap(),
298            expected,
299            "output must be exactly what the shared compressor produces"
300        );
301    }
302
303    #[test]
304    fn array_output_text_is_compressed() {
305        // tee path depends on the data dir; serialize env access so a parallel
306        // test never swaps LEAN_CTX_DATA_DIR between the two compressions (#498).
307        let _lock = crate::core::data_dir::test_env_lock();
308        let raw = long_git_status();
309        let expected = compress_tool_result(&raw, None);
310
311        let body = serde_json::json!({
312            "input": [
313                {
314                    "type": "function_call_output",
315                    "call_id": "call_1",
316                    "output": [{"type": "input_text", "text": raw}]
317                }
318            ]
319        });
320        let bytes = serde_json::to_vec(&body).unwrap();
321        let (out, orig, comp) = compress_request_body(body, bytes.len());
322
323        assert!(comp < orig);
324        let parsed: Value = serde_json::from_slice(&out).unwrap();
325        assert_eq!(
326            parsed["input"][0]["output"][0]["text"].as_str().unwrap(),
327            expected
328        );
329    }
330
331    #[test]
332    fn non_tool_output_items_are_untouched() {
333        let body = serde_json::json!({
334            "input": [
335                {"type": "message", "role": "user", "content": long_git_status()},
336                {"type": "function_call", "call_id": "c", "name": "x", "arguments": "{}"}
337            ]
338        });
339        let bytes = serde_json::to_vec(&body).unwrap();
340        let (out, orig, comp) = compress_request_body(body.clone(), bytes.len());
341
342        assert_eq!(comp, orig, "no function_call_output → passthrough");
343        let reparsed: Value = serde_json::from_slice(&out).unwrap();
344        assert_eq!(reparsed, body);
345    }
346
347    #[test]
348    fn plain_string_input_passthrough() {
349        let body = serde_json::json!({"model": "gpt-5", "input": "hello world"});
350        let bytes = serde_json::to_vec(&body).unwrap();
351        let (out, orig, comp) = compress_request_body(body.clone(), bytes.len());
352        assert_eq!(comp, orig);
353        let reparsed: Value = serde_json::from_slice(&out).unwrap();
354        assert_eq!(reparsed, body);
355    }
356
357    #[test]
358    fn no_input_field_passthrough() {
359        let body = serde_json::json!({"model": "gpt-5", "previous_response_id": "resp_abc"});
360        let bytes = serde_json::to_vec(&body).unwrap();
361        let (out, orig, comp) = compress_request_body(body.clone(), bytes.len());
362        assert_eq!(comp, orig);
363        let reparsed: Value = serde_json::from_slice(&out).unwrap();
364        assert_eq!(reparsed, body);
365    }
366
367    #[test]
368    fn short_output_unchanged() {
369        let body = serde_json::json!({
370            "input": [
371                {"type": "function_call_output", "call_id": "c", "output": "ok"}
372            ]
373        });
374        let bytes = serde_json::to_vec(&body).unwrap();
375        let (out, orig, comp) = compress_request_body(body.clone(), bytes.len());
376        assert_eq!(comp, orig);
377        let reparsed: Value = serde_json::from_slice(&out).unwrap();
378        assert_eq!(reparsed, body);
379    }
380
381    /// `pairs` Responses turns: each is a `function_call` + its matching
382    /// `function_call_output` carrying a long file read.
383    fn responses_read_turns(pairs: usize) -> Vec<Value> {
384        let code = (0..40)
385            .map(|i| format!("    let v{i} = compute_{i}(ctx, opts);"))
386            .collect::<Vec<_>>()
387            .join("\n");
388        let mut input = Vec::new();
389        for t in 0..pairs {
390            input.push(serde_json::json!({
391                "type": "function_call", "call_id": format!("c{t}"),
392                "name": "read_file", "arguments": "{}"
393            }));
394            input.push(serde_json::json!({
395                "type": "function_call_output", "call_id": format!("c{t}"),
396                "output": format!("{code}\n// turn {t}")
397            }));
398        }
399        input
400    }
401
402    #[test]
403    fn cache_aware_prune_stubs_old_reads_keeps_recent_and_pairing() {
404        // Default (isolated) config = cache-aware history mode.
405        let _iso = crate::core::data_dir::isolated_data_dir();
406        // 14 pairs = 28 items → staircase boundary 16.
407        let body = serde_json::json!({"model": "gpt-5", "input": responses_read_turns(14)});
408        let item_count = body["input"].as_array().unwrap().len();
409        let bytes = serde_json::to_vec(&body).unwrap();
410        let (out, orig, comp) = compress_request_body(body, bytes.len());
411        assert!(comp < orig, "old reads must be pruned for savings");
412
413        let parsed: Value = serde_json::from_slice(&out).unwrap();
414        let input = parsed["input"].as_array().unwrap();
415        // Pairing + ordering preserved: not a single item dropped or moved.
416        assert_eq!(input.len(), item_count, "no items may be removed (pairing)");
417        for (i, item) in input.iter().enumerate() {
418            let expect = if i.is_multiple_of(2) {
419                "function_call"
420            } else {
421                "function_call_output"
422            };
423            assert_eq!(item["type"], expect, "item {i} type/order changed");
424        }
425        // An OLD file read (output index 1, before boundary 16) is stubbed.
426        let old = input[1]["output"].as_str().unwrap();
427        assert!(
428            old.contains("Re-read the file"),
429            "old read should be stubbed, got: {old}"
430        );
431        // A RECENT file read (output index 27, after the boundary) keeps its body.
432        let recent = input[27]["output"].as_str().unwrap();
433        assert!(
434            recent.contains("v39"),
435            "recent read must be protected, got: {recent}"
436        );
437    }
438
439    #[test]
440    fn responses_compression_is_deterministic() {
441        // #498: the same request must compress to byte-identical output so the
442        // provider's prompt cache (and our regression diffs) stay stable.
443        let _iso = crate::core::data_dir::isolated_data_dir();
444        let mk = || serde_json::json!({"model": "gpt-5", "input": responses_read_turns(14)});
445        let (a, b) = (mk(), mk());
446        let (la, lb) = (
447            serde_json::to_vec(&a).unwrap().len(),
448            serde_json::to_vec(&b).unwrap().len(),
449        );
450        let (out_a, _, _) = compress_request_body(a, la);
451        let (out_b, _, _) = compress_request_body(b, lb);
452        assert_eq!(out_a, out_b, "identical input must yield identical bytes");
453    }
454
455    #[test]
456    fn cache_aware_responses_prefix_is_byte_stable_across_turns() {
457        // THE cache invariant for the Responses rail: as `input` grows turn by
458        // turn, every item before an already-passed boundary must stay
459        // byte-identical, or OpenAI's automatic prompt cache stops hitting.
460        let _iso = crate::core::data_dir::isolated_data_dir();
461        let mut prev: Vec<String> = Vec::new();
462        let mut prev_boundary = 0;
463        for pairs in 1..=20 {
464            let input = responses_read_turns(pairs);
465            let len = input.len();
466            let body = serde_json::json!({"model": "gpt-5", "input": input});
467            let bytes = serde_json::to_vec(&body).unwrap();
468            let (out, _, _) = compress_request_body(body, bytes.len());
469            let parsed: Value = serde_json::from_slice(&out).unwrap();
470            let items: Vec<String> = parsed["input"]
471                .as_array()
472                .unwrap()
473                .iter()
474                .map(Value::to_string)
475                .collect();
476            for i in 0..prev_boundary {
477                assert_eq!(
478                    prev[i], items[i],
479                    "Responses item {i} changed at turn {pairs} — prompt cache prefix broken"
480                );
481            }
482            prev = items;
483            prev_boundary = crate::proxy::history_prune::prune_boundary(
484                crate::core::config::HistoryMode::CacheAware,
485                len,
486            );
487        }
488    }
489
490    #[test]
491    fn effort_control_sets_nested_reasoning_effort() {
492        // #834 end-to-end through the Responses request path.
493        let _iso = crate::core::data_dir::isolated_data_dir();
494        crate::test_env::remove_var("LEAN_CTX_PROXY_EFFORT");
495        crate::core::config::Config::update_global(|c| {
496            c.proxy.effort = Some("low".into());
497        })
498        .unwrap();
499        let body = serde_json::json!({"model": "gpt-5.5", "input": []});
500        let bytes = serde_json::to_vec(&body).unwrap();
501        let (out, _o, _c) = compress_request_body(body, bytes.len());
502        assert_eq!(
503            serde_json::from_slice::<Value>(&out).unwrap()["reasoning"]["effort"],
504            "low"
505        );
506    }
507}
lean_ctx/proxy/openai_responses.rs

lean_ctx/proxy/
openai_responses.rs
