diff --git a/PLAN.md b/PLAN.md index 67691f9..7307322 100644 --- a/PLAN.md +++ b/PLAN.md @@ -236,11 +236,16 @@ 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. +rollout found" (other CLI versions say "invalid thread id", and malformed ids +are "invalid session id"). A rollout can disappear later too, including between +a successful resume and `turn/start`. 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. A turn request that discovers the +loss returns its in-flight message to that queue before recovery, so the message +that triggered recovery is delivered to the replacement rather than +disappearing. 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 diff --git a/server/src/session/codex.rs b/server/src/session/codex.rs index 2fa5222..9b408d1 100644 --- a/server/src/session/codex.rs +++ b/server/src/session/codex.rs @@ -540,7 +540,11 @@ fn dispatch_waiting(inner: &Arc) { loop { let (message, kind, active_turn) = { let mut state = inner.state.lock().unwrap(); - if state.closed || state.waiting.is_empty() { + if state.closed + || state.waiting.is_empty() + || state.initialize_request.is_some() + || state.thread_request.is_some() + { return; } if state.active_turn.is_none() @@ -932,33 +936,7 @@ fn handle_response(inner: &Arc, line: &Value) { // 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); + replace_missing_thread(inner, state, error); return; } drop(state); @@ -979,6 +957,14 @@ fn handle_response(inner: &Arc, line: &Value) { }; let request = state.pending.remove(at); if let Some(error) = response_error(line) { + // A rollout can vanish after this app-server resumed it but before the next turn reaches + // Codex. Put that turn (and any steers already in flight behind it) back before replacing + // the thread; the request that discovered the loss is still the user's first message in + // the new context, not a failed message to discard. + if missing_thread(&error) && read_thread(&inner.session_dir).is_some() { + replace_missing_thread(inner, state, error); + return; + } let message = state .sent .iter() @@ -1058,7 +1044,49 @@ fn response_error(line: &Value) -> Option { 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") + message.contains("invalid thread id") + || message.contains("invalid session id") + || message.contains("thread not found") + || message.contains("no rollout found for thread id") +} + +fn replace_missing_thread( + inner: &Arc, + mut state: std::sync::MutexGuard<'_, ProtocolState>, + error: String, +) { + protocol_error(inner, error); + let path = inner.session_dir.join(THREAD_FILE); + if let Err(err) = std::fs::remove_file(&path) + && err.kind() != std::io::ErrorKind::NotFound + { + drop(state); + protocol_error( + inner, + format!("couldn't forget the missing Codex thread: {err}"), + ); + return; + } + while let Some(message) = state.sent.pop_back() { + state.waiting.push_front(message); + } + state.pending.clear(); + state.active_turn = None; + state.interrupt_when_started = false; + state.running = !state.waiting.is_empty(); + 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); } fn active_turn_not_steerable(line: &Value) -> bool { @@ -1313,7 +1341,13 @@ mod tests { } #[test] - fn both_missing_thread_failures_are_recognised() { + fn missing_thread_failures_are_recognised() { + assert!(missing_thread( + "invalid thread id: 01a099bd-2e0f-7442-b750-a9f33998c8c5" + )); + assert!(missing_thread( + "invalid session id: invalid character in a thread id" + )); assert!(missing_thread( "thread not found: 01a099bd-2e0f-7442-b750-a9f33998c8c5" )); @@ -1381,8 +1415,9 @@ mod tests { #[test] fn a_message_waiting_for_a_thread_is_durable() { let dir = tempfile::tempdir().expect("tempdir"); + write_thread(dir.path(), "not-resumed-yet"); let (sink, mut events) = mpsc::unbounded_channel(); - let (to_child, _requests) = mpsc::unbounded_channel(); + let (to_child, mut requests) = mpsc::unbounded_channel(); let inner = Arc::new(Inner { sink, state: Mutex::new(ProtocolState { @@ -1413,6 +1448,36 @@ mod tests { let state = inner.state.lock().unwrap(); assert_eq!(state.waiting.len(), 1); assert!(!state.waiting[0].id.is_empty()); + let client_id = state.waiting[0].client_id.clone(); + drop(state); + assert!( + requests.try_recv().is_err(), + "the message was sent before its thread had resumed" + ); + + handle_response( + &inner, + &json!({ + "id": "replacement-request", + "error": {"message": "invalid thread id: not-resumed-yet"} + }), + ); + let replacement: Value = + serde_json::from_str(&requests.try_recv().expect("replacement thread request")) + .expect("request json"); + let replacement_id = replacement["id"].as_str().expect("request id"); + handle_response( + &inner, + &json!({"id": replacement_id, "result": {"thread": {"id": "new-thread"}}}), + ); + + let turn: Value = + serde_json::from_str(&requests.try_recv().expect("waiting message request")) + .expect("request json"); + assert_eq!(turn["method"], "turn/start"); + assert_eq!(turn["params"]["threadId"], "new-thread"); + assert_eq!(turn["params"]["clientUserMessageId"], client_id); + assert_eq!(turn["params"]["input"][0]["text"], "still send this"); } #[test] @@ -1529,4 +1594,80 @@ mod tests { assert!(state.waiting.is_empty()); assert_eq!(state.sent.len(), 1); } + + #[test] + fn a_turn_that_discovers_a_missing_thread_is_retried() { + let dir = tempfile::tempdir().expect("tempdir"); + write_thread(dir.path(), "missing-thread"); + let message = Waiting { + id: "queued-message".to_string(), + client_id: "client-message".to_string(), + steering: false, + text: "do not lose me".to_string(), + attachments: Vec::new(), + }; + 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 { + sent: VecDeque::from([message]), + pending: vec![PendingRequest { + id: "turn-request".to_string(), + client_id: "client-message".to_string(), + kind: RequestKind::Start, + }], + running: true, + ..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": "turn-request", + "error": {"message": "invalid thread id: missing-thread"} + }), + ); + + assert!(matches!( + events.try_recv().expect("visible refusal"), + Event::Error { message } if message == + "Codex refused a request: invalid thread id: missing-thread" + )); + assert_eq!(events.try_recv().expect("context boundary"), Event::Cleared); + assert!( + events.try_recv().is_err(), + "the message was reported dropped" + ); + assert_eq!(read_thread(dir.path()), None); + let replacement: Value = + serde_json::from_str(&requests.try_recv().expect("replacement thread request")) + .expect("request json"); + assert_eq!(replacement["method"], "thread/start"); + let replacement_id = replacement["id"].as_str().expect("request id"); + + handle_response( + &inner, + &json!({"id": replacement_id, "result": {"thread": {"id": "new-thread"}}}), + ); + + let retried: Value = + serde_json::from_str(&requests.try_recv().expect("retried turn request")) + .expect("request json"); + assert_eq!(retried["method"], "turn/start"); + assert_eq!(retried["params"]["threadId"], "new-thread"); + assert_eq!(retried["params"]["clientUserMessageId"], "client-message"); + assert_eq!(retried["params"]["input"][0]["text"], "do not lose me"); + } }