From b9b777acaf05353943d280db47bda90ce4a06cbe Mon Sep 17 00:00:00 2001 From: iris-ai <4+iris-ai@noreply.localhost> Date: Mon, 14 Sep 2026 15:16:05 -0400 Subject: [PATCH] Release messages after Codex recovery --- PLAN.md | 4 ++ server/src/session/codex.rs | 115 +++++++++++++++++++++++++++++++++++- 2 files changed, 118 insertions(+), 1 deletion(-) diff --git a/PLAN.md b/PLAN.md index 2c82e85..2aaa61e 100644 --- a/PLAN.md +++ b/PLAN.md @@ -245,6 +245,10 @@ 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. +That notification retries the waiting queue, so delivery does not depend on a +later message happening to retry it. A message accepted while initialization or +recovery is still finding a thread is recorded as queued; the phone can therefore +reopen without losing the only visible copy before Codex acknowledges it. Codex subscription limits come from the CLI's `account/rateLimits/read` app-server request on the machine that runs Codex. This keeps login and diff --git a/server/src/session/codex.rs b/server/src/session/codex.rs index ffd9277..2941318 100644 --- a/server/src/session/codex.rs +++ b/server/src/session/codex.rs @@ -209,7 +209,16 @@ impl Driver for CodexDriver { }); return; } - let id = state.running.then(super::random_hex).unwrap_or_default(); + // Initialization and missing-thread recovery can leave an otherwise idle session without + // a thread to receive this yet. Announce that wait just like one behind an active turn: the + // phone stops preserving its local bridge once this method returns, so the server event is + // the durable copy until Codex acknowledges the message. + let waiting_for_protocol = state.initialize_request.is_some() + || state.thread_request.is_some() + || read_thread(&self.inner.session_dir).is_none(); + let id = (state.running || waiting_for_protocol) + .then(super::random_hex) + .unwrap_or_default(); let steering = state.running; state.running = true; state.waiting.push_back(Waiting { @@ -735,6 +744,13 @@ fn handle_line(inner: &Arc, translator: &mut Translator, line: &Value) { for event in translator.translate_with_prefix(line, images) { let _ = inner.sink.send(event); } + if parent && method == Some("thread/started") { + // `thread/start` answers before this notification. Ordinarily its response releases the + // queue, but this notification is the authoritative point at which the replacement root + // exists and the translator knows its id. Retrying here makes that handoff level-triggered: + // a message cannot remain stuck until another send happens to call `dispatch_waiting`. + dispatch_waiting(inner); + } if parent && method == Some("turn/completed") { dispatch_waiting(inner); } @@ -1345,4 +1361,101 @@ mod tests { Some("replacement-thread") ); } + + #[test] + fn a_message_waiting_for_a_thread_is_durable() { + let dir = tempfile::tempdir().expect("tempdir"); + let (sink, mut events) = mpsc::unbounded_channel(); + let (to_child, _requests) = mpsc::unbounded_channel(); + let inner = Arc::new(Inner { + sink, + state: Mutex::new(ProtocolState { + thread_request: Some("replacement-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), + }); + let driver = CodexDriver { + inner: Arc::clone(&inner), + }; + + driver.send_user_message("still send this".to_string(), Vec::new()); + + assert!(matches!( + events.try_recv().expect("durable queue event"), + Event::MessageQueued { text, .. } if text == "still send this" + )); + let state = inner.state.lock().unwrap(); + assert_eq!(state.waiting.len(), 1); + assert!(!state.waiting[0].id.is_empty()); + } + + #[test] + fn a_replacement_thread_notification_releases_its_waiting_message() { + let dir = tempfile::tempdir().expect("tempdir"); + let (sink, _events) = mpsc::unbounded_channel(); + let (to_child, mut requests) = mpsc::unbounded_channel(); + let inner = Arc::new(Inner { + sink, + state: Mutex::new(ProtocolState { + waiting: VecDeque::from([Waiting { + id: "queued-message".to_string(), + client_id: "client-message".to_string(), + steering: false, + text: "deliver me".to_string(), + attachments: Vec::new(), + }]), + ..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), + }); + let mut translator = Translator::new( + Arc::clone(&inner.subagents), + Some("missing-thread".to_string()), + false, + ); + + handle_line( + &inner, + &mut translator, + &json!({ + "method": "thread/started", + "params": {"thread": { + "id": "replacement-thread", + "parentThreadId": null + }} + }), + ); + + assert_eq!( + read_thread(dir.path()).as_deref(), + Some("replacement-thread") + ); + let request: Value = serde_json::from_str(&requests.try_recv().expect("turn request")) + .expect("request json"); + assert_eq!(request["method"], "turn/start"); + assert_eq!(request["params"]["threadId"], "replacement-thread"); + assert_eq!(request["params"]["clientUserMessageId"], "client-message"); + let state = inner.state.lock().unwrap(); + assert!(state.waiting.is_empty()); + assert_eq!(state.sent.len(), 1); + } }