Release messages after Codex recovery

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<Inner>, 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);
+    }
 }