From 86276ee9744270e16ba62f28c618741f55429566 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 17 Sep 2026 06:20:29 +0200 Subject: [PATCH] feat(codex): expose advertised context windows and preserve threads when switching --- .../neosh-provider/src/drivers/codex_cli.rs | 242 ++++++++++++++++-- .../neosh-provider/tests/codex_app_server.rs | 117 +++++++++ crates/neosh/tests/builtin_plugins.rs | 61 +++++ docs/release-notes/v0.4.10.md | 13 + 4 files changed, 410 insertions(+), 23 deletions(-) create mode 100644 docs/release-notes/v0.4.10.md diff --git a/crates/neosh-provider/src/drivers/codex_cli.rs b/crates/neosh-provider/src/drivers/codex_cli.rs index a70ab2c..361b4cd 100644 --- a/crates/neosh-provider/src/drivers/codex_cli.rs +++ b/crates/neosh-provider/src/drivers/codex_cli.rs @@ -39,9 +39,8 @@ //! //! Started once and kept, like `claude_cli` — but simpler, because there is nothing here that has //! to be said on a command line. The model, the effort, the sandbox, the approval policy and even -//! the directory are all parameters of `turn/start`, so **nothing ever needs a new process**: a -//! change to any of them takes effect on the next turn with no relaunch at all. What the process -//! holds is the thread, and `thread/resume` puts it back if it ever has to be replaced. +//! the directory are parameters of `turn/start`. Context configuration is thread-scoped instead: +//! changing it replaces the process between turns and resumes the same persisted thread. use std::collections::HashMap; use std::process::Stdio; @@ -113,6 +112,8 @@ struct Conversation { /// two conversations down one pipe. live: tokio::sync::Mutex>, tune: Mutex, + /// Retained across process/configuration failures, so retrying never starts an empty thread. + resume: Mutex>, } /// A `codex app-server` that is still running, between turns as well as during them. @@ -132,6 +133,7 @@ struct Live { cwd: std::path::PathBuf, /// Request ids, which have to be unique within the connection. next_id: u64, + context: Option, } #[derive(Debug)] @@ -692,9 +694,98 @@ async fn discover(program: &str) -> Result, ProviderError> { if out.is_empty() { return Err(ProviderError::BadResponse(format!("{program} listed no models"))); } + add_context_options(&mut out, &context_metadata().await); Ok(out) } +/// model/list omits context limits. Codex's own cache contains the account/model-specific +/// default and maximum; an API model's advertised maximum is not necessarily this CLI's limit. +async fn context_metadata() -> Value { + let home = std::env::var_os("CODEX_HOME") + .map(std::path::PathBuf::from) + .or_else(|| std::env::var_os("HOME").map(|p| std::path::PathBuf::from(p).join(".codex"))); + let Some(home) = home else { return Value::Null }; + let path = home.join("models_cache.json"); + if !tokio::fs::metadata(&path).await.is_ok_and(|m| m.len() <= 8 * 1024 * 1024) { + return Value::Null; + } + tokio::fs::read(path) + .await + .ok() + .and_then(|b| serde_json::from_slice(&b).ok()) + .unwrap_or(Value::Null) +} + +fn add_context_options(models: &mut [ModelInfo], cache: &Value) { + let Some(entries) = cache.get("models").and_then(Value::as_array) else { return }; + for model in models { + let Some(entry) = entries.iter().find(|v| str_at(v, "slug") == Some(model.id.as_ref())) + else { + continue; + }; + let Some(standard) = + entry.get("context_window").and_then(Value::as_u64).filter(|n| *n >= 1000) + else { + continue; + }; + let Some(maximum) = entry + .get("max_context_window") + .and_then(Value::as_u64) + .filter(|n| *n > standard && *n <= i64::MAX as u64) + else { + continue; + }; + let label = |n: u64| { + if n % 1_000_000 == 0 { + format!("{}M", n / 1_000_000) + } else if n % 1000 == 0 { + format!("{}k", n / 1000) + } else { + n.to_string() + } + }; + model.capabilities.option_descriptors.retain(|d| d.id() != "context"); + model.capabilities.option_descriptors.push(ProviderOptionDescriptor::Select { + id: "context".into(), label: "Context".into(), + description: Some("Applies next turn. Codex reserves some space and may compact when shrinking.".into()), + options: vec![ + OptionChoice { id: "default".into(), label: "Default context".into(), + description: Some("Use Codex's own configuration and compaction threshold.".into()), is_default: true }, + OptionChoice { id: standard.to_string(), label: label(standard), + description: Some("Standard window advertised by Codex.".into()), is_default: false }, + OptionChoice { id: maximum.to_string(), label: label(maximum), + description: Some("Largest window advertised by Codex. Long requests can consume more allowance.".into()), is_default: false }, + ], + current_value: Some("default".into()), prompt_injected_values: Vec::new(), + }); + } +} + +fn selected_context( + models: &[ModelInfo], + selection: &neosh_proto::ModelSelection, +) -> Result, String> { + let Some(value) = selection.option_str("context").filter(|s| *s != "default") else { + return Ok(None); + }; + let offered = models.iter().find(|m| m.id == selection.model).is_some_and(|m| { + m.capabilities.option_descriptors.iter().any(|d| { + matches!(d, + ProviderOptionDescriptor::Select { id, options, .. } + if id == "context" && options.iter().any(|o| o.id == value)) + }) + }); + if !offered { + return Err("Codex no longer advertises this context window; select Default context in model options.".into()); + } + value + .parse::() + .ok() + .filter(|n| *n >= 1000 && *n <= i64::MAX as u64) + .map(Some) + .ok_or_else(|| "Invalid Codex context window.".into()) +} + /// One entry of a `model/list` reply. /// /// `pub` because the shape is worth testing against a recorded reply without a `codex` on `PATH`, @@ -887,7 +978,9 @@ impl Provider for CodexCliProvider { Ok(models) => Ok(models), Err(e) => { tracing::debug!(program = %self.program, "codex model discovery failed: {e}"); - Ok(instance.models.clone()) + let mut models = instance.models.clone(); + add_context_options(&mut models, &context_metadata().await); + Ok(models) } } } @@ -915,11 +1008,30 @@ impl Provider for CodexCliProvider { let mode = *self.mode.lock().expect("mode lock poisoned"); let asker = self.asker.lock().expect("asker lock poisoned").clone(); let tier = service_tier(instance, &request.selection).to_string(); + let mut context_models = instance.models.clone(); + if !context_models.iter().any(|m| m.id == request.selection.model) { + context_models.push(ModelInfo::undescribed( + request.selection.model.as_ref(), + request.selection.model.as_ref(), + )); + } tokio::spawn(async move { + if request.selection.option_str("context").is_some_and(|v| v != "default") { + add_context_options(&mut context_models, &context_metadata().await); + } + let context = match selected_context(&context_models, &request.selection) { + Ok(context) => context, + Err(message) => { + let _ = tx.send(ProviderEvent::Error { message, retryable: false }).await; + return; + } + }; let mut guard = slot.live.lock().await; - if let Err(message) = - run_turn(&program, &slot, &mut guard, request, &tier, mode, asker, cancel, &tx).await + if let Err(message) = run_turn( + &program, &slot, &mut guard, request, &tier, context, mode, asker, cancel, &tx, + ) + .await { // A failure the process cannot recover from leaves nothing worth keeping: the next // turn starts a fresh app-server rather than writing into a pipe whose other end @@ -948,15 +1060,15 @@ async fn run_turn( slot: &mut Option, request: TurnRequest, service_tier: &str, + context: Option, mode: PermissionMode, asker: Option>, cancel: CancellationToken, tx: &mpsc::Sender, ) -> Result<(), String> { - // The thread's directory is what the agent's tools resolve against, so a conversation that has - // moved needs a new thread — the one thing here that a running server cannot be talked out of. - // Even that keeps the process. - if slot.as_ref().is_some_and(|l| l.cwd != request.cwd) + // Resume in a fresh process: resuming an already loaded thread can ignore config overrides. + // This runs under the conversation lock, after the preceding turn was completed/drained. + if slot.as_ref().is_some_and(|l| l.cwd != request.cwd || l.context != context) && let Some(live) = slot.take() { live.close().await; @@ -966,7 +1078,14 @@ async fn run_turn( } let live = slot.as_mut().expect("a server was just put there"); if live.thread.is_none() { - live.start_thread(&request.cwd, mode).await?; + let resume = conversation + .resume + .lock() + .expect("resume lock poisoned") + .clone() + .or_else(|| request.resume.clone()); + live.start_thread(&request, mode, context, resume.as_deref()).await?; + *conversation.resume.lock().expect("resume lock poisoned") = live.thread.clone(); } // Everything that would have been a command-line argument is a parameter of this one call, so @@ -1109,9 +1228,13 @@ async fn run_turn( } for ev in app_server_event(&v, &mut state) { finished |= matches!(ev, ProviderEvent::MessageStop); + let began = matches!(ev, ProviderEvent::MessageStart { .. }); // A dropped receiver triggers interruption above. Keep draining to // this turn's boundary before handing the process to the next turn. let _ = tx.send(ev).await; + if began && let Some(token) = &live.thread { + let _ = tx.send(activity(Activity::Resume { token: token.clone() })).await; + } } if finished { break; @@ -1276,6 +1399,7 @@ impl Live { thread: None, cwd: cwd.to_path_buf(), next_id: 1, + context: None, }; live.handshake().await?; Ok(live) @@ -1300,19 +1424,34 @@ impl Live { .map_err(|e| format!("codex app-server: {e}")) } - async fn start_thread(&mut self, cwd: &std::path::Path, mode: PermissionMode) -> Result<(), String> { + async fn start_thread( + &mut self, + request: &TurnRequest, + mode: PermissionMode, + context: Option, + resume: Option<&str>, + ) -> Result<(), String> { + let mut params = json!({ + "cwd": request.cwd.display().to_string(), + "model": request.selection.model.as_ref(), + "sandbox": match mode { + PermissionMode::Deny | PermissionMode::Ask => "read-only", + PermissionMode::AllowListed => "workspace-write", + PermissionMode::Allow => "danger-full-access", + }, + }); + if let Some(window) = context { + params["config"] = json!({ + "model_context_window": window, + // Leave headroom for the response and Codex's effective-window reservation. + "model_auto_compact_token_limit": window / 10 * 9, + }); + } + if let Some(thread) = resume { + params["threadId"] = json!(thread); + } let id = self - .request( - "thread/start", - json!({ - "cwd": cwd.display().to_string(), - "sandbox": match mode { - PermissionMode::Deny | PermissionMode::Ask => "read-only", - PermissionMode::AllowListed => "workspace-write", - PermissionMode::Allow => "danger-full-access", - }, - }), - ) + .request(if resume.is_some() { "thread/resume" } else { "thread/start" }, params) .await?; let reply = self.await_reply(id).await?; self.thread = reply @@ -1321,6 +1460,7 @@ impl Live { .map(str::to_string) .ok_or_else(|| "codex app-server started no thread".to_string())? .into(); + self.context = context; Ok(()) } @@ -1482,6 +1622,62 @@ mod tests { .collect() } + #[test] + fn context_choices_follow_codex_limits_and_do_not_invent_one_million() { + let mut models = vec![ModelInfo::undescribed("test-model", "Test")]; + let cache = json!({"models":[{"slug":"test-model","context_window":272000,"max_context_window":872000}]}); + add_context_options(&mut models, &cache); + add_context_options(&mut models, &cache); + assert_eq!(models[0].capabilities.option_descriptors.len(), 1); + let ProviderOptionDescriptor::Select { options, .. } = + &models[0].capabilities.option_descriptors[0] + else { + panic!("select") + }; + assert_eq!( + options.iter().map(|o| o.label.as_str()).collect::>(), + ["Default context", "272k", "872k"] + ); + let mut selection = neosh_proto::ModelSelection { + instance: "codex-cli".into(), + model: "test-model".into(), + options: vec![], + }; + assert_eq!(selected_context(&models, &selection).expect("default"), None); + selection.options.push(neosh_proto::OptionSelection { + id: "context".into(), + value: neosh_proto::ProviderOptionValue::Text("872000".into()), + }); + assert_eq!(selected_context(&models, &selection).expect("extended"), Some(872000)); + selection.options[0].value = neosh_proto::ProviderOptionValue::Text("1000000".into()); + assert!(selected_context(&models, &selection).is_err()); + add_context_options( + &mut models, + &json!({"models":[{"slug":"test-model","context_window":272000,"max_context_window":1000000}]}), + ); + assert_eq!(selected_context(&models, &selection).expect("advertised 1M"), Some(1000000)); + } + + #[test] + fn absent_or_invalid_context_capabilities_do_not_offer_a_switcher() { + for entry in [ + Value::Null, + json!({}), + json!({"context_window":272000}), + json!({"context_window":272000,"max_context_window":272000}), + json!({"context_window":0,"max_context_window":1000000}), + json!({"context_window":272000,"max_context_window":-1}), + ] { + let mut models = vec![ModelInfo::undescribed("test-model", "Test")]; + let mut entry = entry; + if let Some(map) = entry.as_object_mut() { + map.insert("slug".into(), json!("test-model")); + } + add_context_options(&mut models, &json!({"models":[entry]})); + assert!(models[0].capabilities.option_descriptors.is_empty()); + } + } + /// A turn as the app-server reports it: the handshake shapes are a real capture from /// `codex-cli 0.148.0`; the answer path follows the protocol schema, since a logged-out codex /// cannot produce one to record. diff --git a/crates/neosh-provider/tests/codex_app_server.rs b/crates/neosh-provider/tests/codex_app_server.rs index 8f48a97..99d99f8 100644 --- a/crates/neosh-provider/tests/codex_app_server.rs +++ b/crates/neosh-provider/tests/codex_app_server.rs @@ -536,3 +536,120 @@ async fn speed_can_be_enabled_and_disabled_on_the_same_conversation() { assert_eq!(tiers, ["priority", "default", "priority", "default"]); p.shutdown(&SessionId::from("c1")); } + +const CONTEXT_FAKE: &str = r#"#!/bin/sh +window=258400 +while IFS= read -r line; do + printf '%s\n' "$line" >> "$0.requests" + id=$(printf '%s\n' "$line" | sed -n 's/.*"id":\([0-9]*\).*/\1/p') + case "$line" in + *'"method":"initialize"'*) printf '{"id":%s,"result":{}}\n' "$id" ;; + *'"method":"thread/start"'*|*'"method":"thread/resume"'*) + if [ -f "$0.reject" ]; then + rm "$0.reject" + printf '{"id":%s,"error":{"message":"retry context change"}}\n' "$id" + continue + fi + case "$line" in *'"model_context_window":1000000'*) window=950000 ;; esac + printf '{"id":%s,"result":{"thread":{"id":"t1"}}}\n' "$id" ;; + *'"method":"turn/start"'*) + printf '{"id":%s,"result":{"turn":{"id":"u1"}}}\n' "$id" + echo '{"method":"turn/started","params":{"threadId":"t1","turn":{"id":"u1"}}}' + printf '{"method":"thread/tokenUsage/updated","params":{"threadId":"t1","turnId":"u1","tokenUsage":{"last":{"totalTokens":12000},"modelContextWindow":%s}}}\n' "$window" + echo '{"method":"turn/completed","params":{"threadId":"t1","turn":{"id":"u1","status":"completed"}}}' ;; + esac +done +"#; + +#[tokio::test] +async fn context_changes_resume_the_same_thread_and_default_clears_overrides() { + let fake = Fake::new("context-switch"); + support::write_executable(&fake.dir.join("codex"), CONTEXT_FAKE); + let p = CodexCliProvider::new(fake.program()); + let mut instance = inst(); + let mut model = neosh_proto::ModelInfo::undescribed("test-context-model", "Test"); + model.capabilities.option_descriptors.push(neosh_proto::ProviderOptionDescriptor::Select { + id: "context".into(), + label: "Context".into(), + description: None, + options: ["272000", "1000000"] + .into_iter() + .map(|id| neosh_proto::OptionChoice { + id: id.into(), + label: id.into(), + description: None, + is_default: false, + }) + .collect(), + current_value: None, + prompt_injected_values: vec![], + }); + instance.models.push(model); + for (choice, expected) in [ + (None, 258400), + (Some("272000"), 258400), + (Some("1000000"), 950000), + (Some("1000000"), 950000), + (Some("default"), 258400), + ] { + let mut request = req(&fake.dir); + request.selection.model = "test-context-model".into(); + if let Some(value) = choice { + request.selection.options.push(neosh_proto::OptionSelection { + id: "context".into(), + value: neosh_proto::ProviderOptionValue::Text(value.into()), + }); + } + // A rejected resume must retain the thread for a retry, not silently start a new one. + if choice == Some("272000") { + std::fs::write(fake.dir.join("codex.reject"), "").expect("reject once"); + let failed: Vec<_> = + p.stream(&instance, request.clone(), CancellationToken::new()).collect().await; + assert!(failed.iter().any(|e| matches!(e, ProviderEvent::Error { .. })), "{failed:?}"); + } + let events = tokio::time::timeout( + std::time::Duration::from_secs(10), + p.stream(&instance, request, CancellationToken::new()).collect::>(), + ) + .await + .expect("turn finishes"); + assert!(events.contains(&ProviderEvent::MessageStop), "{events:?}"); + assert!( + events.iter().any(|e| matches!(e, ProviderEvent::Activity { + activity: neosh_proto::Activity::Context { used: 12000, total } + } if *total == expected)), + "{events:?}" + ); + assert!( + events.iter().any(|e| matches!(e, ProviderEvent::Activity { + activity: neosh_proto::Activity::Resume { token } + } if token == "t1")), + "{events:?}" + ); + } + p.shutdown(&SessionId::from("c1")); + // A new driver can also resume the saved token after neosh itself restarts. + let p = CodexCliProvider::new(fake.program()); + let mut request = req(&fake.dir); + request.resume = Some("t1".into()); + let events: Vec<_> = p.stream(&instance, request, CancellationToken::new()).collect().await; + assert!(events.contains(&ProviderEvent::MessageStop), "{events:?}"); + let lines = std::fs::read_to_string(fake.dir.join("codex.requests")).expect("wire log"); + let requests: Vec = + lines.lines().map(|s| serde_json::from_str(s).expect("json")).collect(); + assert_eq!(requests.iter().filter(|r| r["method"] == "thread/start").count(), 1); + let resumes: Vec<_> = requests.iter().filter(|r| r["method"] == "thread/resume").collect(); + assert_eq!( + resumes.len(), + 5, + "changes, rejected retry, and saved resume; unchanged context reuses process" + ); + for r in &resumes { + assert_eq!(r["params"]["threadId"], "t1"); + } + assert_eq!(resumes[0]["params"]["config"]["model_context_window"], 272000); + assert_eq!(resumes[2]["params"]["config"]["model_context_window"], 1000000); + assert_eq!(resumes[2]["params"]["config"]["model_auto_compact_token_limit"], 900000); + assert!(resumes[3]["params"].get("config").is_none(), "Default restores Codex configuration"); + p.shutdown(&SessionId::from("c1")); +} diff --git a/crates/neosh/tests/builtin_plugins.rs b/crates/neosh/tests/builtin_plugins.rs index 0a2f76c..5dcfa45 100644 --- a/crates/neosh/tests/builtin_plugins.rs +++ b/crates/neosh/tests/builtin_plugins.rs @@ -2041,6 +2041,67 @@ fn model_speed_is_visible_and_switches_between_standard_and_fast() { s.wait_for(r#"selection: lab/brainy [{"id":"service_tier","value":"default"}]"#); } +#[test] +fn model_context_is_visible_and_switches_between_default_and_expanded() { + let sb = Sandbox::new("model-context"); + let dir = sb.root.join("config/plugins/lab"); + std::fs::create_dir_all(&dir).expect("mkdir"); + std::fs::write( + dir.join("plugin.toml"), + "name = \"lab\"\nversion = \"0.1.0\"\nentry = \"main.ts\"\npermissions = [\"providers\"]\n", + ) + .expect("manifest"); + let plugin = INVENTED_KNOB + .replace("vibe", "context") + .replace("Vibe", "Context") + .replace("chill", "default") + .replace("Chill", "Default context") + .replace("intense", "1000000") + .replace("Intense", "1M") + .replace( + r#"neosh.event.on("neosh.ready", () => neosh.notify("lab ready"));"#, + r#"neosh.event.on("neosh.ready", async () => { + // A stored conversation may not have the newly discovered context option yet. + await neosh.agent.setSelection({ instance: "lab", model: "brainy", options: [] }); + neosh.notify("lab ready"); + });"#, + ); + std::fs::write(dir.join("main.ts"), plugin).expect("plugin"); + sb.write_config("[options]\n\"agent.model\" = \"lab/brainy\"\n"); + let mut s = sb.start_letting_config_choose(); + s.wait_for("lab ready"); + assert!( + s.pump(|s| s.status_now().join("").contains("Default context")), + "default context in footer: {:?}", + s.status_now() + ); + s.ctrl("e"); + s.wait_open("[model options]"); + s.wait_for("Context"); + s.special("right"); + s.special("enter"); + s.wait_closed("[model options]"); + assert!( + s.pump(|s| s.status_now().join("").contains("1M")), + "expanded context in footer: {:?}", + s.status_now() + ); + s.send(&command("lab.report")); + s.wait_for(r#"selection: lab/brainy [{"id":"context","value":"1000000"}]"#); + s.ctrl("e"); + s.wait_open("[model options]"); + s.special("left"); + s.special("enter"); + s.wait_closed("[model options]"); + assert!( + s.pump(|s| s.status_now().join("").contains("Default context")), + "default context in footer: {:?}", + s.status_now() + ); + s.send(&command("lab.report")); + s.wait_for(r#"selection: lab/brainy [{"id":"context","value":"default"}]"#); +} + #[test] fn a_driver_can_invent_an_option_and_the_switcher_renders_it() { let sb = Sandbox::new("invented"); diff --git a/docs/release-notes/v0.4.10.md b/docs/release-notes/v0.4.10.md new file mode 100644 index 0000000..280132b --- /dev/null +++ b/docs/release-notes/v0.4.10.md @@ -0,0 +1,13 @@ +# v0.4.10 — Codex context-window controls + +Open **Ctrl+E**, select **Context** with **j/k**, and choose a window with **h/l** or the arrow keys. The choice also appears beside the model in the footer and applies on the next turn. + +Choices come from Codex’s own model metadata: **Default context** preserves your Codex configuration, while the numeric choices select its advertised standard and extended windows. A **1M** choice is offered when Codex advertises that capacity. Limits can differ from the model’s API specifications: the Codex installation used for verification advertised 272k and 872k for Astra and the GPT-5.6 models. Models without a known larger window do not get an invented option. + +Codex reserves part of the configured window. The usage meter continues to show the usable window reported by the running agent; it does not substitute the requested capacity. Selecting a numeric window also sets automatic compaction at 90% of that capacity. Shrinking the window may cause Codex to compact history on the next turn. + +Window changes resume the same conversation in a fresh Codex process. Returning to Default removes both overrides. Codex thread IDs are now saved so the conversation can also resume after a workspace restart, and a rejected configuration change retains the thread for retry. + +The controls use the public model-options API and work with the existing options panel and plugin overrides. Your Codex configuration files are not changed. + +After upgrading, restart the workspace once active turns have finished. Existing Codex conversations created before this release have no saved Codex thread ID until they run through the updated driver.