iris/ai-app
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Preserve messages during Codex thread recovery

Browse Source
This commit is contained in:
iris-ai committed 2026-09-15 15:30:22 -04:00
1 parent 3f94eeb6d6
commit 06bf1c8f81
2 files changed
+182 -36
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/session/codex.rs
+172 -31
View File
@@ -540,7 +540,11 @@ fn dispatch_waiting(inner: &Arc<Inner>) {
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<Inner>, 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<Inner>, 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<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")
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<Inner>,
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");
}
}