Make thread unsubscribe test deterministic (#18000)

## Summary
- replace the unsubscribe-during-turn test's sleep/polling flow with a
gated streaming SSE response
- add request-count notification support to the streaming SSE test
server so the test can wait for the in-flight Responses request
deterministically

## Scope
- codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs
- codex-rs/core/tests/common/streaming_sse.rs

## Validation
- Not run locally; this is a narrow extraction from the prior CI-green
branch.

---------

Co-authored-by: Codex <noreply@openai.com>

codex-rs/core/tests/common/streaming_sse.rs

@@ -7,6 +7,7 @@ use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
 use tokio::net::TcpListener;
 use tokio::sync::Mutex as TokioMutex;
+use tokio::sync::Notify;
 use tokio::sync::oneshot;
 
 /// Streaming SSE chunk payload gated by a per-chunk signal.
@@ -20,6 +21,7 @@ pub struct StreamingSseChunk {
 pub struct StreamingSseServer {
     uri: String,
     requests: Arc<TokioMutex<Vec<Vec<u8>>>>,
+    request_notify: Arc<Notify>,
     shutdown: oneshot::Sender<()>,
     task: tokio::task::JoinHandle<()>,
 }
@@ -33,6 +35,15 @@ impl StreamingSseServer {
         self.requests.lock().await.clone()
     }
 
+    pub async fn wait_for_request_count(&self, count: usize) {
+        loop {
+            if self.requests.lock().await.len() >= count {
+                return;
+            }
+            self.request_notify.notified().await;
+        }
+    }
+
     pub async fn shutdown(self) {
         let _ = self.shutdown.send(());
         let _ = self.task.await;
@@ -67,7 +78,9 @@ pub async fn start_streaming_sse_server(
         completions: VecDeque::from(completion_senders),
     }));
     let requests = Arc::new(TokioMutex::new(Vec::new()));
+    let request_notify = Arc::new(Notify::new());
     let requests_for_task = Arc::clone(&requests);
+    let request_notify_for_task = Arc::clone(&request_notify);
     let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
 
     let task = tokio::spawn(async move {
@@ -78,6 +91,7 @@ pub async fn start_streaming_sse_server(
                     let (mut stream, _) = accept_res.expect("accept streaming SSE connection");
                     let state = Arc::clone(&state);
                     let requests = Arc::clone(&requests_for_task);
+                    let request_notify = Arc::clone(&request_notify_for_task);
                     tokio::spawn(async move {
                         let (request, body_prefix) = read_http_request(&mut stream).await;
                         let Some((method, path)) = parse_request_line(&request) else {
@@ -113,6 +127,7 @@ pub async fn start_streaming_sse_server(
                                 }
                             };
                             requests.lock().await.push(body);
+                            request_notify.notify_one();
                             let Some((chunks, completion)) = take_next_stream(&state).await else {
                                 let _ = write_http_response(&mut stream, /*status*/ 500, "no responses queued", "text/plain").await;
                                 return;
@@ -149,6 +164,7 @@ pub async fn start_streaming_sse_server(
         StreamingSseServer {
             uri,
             requests,
+            request_notify,
             shutdown: shutdown_tx,
             task,
         },