From 88033a3b3e496808d18ecc94996129a4fae5a1b6 Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Sun, 19 Apr 2026 22:10:29 +0000 Subject: [PATCH] api: propagate request ids through admin test runs --- apps/admin-api/src/app.rs | 109 ++++++++++++++++++++++++ apps/admin-api/src/main.rs | 1 + apps/admin-api/src/request_context.rs | 64 ++++++++++++++ apps/admin-api/src/routes/operations.rs | 5 +- apps/admin-api/src/service.rs | 11 ++- 5 files changed, 188 insertions(+), 2 deletions(-) create mode 100644 apps/admin-api/src/request_context.rs diff --git a/apps/admin-api/src/app.rs b/apps/admin-api/src/app.rs index f3bef6f..443b945 100644 --- a/apps/admin-api/src/app.rs +++ b/apps/admin-api/src/app.rs @@ -5,6 +5,7 @@ use axum::{ use crate::{ auth::{require_session, require_workspace_session}, + request_context::apply_request_context, routes::{ access::{ create_invitation, create_platform_api_key, delete_invitation, delete_membership, @@ -205,6 +206,7 @@ pub fn build_app(state: AppState) -> Router { .merge(protected_auth_router), ) .nest("/api/admin", admin_router) + .layer(middleware::from_fn(apply_request_context)) .with_state(state) } @@ -1306,6 +1308,113 @@ mod tests { assert_eq!(operation_usage["rollup"]["calls_total"], 1); } + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn preserves_request_id_for_test_run_invocations() { + let registry = test_registry().await; + let storage_root = test_storage_root("observability_request_id"); + let upstream_base_url = spawn_upstream_server().await; + let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await; + let client = authorized_client(&base_url).await; + + let created = client + .post(format!("{base_url}/operations")) + .json(&test_operation_payload( + &upstream_base_url, + "crm_observability_request_id", + )) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let operation_id = created["operation_id"].as_str().unwrap().to_owned(); + + let response = client + .post(format!("{base_url}/operations/{operation_id}/test-runs")) + .header("x-request-id", "req_test_123") + .json(&json!({ + "version": 1, + "input": { "email": "user@example.com" } + })) + .send() + .await + .unwrap(); + + assert_eq!( + response.headers()["x-request-id"].to_str().unwrap(), + "req_test_123" + ); + response.error_for_status().unwrap(); + + let logs = client + .get(format!("{base_url}/logs?period=7d")) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + + assert_eq!(logs["items"][0]["log"]["request_id"], "req_test_123"); + } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn generates_request_id_for_test_run_invocations() { + let registry = test_registry().await; + let storage_root = test_storage_root("observability_generated_request_id"); + let upstream_base_url = spawn_upstream_server().await; + let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await; + let client = authorized_client(&base_url).await; + + let created = client + .post(format!("{base_url}/operations")) + .json(&test_operation_payload( + &upstream_base_url, + "crm_observability_generated_request_id", + )) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let operation_id = created["operation_id"].as_str().unwrap().to_owned(); + + let response = client + .post(format!("{base_url}/operations/{operation_id}/test-runs")) + .json(&json!({ + "version": 1, + "input": { "email": "user@example.com" } + })) + .send() + .await + .unwrap(); + let request_id = response + .headers() + .get("x-request-id") + .unwrap() + .to_str() + .unwrap() + .to_owned(); + + assert!(!request_id.is_empty()); + response.error_for_status().unwrap(); + + let logs = client + .get(format!("{base_url}/logs?period=7d")) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + + assert_eq!(logs["items"][0]["log"]["request_id"], request_id); + } + #[tokio::test(flavor = "multi_thread")] #[serial] async fn creates_publishes_and_tests_graphql_operation() { diff --git a/apps/admin-api/src/main.rs b/apps/admin-api/src/main.rs index e01d7f5..ce23057 100644 --- a/apps/admin-api/src/main.rs +++ b/apps/admin-api/src/main.rs @@ -1,6 +1,7 @@ mod app; mod auth; mod error; +mod request_context; mod routes; mod service; mod state; diff --git a/apps/admin-api/src/request_context.rs b/apps/admin-api/src/request_context.rs new file mode 100644 index 0000000..3b1ec06 --- /dev/null +++ b/apps/admin-api/src/request_context.rs @@ -0,0 +1,64 @@ +use axum::{ + extract::Request, + http::{HeaderMap, HeaderName, HeaderValue}, + middleware::Next, + response::Response, +}; +use uuid::Uuid; + +pub const REQUEST_ID_HEADER: HeaderName = HeaderName::from_static("x-request-id"); +const MAX_REQUEST_ID_LEN: usize = 128; + +#[derive(Clone, Debug)] +pub struct RequestContext { + pub request_id: String, +} + +pub async fn apply_request_context(mut request: Request, next: Next) -> Response { + let context = RequestContext { + request_id: resolve_request_id(request.headers()), + }; + request.extensions_mut().insert(context.clone()); + + let mut response = next.run(request).await; + if let Ok(value) = HeaderValue::from_str(&context.request_id) { + response.headers_mut().insert(REQUEST_ID_HEADER, value); + } + response +} + +fn resolve_request_id(headers: &HeaderMap) -> String { + headers + .get(&REQUEST_ID_HEADER) + .and_then(|value| value.to_str().ok()) + .map(str::trim) + .filter(|value| is_valid_request_id(value)) + .map(ToOwned::to_owned) + .unwrap_or_else(|| Uuid::now_v7().to_string()) +} + +fn is_valid_request_id(value: &str) -> bool { + !value.is_empty() + && value.len() <= MAX_REQUEST_ID_LEN + && value + .bytes() + .all(|byte| matches!(byte, 0x21..=0x7e) && byte != b',' && byte != b';') +} + +#[cfg(test)] +mod tests { + use super::is_valid_request_id; + + #[test] + fn accepts_visible_ascii_request_ids() { + assert!(is_valid_request_id("req_test_123")); + assert!(is_valid_request_id("trace-123/abc")); + } + + #[test] + fn rejects_empty_or_control_request_ids() { + assert!(!is_valid_request_id("")); + assert!(!is_valid_request_id("bad value")); + assert!(!is_valid_request_id("bad\nvalue")); + } +} diff --git a/apps/admin-api/src/routes/operations.rs b/apps/admin-api/src/routes/operations.rs index 68239ab..7c32f99 100644 --- a/apps/admin-api/src/routes/operations.rs +++ b/apps/admin-api/src/routes/operations.rs @@ -1,7 +1,7 @@ use axum::{ Json, body::Bytes, - extract::{Path, Query, State}, + extract::{Extension, Path, Query, State}, http::{HeaderMap, StatusCode, header}, response::IntoResponse, }; @@ -10,6 +10,7 @@ use serde_json::{Value, json}; use crate::{ error::ApiError, + request_context::RequestContext, service::{ ExportQuery, GenerateDraftPayload, ImportQuery, NewVersionPayload, OperationPayload, PublishPayload, TestRunPayload, UpdateOperationPayload, @@ -172,6 +173,7 @@ pub async fn archive_operation( pub async fn run_test( Path(path): Path, State(state): State, + Extension(request_context): Extension, Json(payload): Json, ) -> Result, ApiError> { let result = state @@ -180,6 +182,7 @@ pub async fn run_test( &path.workspace_id.as_str().into(), &path.operation_id.as_str().into(), payload, + &request_context.request_id, ) .await?; Ok(Json(json!(result))) diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index eef5d4c..95d6394 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -594,6 +594,7 @@ struct InvocationRecordRequest<'a> { workspace_id: &'a WorkspaceId, agent_id: Option<&'a AgentId>, operation: &'a RegistryOperation, + request_id: Option<&'a str>, source: InvocationSource, level: InvocationLevel, status: InvocationStatus, @@ -2093,6 +2094,7 @@ impl AdminService { workspace_id: &WorkspaceId, operation_id: &OperationId, payload: TestRunPayload, + request_id: &str, ) -> Result { let record = self .get_operation_version(workspace_id, operation_id, payload.version) @@ -2112,6 +2114,7 @@ impl AdminService { workspace_id, agent_id: None, operation: &record.snapshot, + request_id: Some(request_id), source: InvocationSource::AdminTestRun, level: InvocationLevel::Error, status: InvocationStatus::Error, @@ -2180,6 +2183,7 @@ impl AdminService { workspace_id, agent_id: None, operation: &record.snapshot, + request_id: Some(request_id), source: InvocationSource::AdminTestRun, level: InvocationLevel::Info, status: InvocationStatus::Ok, @@ -2209,6 +2213,7 @@ impl AdminService { workspace_id, agent_id: None, operation: &record.snapshot, + request_id: Some(request_id), source: InvocationSource::AdminTestRun, level: InvocationLevel::Error, status: InvocationStatus::Error, @@ -4194,6 +4199,7 @@ impl AdminService { workspace_id, agent_id: Some(revops_agent_id), operation: &rest_operation.snapshot, + request_id: None, source: InvocationSource::AgentToolCall, level: InvocationLevel::Info, status: InvocationStatus::Ok, @@ -4216,6 +4222,7 @@ impl AdminService { workspace_id, agent_id: Some(revops_agent_id), operation: &graphql_operation.snapshot, + request_id: None, source: InvocationSource::AgentToolCall, level: InvocationLevel::Info, status: InvocationStatus::Ok, @@ -4238,6 +4245,7 @@ impl AdminService { workspace_id, agent_id: Some(revops_agent_id), operation: &graphql_operation.snapshot, + request_id: None, source: InvocationSource::AgentToolCall, level: InvocationLevel::Error, status: InvocationStatus::Error, @@ -4262,6 +4270,7 @@ impl AdminService { workspace_id, agent_id: None, operation: &grpc_operation.snapshot, + request_id: None, source: InvocationSource::AdminTestRun, level: InvocationLevel::Info, status: InvocationStatus::Ok, @@ -4298,7 +4307,7 @@ impl AdminService { status: request.status, tool_name: request.operation.name.clone(), message: request.message, - request_id: None, + request_id: request.request_id.map(ToOwned::to_owned), status_code: request.status_code, duration_ms: request.duration_ms, error_kind: request.error_kind,