diff --git a/PLAN.md b/PLAN.md index c1b3fa7..2c82e85 100644 --- a/PLAN.md +++ b/PLAN.md @@ -233,6 +233,19 @@ instead of being lost. An interrupt requested while a turn is still starting is applied once Codex supplies that turn's id, so it cannot leak forward and hide a later failure. +A missing thread is recoverable (2026-09-14). Codex returns a thread id before +its first turn creates a rollout, so restarting in between can leave ai-app +holding an id that `thread/resume` rejects as either "thread not found" or "no +rollout found". A rollout can disappear later too. Both mean the model context +is gone but ai-app's common transcript is not: the driver forgets only that +stale id, reports Codex's refusal, records a context-clear boundary, starts a +fresh Codex thread and then delivers anything queued. The error stays visible +because losing model context is material even when the process can heal it. +Other resume failures remain errors rather than silently discarding context. +The replacement's `thread/started` notification is a new root despite not +matching the translator's old root id; its null `parentThreadId` distinguishes +it from a subagent and moves the translator to the replacement conversation. + Codex subscription limits come from the CLI's `account/rateLimits/read` app-server request on the machine that runs Codex. This keeps login and token refresh inside the CLI. Its primary and secondary windows are normalized diff --git a/server/src/session/codex.rs b/server/src/session/codex.rs index 33712f0..ffd9277 100644 --- a/server/src/session/codex.rs +++ b/server/src/session/codex.rs @@ -894,6 +894,41 @@ fn handle_response(inner: &Arc, line: &Value) { if state.thread_request.as_deref() == Some(id) { state.thread_request = None; if let Some(error) = response_error(line) { + // `thread/start` returns an id before the first turn creates its rollout. If the + // app-server is restarted in that window, the id we correctly persisted is not one + // Codex can resume. A rollout can also disappear independently of ai-app. In either + // case our common transcript still exists, so start a blank Codex thread and mark the + // point where the model's context stopped instead of stranding every queued message. + if missing_thread(&error) && read_thread(&inner.session_dir).is_some() { + protocol_error(inner, error.clone()); + let path = inner.session_dir.join(THREAD_FILE); + match std::fs::remove_file(&path) { + Ok(()) => {} + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => { + drop(state); + protocol_error( + inner, + format!("couldn't forget the missing Codex thread: {err}"), + ); + return; + } + } + let request = request_id(); + state.thread_request = Some(request.clone()); + drop(state); + save_state(inner); + send_json( + inner, + json!({ + "id": request, + "method": "thread/start", + "params": thread_params(inner) + }), + ); + let _ = inner.sink.send(Event::Cleared); + return; + } drop(state); protocol_error(inner, error); return; @@ -989,6 +1024,11 @@ fn response_error(line: &Value) -> Option { .map(str::to_string) } +fn missing_thread(message: &str) -> bool { + let message = message.to_ascii_lowercase(); + message.contains("thread not found") || message.contains("no rollout found for thread id") +} + fn active_turn_not_steerable(line: &Value) -> bool { line.pointer("/error/data/codexErrorInfo/activeTurnNotSteerable") .is_some() @@ -1239,4 +1279,70 @@ mod tests { ); assert_eq!(context_from_rollout(""), None); } + + #[test] + fn both_missing_thread_failures_are_recognised() { + assert!(missing_thread( + "thread not found: 01a099bd-2e0f-7442-b750-a9f33998c8c5" + )); + assert!(missing_thread( + "no rollout found for thread id 01a099bd-2e0f-7442-b750-a9f33998c8c5" + )); + assert!(!missing_thread("thread is already running")); + } + + #[test] + fn a_missing_thread_is_reported_then_replaced() { + let dir = tempfile::tempdir().expect("tempdir"); + write_thread(dir.path(), "missing-thread"); + let (sink, mut events) = mpsc::unbounded_channel(); + let (to_child, mut requests) = mpsc::unbounded_channel(); + let inner = Arc::new(Inner { + sink, + state: Mutex::new(ProtocolState { + thread_request: Some("resume-request".to_string()), + ..ProtocolState::default() + }), + settings: Mutex::new(Settings { + model: None, + permission_mode: None, + effort: None, + }), + to_child, + transport: Transport::Here, + session_dir: dir.path().to_path_buf(), + subagents: Arc::new(Subagents::new(dir.path().to_path_buf())), + reading: AtomicBool::new(true), + }); + + handle_response( + &inner, + &json!({ + "id": "resume-request", + "error": {"message": "thread not found: missing-thread"} + }), + ); + + assert!(matches!( + events.try_recv().expect("visible refusal"), + Event::Error { message } if message == + "Codex refused a request: thread not found: missing-thread" + )); + assert_eq!(events.try_recv().expect("context boundary"), Event::Cleared); + assert_eq!(read_thread(dir.path()), None); + let request: Value = + serde_json::from_str(&requests.try_recv().expect("replacement thread request")) + .expect("request json"); + assert_eq!(request["method"], "thread/start"); + let request_id = request["id"].as_str().expect("request id"); + + handle_response( + &inner, + &json!({"id": request_id, "result": {"thread": {"id": "replacement-thread"}}}), + ); + assert_eq!( + read_thread(dir.path()).as_deref(), + Some("replacement-thread") + ); + } } diff --git a/server/src/session/codex/translate.rs b/server/src/session/codex/translate.rs index ebc1eb4..a7b4b53 100644 --- a/server/src/session/codex/translate.rs +++ b/server/src/session/codex/translate.rs @@ -42,12 +42,13 @@ impl Translator { if line.get("id").is_some() && line.get("method").is_none() { return true; } - if line.get("method").and_then(Value::as_str) == Some("thread/started") - && line - .pointer("/params/thread/parentThreadId") - .is_some_and(|parent| !parent.is_null()) - { - return false; + let kind = line.get("method").and_then(Value::as_str); + let body = kind.and_then(|_| line.get("params")).unwrap_or(line); + if let Some(root) = started_thread_is_root(kind, body) { + // A clear or missing-rollout recovery replaces the root id inside this same + // app-server. The new root cannot match the id this translator still holds; its null + // parent is the authoritative distinction from a newly spawned subagent. + return root; } match (self.thread_id.as_deref(), notification_thread(line)) { (Some(parent), Some(thread)) => parent == thread, @@ -80,10 +81,13 @@ impl Translator { self.remember_spawned_thread(kind, body); let thread = notification_thread(line); + let root_started = started_thread_is_root(kind, body) == Some(true); let child = thread.filter(|thread| { - self.thread_id - .as_deref() - .is_some_and(|parent| parent != *thread) + !root_started + && self + .thread_id + .as_deref() + .is_some_and(|parent| parent != *thread) }); if let Some(id) = child { self.ensure_child(id, "subagent", None); @@ -467,6 +471,13 @@ fn subagent_title(path: &str) -> String { .replace('_', " ") } +fn started_thread_is_root(kind: Option<&str>, body: &Value) -> Option { + (kind == Some("thread/started")).then(|| { + body.pointer("/thread/parentThreadId") + .is_none_or(Value::is_null) + }) +} + fn start_item(item: &Value) -> Vec { if matches!( item.get("type").and_then(Value::as_str), @@ -935,6 +946,25 @@ mod tests { assert!(translator.limited()); } + #[test] + fn a_replacement_root_is_not_mistaken_for_a_subagent() { + let mut translator = Translator { + thread_id: Some("old-root".to_string()), + ..Translator::default() + }; + let replacement = line( + r#"{"method":"thread/started","params":{"thread":{"id":"new-root","parentThreadId":null}}}"#, + ); + assert!(translator.is_parent(&replacement)); + assert!(translator.translate(&replacement).is_empty()); + assert_eq!(translator.thread_id.as_deref(), Some("new-root")); + + let child = line( + r#"{"method":"thread/started","params":{"thread":{"id":"child","parentThreadId":"new-root"}}}"#, + ); + assert!(!translator.is_parent(&child)); + } + #[test] fn command_translation_only_hides_the_known_bash_wrapper() { let legacy = tool(&line(