From 3f94eeb6d6ec6de2f09f71462cfd0bc9f7e951af Mon Sep 17 00:00:00 2001 From: iris-ai <4+iris-ai@noreply.localhost> Date: Tue, 15 Sep 2026 14:43:23 -0400 Subject: [PATCH] Create Codex threads on first message after clear --- PLAN.md | 4 ++ server/src/session/codex.rs | 89 +++++++++++++++++++++++++++++++++---- 2 files changed, 84 insertions(+), 9 deletions(-) diff --git a/PLAN.md b/PLAN.md index 9c10176..67691f9 100644 --- a/PLAN.md +++ b/PLAN.md @@ -249,6 +249,10 @@ 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. +Clearing does not eagerly create that replacement: it forgets the old id and +the next message starts a thread and its first turn together. An empty replacement +would become exactly the unresumable id above if the app-server restarted between +the clear and the next message. 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 2941318..2fa5222 100644 --- a/server/src/session/codex.rs +++ b/server/src/session/codex.rs @@ -213,9 +213,9 @@ impl Driver for CodexDriver { // 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 missing_thread = read_thread(&self.inner.session_dir).is_none(); + let waiting_for_protocol = + state.initialize_request.is_some() || state.thread_request.is_some() || missing_thread; let id = (state.running || waiting_for_protocol) .then(super::random_hex) .unwrap_or_default(); @@ -228,6 +228,14 @@ impl Driver for CodexDriver { text: text.clone(), attachments: attachments.clone(), }); + let start_thread = (missing_thread + && state.initialize_request.is_none() + && state.thread_request.is_none()) + .then(|| { + let request = request_id(); + state.thread_request = Some(request.clone()); + request + }); drop(state); save_state(&self.inner); if !id.is_empty() { @@ -237,6 +245,16 @@ impl Driver for CodexDriver { attachments, }); } + if let Some(request) = start_thread { + send_json( + &self.inner, + json!({ + "id": request, + "method": "thread/start", + "params": thread_params(&self.inner) + }), + ); + } dispatch_waiting(&self.inner); } @@ -333,13 +351,11 @@ impl Driver for CodexDriver { }); return; } - let request = request_id(); - self.inner.state.lock().unwrap().thread_request = Some(request.clone()); + // A thread has no durable rollout until its first turn. Creating an empty replacement here + // leaves an id that a restarted app-server cannot resume; start it with the next message + // instead, when Codex can make the thread and its first turn together. + self.inner.state.lock().unwrap().thread_request = None; save_state(&self.inner); - send_json( - &self.inner, - json!({"id": request, "method": "thread/start", "params": thread_params(&self.inner)}), - ); let _ = self.inner.sink.send(Event::Cleared); } @@ -1399,6 +1415,61 @@ mod tests { assert!(!state.waiting[0].id.is_empty()); } + #[test] + fn clear_defers_the_replacement_thread_until_its_first_message() { + let dir = tempfile::tempdir().expect("tempdir"); + write_thread(dir.path(), "old-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::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.clear(); + + assert_eq!(events.try_recv().expect("clear boundary"), Event::Cleared); + assert_eq!(read_thread(dir.path()), None); + assert!( + requests.try_recv().is_err(), + "clear eagerly started a thread" + ); + + driver.send_user_message("first new message".to_string(), Vec::new()); + + assert!(matches!( + events.try_recv().expect("durable queue event"), + Event::MessageQueued { text, .. } if text == "first new message" + )); + let start: Value = serde_json::from_str(&requests.try_recv().expect("thread start")) + .expect("request json"); + assert_eq!(start["method"], "thread/start"); + let request_id = start["id"].as_str().expect("request id"); + + handle_response( + &inner, + &json!({"id": request_id, "result": {"thread": {"id": "new-thread"}}}), + ); + + let turn: Value = + serde_json::from_str(&requests.try_recv().expect("first turn")).expect("request json"); + assert_eq!(turn["method"], "turn/start"); + assert_eq!(turn["params"]["threadId"], "new-thread"); + } + #[test] fn a_replacement_thread_notification_releases_its_waiting_message() { let dir = tempfile::tempdir().expect("tempdir");