From ed116c2b2cd6e8af160f1f256419f7a71655d1bc Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 9 Jul 2026 04:09:20 +0000 Subject: [PATCH] Record app-server RPC completion outcomes --- codex-rs/app-server/src/app_server_tracing.rs | 2 + .../src/message_processor_tracing_tests.rs | 8 +++ codex-rs/app-server/src/outgoing_message.rs | 64 +++++++++++++++++-- 3 files changed, 68 insertions(+), 6 deletions(-) diff --git a/codex-rs/app-server/src/app_server_tracing.rs b/codex-rs/app-server/src/app_server_tracing.rs index 6e8133740f..119fc281d1 100644 --- a/codex-rs/app-server/src/app_server_tracing.rs +++ b/codex-rs/app-server/src/app_server_tracing.rs @@ -109,6 +109,8 @@ fn app_server_request_span_template( app_server.api_version = "v2", app_server.client_name = field::Empty, app_server.client_version = field::Empty, + rpc.result = field::Empty, + rpc.error_code = field::Empty, turn.id = field::Empty, ) } diff --git a/codex-rs/app-server/src/message_processor_tracing_tests.rs b/codex-rs/app-server/src/message_processor_tracing_tests.rs index 7c4ae8c985..08486c0f89 100644 --- a/codex-rs/app-server/src/message_processor_tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor_tracing_tests.rs @@ -615,6 +615,10 @@ fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() -> Resul assert!(server_request_span.parent_span_is_remote); assert_eq!(server_request_span.span_context.trace_id(), remote_trace_id); assert_ne!(server_request_span.span_context.span_id(), SpanId::INVALID); + assert_eq!( + span_attr(server_request_span, "rpc.result"), + Some("success") + ); assert_has_internal_descendant_at_min_depth( &spans, server_request_span, @@ -705,6 +709,10 @@ async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> { span_attr(server_request_span, "turn.id"), Some(turn_start_response.turn.id.as_str()) ); + assert_eq!( + span_attr(server_request_span, "rpc.result"), + Some("success") + ); assert_span_descends_from(&spans, core_turn_span, server_request_span); harness.shutdown().await; diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 16d5c2ff9d..760b77f018 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -38,6 +38,25 @@ use codex_protocol::account::PlanType; pub(crate) type ClientRequestResult = std::result::Result; +#[derive(Clone, Copy)] +enum RpcResult { + Success, + Error, + Disconnected, + Replaced, +} + +impl RpcResult { + const fn as_str(self) -> &'static str { + match self { + Self::Success => "success", + Self::Error => "error", + Self::Disconnected => "disconnected", + Self::Replaced => "replaced", + } + } +} + /// Stable identifier for a client request scoped to a transport connection. #[derive(Clone, Debug, Eq, Hash, PartialEq)] pub(crate) struct ConnectionRequestId { @@ -78,6 +97,14 @@ impl RequestContext { fn record_turn_id(&self, turn_id: &str) { self.span.record("turn.id", turn_id); } + + fn record_result(&self, result: RpcResult) { + self.span.record("rpc.result", result.as_str()); + } + + fn record_error_code(&self, error: &JSONRPCErrorError) { + self.span.record("rpc.error_code", error.code); + } } #[derive(Debug)] @@ -222,16 +249,21 @@ impl OutgoingMessageSender { pub(crate) async fn register_request_context(&self, request_context: RequestContext) { let mut request_contexts = self.request_contexts.lock().await; - if request_contexts - .insert(request_context.request_id.clone(), request_context) - .is_some() + if let Some(replaced) = + request_contexts.insert(request_context.request_id.clone(), request_context) { + replaced.record_result(RpcResult::Replaced); warn!("replaced unresolved request context"); } } pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) { let mut request_contexts = self.request_contexts.lock().await; + for (request_id, request_context) in request_contexts.iter() { + if request_id.connection_id == connection_id { + request_context.record_result(RpcResult::Disconnected); + } + } request_contexts.retain(|request_id, _| request_id.connection_id != connection_id); } @@ -682,6 +714,9 @@ impl OutgoingMessageSender { request_id: ConnectionRequestId, error: JSONRPCErrorError, ) { + if let Some(request_context) = request_context.as_ref() { + request_context.record_error_code(&error); + } let outgoing_message = OutgoingMessage::Error(OutgoingError { id: request_id.request_id, error, @@ -702,19 +737,36 @@ impl OutgoingMessageSender { message: OutgoingMessage, message_kind: &'static str, ) { + let completed_result = match &message { + OutgoingMessage::Response(_) => Some(RpcResult::Success), + OutgoingMessage::Error(_) => Some(RpcResult::Error), + OutgoingMessage::Request(_) | OutgoingMessage::AppServerNotification(_) => None, + }; let send_fut = self.sender.send(OutgoingEnvelope::ToConnection { connection_id, message, write_complete_tx: None, }); - let send_result = if let Some(request_context) = request_context { + let send_result = if let Some(request_context) = request_context.as_ref() { send_fut.instrument(request_context.span()).await } else { send_fut.await }; - if let Err(err) = send_result { - warn!("failed to send {message_kind} to client: {err:?}"); + match send_result { + Ok(()) => { + if let (Some(request_context), Some(result)) = + (request_context.as_ref(), completed_result) + { + request_context.record_result(result); + } + } + Err(err) => { + if let Some(request_context) = request_context.as_ref() { + request_context.record_result(RpcResult::Disconnected); + } + warn!("failed to send {message_kind} to client: {err:?}"); + } } } }