systemprompt_api/services/gateway/audit/
mod.rs

1//! Persistence of gateway request lifecycle to the AI-request audit trail.
2//!
3//! [`GatewayAudit`] opens a record when a request arrives (see the `open`
4//! submodule), records the canonical messages and request payload, then closes
5//! it on completion with token usage, resolved cost, latency, captured tool
6//! calls, and the response payload (see the `complete` submodule) — or marks it
7//! failed. [`GatewayRequestContext`] carries the identifiers and routing
8//! metadata bound to a single request.
9//!
10//! Copyright (c) systemprompt.io — Business Source License 1.1.
11//! See <https://systemprompt.io> for licensing details.
12
13mod complete;
14mod message_text;
15mod open;
16pub mod payload;
17
18#[cfg(feature = "test-api")]
19pub mod test_api {
20    pub use super::message_text::flatten_message_content;
21}
22
23use std::sync::{Arc, Mutex};
24use std::time::Instant;
25
26use anyhow::Result;
27use systemprompt_ai::models::RequestStatus;
28use systemprompt_ai::repository::{AiRequestPayloadRepository, AiRequestRepository};
29use systemprompt_identifiers::{
30    AiRequestId, ClientId, ContextId, GatewayConversationId, SessionId, TraceId, UserId,
31};
32use systemprompt_security::policy::types::AccessScope;
33
34/// Method, path, and start instant captured by the gateway access-log
35/// middleware, carried so terminal outcomes can be logged against the same
36/// request line after the response body has finished streaming.
37#[derive(Debug, Clone)]
38pub struct GatewayAccessLog {
39    pub method: String,
40    pub path: String,
41    pub started: Instant,
42}
43
44#[derive(Debug, Clone)]
45pub struct GatewayRequestContext {
46    pub ai_request_id: AiRequestId,
47    pub user_id: UserId,
48    pub session_id: Option<SessionId>,
49    pub context_id: ContextId,
50    pub gateway_conversation_id: Option<GatewayConversationId>,
51    pub trace_id: Option<TraceId>,
52    // Why: governance policies read the caller's tier; an API key carries no
53    // roles and is therefore `Unknown`.
54    pub access_scope: AccessScope,
55    pub client_id: Option<ClientId>,
56    pub provider: String,
57    pub model: String,
58    pub requested_model: Option<String>,
59    pub max_tokens: Option<u32>,
60    pub is_streaming: bool,
61    pub wire_protocol: String,
62    pub access_log: Option<GatewayAccessLog>,
63}
64
65#[expect(
66    missing_debug_implementations,
67    reason = "service type holds repository clients that intentionally do not implement Debug"
68)]
69pub struct GatewayAudit {
70    requests: Arc<AiRequestRepository>,
71    payloads: Arc<AiRequestPayloadRepository>,
72    context_materializer: systemprompt_traits::DynContextMaterializer,
73    pub ctx: GatewayRequestContext,
74    served_model: Mutex<Option<String>>,
75    started_at: Instant,
76}
77
78impl GatewayAudit {
79    pub fn new(repos: &super::GatewayRepositories, ctx: GatewayRequestContext) -> Self {
80        Self {
81            requests: Arc::clone(&repos.requests),
82            payloads: Arc::clone(&repos.payloads),
83            context_materializer: Arc::clone(&repos.context_materializer),
84            ctx,
85            served_model: Mutex::new(None),
86            started_at: Instant::now(),
87        }
88    }
89
90    pub async fn set_served_model(&self, model: &str) {
91        if model.is_empty() || model == self.ctx.model {
92            return;
93        }
94        if let Ok(mut slot) = self.served_model.lock() {
95            *slot = Some(model.to_owned());
96        }
97        if let Err(e) = self
98            .requests
99            .update_model(&self.ctx.ai_request_id, model)
100            .await
101        {
102            tracing::warn!(error = %e, "update_model failed");
103        }
104    }
105
106    pub async fn set_prepared_body_digest(&self, body: &[u8]) {
107        let sha256 = payload::digest_hex(body);
108        if let Err(e) = self
109            .payloads
110            .upsert_prepared_sha256(&self.ctx.ai_request_id, &sha256)
111            .await
112        {
113            tracing::warn!(error = %e, ai_request_id = %self.ctx.ai_request_id, "prepared body digest write failed");
114        }
115    }
116
117    pub async fn set_system_prompt_override(&self, descriptor: &str) {
118        if let Err(e) = self
119            .requests
120            .update_system_prompt_override(&self.ctx.ai_request_id, descriptor)
121            .await
122        {
123            tracing::warn!(error = %e, "update_system_prompt_override failed");
124        }
125    }
126
127    pub async fn set_route_match(&self, descriptor: &str) {
128        if let Err(e) = self
129            .requests
130            .update_route_match(&self.ctx.ai_request_id, descriptor)
131            .await
132        {
133            tracing::warn!(error = %e, "update_route_match failed");
134        }
135    }
136
137    pub async fn fail(&self, error: &str) -> Result<()> {
138        let latency_ms = self.elapsed_ms();
139        if let Err(e) = self
140            .requests
141            .update_error(&self.ctx.ai_request_id, RequestStatus::Failed, error)
142            .await
143        {
144            tracing::warn!(error = %e, "audit fail update failed");
145        }
146        // Why: `update_error` writes no usage columns, so a failed row keeps the
147        // zeros it was opened with — say so, or the reader takes them as "nothing
148        // was consumed" when a partial stream may well have been billed upstream.
149        tracing::warn!(
150            ai_request_id = %self.ctx.ai_request_id,
151            user_id = %self.ctx.user_id,
152            provider = %self.ctx.provider,
153            model = %self.effective_model(),
154            requested_model = %self.ctx.model,
155            wire_protocol = %self.ctx.wire_protocol,
156            status = RequestStatus::Failed.as_str(),
157            latency_ms,
158            tokens_recorded = false,
159            error,
160            "Gateway audit: request failed"
161        );
162        Ok(())
163    }
164
165    pub(crate) fn elapsed_ms(&self) -> i32 {
166        self.started_at.elapsed().as_millis().min(i32::MAX as u128) as i32
167    }
168}
systemprompt_api/services/gateway/audit/mod.rs

systemprompt_api/services/gateway/audit/
mod.rs
