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, ContextId, GatewayConversationId, SessionId, TraceId, UserId,
31};
32use systemprompt_security::policy::types::AccessScope;
33
34#[derive(Debug, Clone)]
35pub struct GatewayRequestContext {
36    pub ai_request_id: AiRequestId,
37    pub user_id: UserId,
38    pub session_id: Option<SessionId>,
39    pub context_id: ContextId,
40    pub gateway_conversation_id: Option<GatewayConversationId>,
41    pub trace_id: Option<TraceId>,
42    // Why: governance policies read the caller's tier; an API key carries no
43    // roles and is therefore `Unknown`.
44    pub access_scope: AccessScope,
45    pub provider: String,
46    pub model: String,
47    pub requested_model: Option<String>,
48    pub max_tokens: Option<u32>,
49    pub is_streaming: bool,
50    pub wire_protocol: String,
51}
52
53#[expect(
54    missing_debug_implementations,
55    reason = "service type holds repository clients that intentionally do not implement Debug"
56)]
57pub struct GatewayAudit {
58    requests: Arc<AiRequestRepository>,
59    payloads: Arc<AiRequestPayloadRepository>,
60    context_materializer: systemprompt_traits::DynContextMaterializer,
61    pub ctx: GatewayRequestContext,
62    served_model: Mutex<Option<String>>,
63    started_at: Instant,
64}
65
66impl GatewayAudit {
67    pub fn new(repos: &super::GatewayRepositories, ctx: GatewayRequestContext) -> Self {
68        Self {
69            requests: Arc::clone(&repos.requests),
70            payloads: Arc::clone(&repos.payloads),
71            context_materializer: Arc::clone(&repos.context_materializer),
72            ctx,
73            served_model: Mutex::new(None),
74            started_at: Instant::now(),
75        }
76    }
77
78    pub async fn set_served_model(&self, model: &str) {
79        if model.is_empty() || model == self.ctx.model {
80            return;
81        }
82        if let Ok(mut slot) = self.served_model.lock() {
83            *slot = Some(model.to_owned());
84        }
85        if let Err(e) = self
86            .requests
87            .update_model(&self.ctx.ai_request_id, model)
88            .await
89        {
90            tracing::warn!(error = %e, "update_model failed");
91        }
92    }
93
94    pub async fn set_prepared_body_digest(&self, body: &[u8]) {
95        let sha256 = payload::digest_hex(body);
96        if let Err(e) = self
97            .payloads
98            .upsert_prepared_sha256(&self.ctx.ai_request_id, &sha256)
99            .await
100        {
101            tracing::warn!(error = %e, ai_request_id = %self.ctx.ai_request_id, "prepared body digest write failed");
102        }
103    }
104
105    pub async fn set_system_prompt_override(&self, descriptor: &str) {
106        if let Err(e) = self
107            .requests
108            .update_system_prompt_override(&self.ctx.ai_request_id, descriptor)
109            .await
110        {
111            tracing::warn!(error = %e, "update_system_prompt_override failed");
112        }
113    }
114
115    pub async fn set_route_match(&self, descriptor: &str) {
116        if let Err(e) = self
117            .requests
118            .update_route_match(&self.ctx.ai_request_id, descriptor)
119            .await
120        {
121            tracing::warn!(error = %e, "update_route_match failed");
122        }
123    }
124
125    pub async fn fail(&self, error: &str) -> Result<()> {
126        if let Err(e) = self
127            .requests
128            .update_error(&self.ctx.ai_request_id, RequestStatus::Failed, error)
129            .await
130        {
131            tracing::warn!(error = %e, "audit fail update failed");
132        }
133        tracing::info!(
134            ai_request_id = %self.ctx.ai_request_id,
135            user_id = %self.ctx.user_id,
136            provider = %self.ctx.provider,
137            model = %self.ctx.model,
138            error,
139            "Gateway audit: request failed"
140        );
141        Ok(())
142    }
143}
systemprompt_api/services/gateway/audit/mod.rs

systemprompt_api/services/gateway/audit/
mod.rs
