feat(tracing): tag app-server turn spans with turn_id (#15206)

So we can find and filter spans by `turn.id`.

We do this for the `turn/start`, `turn/steer`, and `turn/interrupt`
APIs.

Author: owenlin0

codex-rs/app-server/src/app_server_tracing.rs

@@ -107,6 +107,7 @@ fn app_server_request_span_template(
         app_server.api_version = "v2",
         app_server.client_name = field::Empty,
         app_server.client_version = field::Empty,
+        turn.id = field::Empty,
     )
 }
 

codex-rs/app-server/src/codex_message_processor.rs

@@ -6024,6 +6024,9 @@ impl CodexMessageProcessor {
 
         match turn_id {
             Ok(turn_id) => {
+                self.outgoing
+                    .record_request_turn_id(&request_id, &turn_id)
+                    .await;
                 let turn = Turn {
                     id: turn_id.clone(),
                     items: vec![],
@@ -6076,6 +6079,9 @@ impl CodexMessageProcessor {
             .await;
             return;
         }
+        self.outgoing
+            .record_request_turn_id(&request_id, &params.expected_turn_id)
+            .await;
         if let Err(error) = Self::validate_v2_input_limit(&params.input) {
             self.outgoing.send_error(request_id, error).await;
             return;
@@ -6556,7 +6562,10 @@ impl CodexMessageProcessor {
         request_id: ConnectionRequestId,
         params: TurnInterruptParams,
     ) {
-        let TurnInterruptParams { thread_id, .. } = params;
+        let TurnInterruptParams { thread_id, turn_id } = params;
+        self.outgoing
+            .record_request_turn_id(&request_id, &turn_id)
+            .await;
 
         let (thread_uuid, thread) = match self.load_thread(&thread_id).await {
             Ok(v) => v,

codex-rs/app-server/src/message_processor/tracing_tests.rs

@@ -580,7 +580,7 @@ async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> {
         parent_span_id: remote_parent_span_id,
         context: remote_trace,
     } = RemoteTrace::new("00000000000000000000000000000077", "0000000000000088");
-    let _: TurnStartResponse = harness
+    let turn_start_response: TurnStartResponse = harness
         .request(
             ClientRequest::TurnStart {
                 request_id: RequestId::Integer(3),
@@ -628,6 +628,10 @@ async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> {
     assert_eq!(server_request_span.parent_span_id, remote_parent_span_id);
     assert!(server_request_span.parent_span_is_remote);
     assert_eq!(server_request_span.span_context.trace_id(), remote_trace_id);
+    assert_eq!(
+        span_attr(server_request_span, "turn.id"),
+        Some(turn_start_response.turn.id.as_str())
+    );
     assert_span_descends_from(&spans, core_turn_span, server_request_span);
     harness.shutdown().await;
 

codex-rs/app-server/src/outgoing_message.rs

@@ -75,6 +75,10 @@ impl RequestContext {
     pub(crate) fn span(&self) -> Span {
         self.span.clone()
     }
+
+    fn record_turn_id(&self, turn_id: &str) {
+        self.span.record("turn.id", turn_id);
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -217,6 +221,17 @@ impl OutgoingMessageSender {
             .and_then(RequestContext::request_trace)
     }
 
+    pub(crate) async fn record_request_turn_id(
+        &self,
+        request_id: &ConnectionRequestId,
+        turn_id: &str,
+    ) {
+        let request_contexts = self.request_contexts.lock().await;
+        if let Some(request_context) = request_contexts.get(request_id) {
+            request_context.record_turn_id(turn_id);
+        }
+    }
+
     async fn take_request_context(
         &self,
         request_id: &ConnectionRequestId,