diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index a6e851f1d7..d0289b1f87 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -2469,22 +2469,28 @@ async fn thread_resume_defers_updated_at_until_turn_start() -> Result<()> { ) .await??; - let read_id = mcp - .send_thread_read_request(ThreadReadParams { - thread_id: thread_id.clone(), - include_turns: false, - }) - .await?; - let read_resp: JSONRPCResponse = timeout( - DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_response_message(RequestId::Integer(read_id)), - ) + timeout(DEFAULT_READ_TIMEOUT, async { + loop { + let read_id = mcp + .send_thread_read_request(ThreadReadParams { + thread_id: thread_id.clone(), + include_turns: false, + }) + .await?; + let read_resp: JSONRPCResponse = mcp + .read_stream_until_response_message(RequestId::Integer(read_id)) + .await?; + let ThreadReadResponse { + thread: after_turn_start, + .. + } = to_response::(read_resp)?; + if after_turn_start.recency_at > before_resume.recency_at { + return Ok::<(), anyhow::Error>(()); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) .await??; - let ThreadReadResponse { - thread: after_turn_start, - .. - } = to_response::(read_resp)?; - assert!(after_turn_start.recency_at > before_resume.recency_at); timeout( DEFAULT_READ_TIMEOUT, diff --git a/codex-rs/thread-store/src/live_thread_tests.rs b/codex-rs/thread-store/src/live_thread_tests.rs index 8554225edf..47e74f7412 100644 --- a/codex-rs/thread-store/src/live_thread_tests.rs +++ b/codex-rs/thread-store/src/live_thread_tests.rs @@ -323,7 +323,7 @@ async fn queued_generation_keeps_its_barrier_when_in_flight_update_fails() { #[derive(Clone)] struct WorkerSpanParentLayer { - parent_name: Arc>>, + parent_name: Arc>>>, } impl Layer for WorkerSpanParentLayer @@ -342,23 +342,46 @@ where if span.metadata().name() != "thread_store.metadata_update_worker" { return; } - *self.parent_name.lock().expect("parent name lock") = span - .parent() - .map(|parent| parent.metadata().name().to_string()); + *self.parent_name.lock().expect("parent name lock") = Some( + span.parent() + .map(|parent| parent.metadata().name().to_string()), + ); } } #[tokio::test(flavor = "current_thread")] async fn metadata_worker_span_preserves_append_trace() { - let (store, mut updates) = ControlledThreadStore::new(); - let live_thread = create_live_thread(Arc::clone(&store)).await; let parent_name = Arc::new(StdMutex::new(None)); let _guard = tracing_subscriber::registry() .with(WorkerSpanParentLayer { parent_name: Arc::clone(&parent_name), }) .set_default(); + + // Register both production callsites before rebuilding the process-global interest cache. + // The subscriber must be active first because tracing otherwise short-circuits at level OFF. + { + let (store, mut updates) = ControlledThreadStore::new(); + let live_thread = create_live_thread(Arc::clone(&store)).await; + live_thread + .append_items(&[user_message("warm up tracing callsites")]) + .await + .expect("warm-up append should succeed"); + next_update(&mut updates) + .await + .completion + .send(Ok(())) + .expect("warm-up metadata worker should still be waiting"); + live_thread + .flush() + .await + .expect("warm-up flush should drain metadata"); + } tracing::callsite::rebuild_interest_cache(); + *parent_name.lock().expect("parent name lock") = None; + + let (store, mut updates) = ControlledThreadStore::new(); + let live_thread = create_live_thread(Arc::clone(&store)).await; live_thread .append_items(&[user_message("trace me")]) @@ -374,9 +397,10 @@ async fn metadata_worker_span_preserves_append_trace() { .await .expect("flush should drain metadata"); + let parent_name = parent_name.lock().expect("parent name lock"); assert_eq!( - parent_name.lock().expect("parent name lock").as_deref(), - Some("thread_store.live_append.observe_metadata") + parent_name.as_ref().map(Option::as_deref), + Some(Some("thread_store.live_append.observe_metadata")) ); }