Recover Codex sessions with missing rollouts

This commit is contained in:
iris-ai committed 2026-09-14 15:01:47 -04:00
1 parent 3af2502982
commit 46831520e3
3 files changed
+158 -9

No files matched your search

+13
View File
@@ -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 is applied once Codex supplies that turn's id, so it cannot leak forward and
hide a later failure. 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` Codex subscription limits come from the CLI's `account/rateLimits/read`
app-server request on the machine that runs Codex. This keeps login and 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 token refresh inside the CLI. Its primary and secondary windows are normalized
+106
View File
@@ -894,6 +894,41 @@ fn handle_response(inner: &Arc<Inner>, line: &Value) {
if state.thread_request.as_deref() == Some(id) { if state.thread_request.as_deref() == Some(id) {
state.thread_request = None; state.thread_request = None;
if let Some(error) = response_error(line) { 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); drop(state);
protocol_error(inner, error); protocol_error(inner, error);
return; return;
@@ -989,6 +1024,11 @@ fn response_error(line: &Value) -> Option<String> {
.map(str::to_string) .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 { fn active_turn_not_steerable(line: &Value) -> bool {
line.pointer("/error/data/codexErrorInfo/activeTurnNotSteerable") line.pointer("/error/data/codexErrorInfo/activeTurnNotSteerable")
.is_some() .is_some()
@@ -1239,4 +1279,70 @@ mod tests {
); );
assert_eq!(context_from_rollout(""), None); 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")
);
}
} }
+39 -9
View File
@@ -42,12 +42,13 @@ impl Translator {
if line.get("id").is_some() && line.get("method").is_none() { if line.get("id").is_some() && line.get("method").is_none() {
return true; return true;
} }
if line.get("method").and_then(Value::as_str) == Some("thread/started") let kind = line.get("method").and_then(Value::as_str);
&& line let body = kind.and_then(|_| line.get("params")).unwrap_or(line);
.pointer("/params/thread/parentThreadId") if let Some(root) = started_thread_is_root(kind, body) {
.is_some_and(|parent| !parent.is_null()) // 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
return false; // parent is the authoritative distinction from a newly spawned subagent.
return root;
} }
match (self.thread_id.as_deref(), notification_thread(line)) { match (self.thread_id.as_deref(), notification_thread(line)) {
(Some(parent), Some(thread)) => parent == thread, (Some(parent), Some(thread)) => parent == thread,
@@ -80,10 +81,13 @@ impl Translator {
self.remember_spawned_thread(kind, body); self.remember_spawned_thread(kind, body);
let thread = notification_thread(line); let thread = notification_thread(line);
let root_started = started_thread_is_root(kind, body) == Some(true);
let child = thread.filter(|thread| { let child = thread.filter(|thread| {
self.thread_id !root_started
.as_deref() && self
.is_some_and(|parent| parent != *thread) .thread_id
.as_deref()
.is_some_and(|parent| parent != *thread)
}); });
if let Some(id) = child { if let Some(id) = child {
self.ensure_child(id, "subagent", None); self.ensure_child(id, "subagent", None);
@@ -467,6 +471,13 @@ fn subagent_title(path: &str) -> String {
.replace('_', " ") .replace('_', " ")
} }
fn started_thread_is_root(kind: Option<&str>, body: &Value) -> Option<bool> {
(kind == Some("thread/started")).then(|| {
body.pointer("/thread/parentThreadId")
.is_none_or(Value::is_null)
})
}
fn start_item(item: &Value) -> Vec<Event> { fn start_item(item: &Value) -> Vec<Event> {
if matches!( if matches!(
item.get("type").and_then(Value::as_str), item.get("type").and_then(Value::as_str),
@@ -935,6 +946,25 @@ mod tests {
assert!(translator.limited()); 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] #[test]
fn command_translation_only_hides_the_known_bash_wrapper() { fn command_translation_only_hides_the_known_bash_wrapper() {
let legacy = tool(&line( let legacy = tool(&line(