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_database::DbPool;
30use systemprompt_identifiers::{
31    AiRequestId, ContextId, GatewayConversationId, SessionId, TraceId, UserId,
32};
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    pub provider: String,
43    /// The upstream model the request dispatches to (after route rewrite). The
44    /// audit `model` column is opened from this, then overwritten by
45    /// `set_served_model` with the model the provider echoes back.
46    pub model: String,
47    /// The model the client requested on the wire, before route rewrite.
48    /// Persisted to `ai_requests.requested_model` so an audit retains both.
49    pub requested_model: Option<String>,
50    pub max_tokens: Option<u32>,
51    pub is_streaming: bool,
52    pub wire_protocol: String,
53}
54
55#[expect(
56    missing_debug_implementations,
57    reason = "service type holds repository clients that intentionally do not implement Debug"
58)]
59pub struct GatewayAudit {
60    requests: Arc<AiRequestRepository>,
61    payloads: Arc<AiRequestPayloadRepository>,
62    pub ctx: GatewayRequestContext,
63    served_model: Mutex<Option<String>>,
64    started_at: Instant,
65}
66
67impl GatewayAudit {
68    pub fn new(
69        db: &DbPool,
70        ctx: GatewayRequestContext,
71    ) -> Result<Self, systemprompt_ai::error::RepositoryError> {
72        let requests = Arc::new(AiRequestRepository::new(db)?);
73        let payloads = Arc::new(AiRequestPayloadRepository::new(db)?);
74        Ok(Self {
75            requests,
76            payloads,
77            ctx,
78            served_model: Mutex::new(None),
79            started_at: Instant::now(),
80        })
81    }
82
83    pub async fn set_served_model(&self, model: &str) {
84        if model.is_empty() || model == self.ctx.model {
85            return;
86        }
87        if let Ok(mut slot) = self.served_model.lock() {
88            *slot = Some(model.to_owned());
89        }
90        if let Err(e) = self
91            .requests
92            .update_model(&self.ctx.ai_request_id, model)
93            .await
94        {
95            tracing::warn!(error = %e, "update_model failed");
96        }
97    }
98
99    pub async fn set_system_prompt_override(&self, descriptor: &str) {
100        if let Err(e) = self
101            .requests
102            .update_system_prompt_override(&self.ctx.ai_request_id, descriptor)
103            .await
104        {
105            tracing::warn!(error = %e, "update_system_prompt_override failed");
106        }
107    }
108
109    pub async fn set_route_match(&self, descriptor: &str) {
110        if let Err(e) = self
111            .requests
112            .update_route_match(&self.ctx.ai_request_id, descriptor)
113            .await
114        {
115            tracing::warn!(error = %e, "update_route_match failed");
116        }
117    }
118
119    pub async fn fail(&self, error: &str) -> Result<()> {
120        if let Err(e) = self
121            .requests
122            .update_error(&self.ctx.ai_request_id, RequestStatus::Failed, error)
123            .await
124        {
125            tracing::warn!(error = %e, "audit fail update failed");
126        }
127        tracing::info!(
128            ai_request_id = %self.ctx.ai_request_id,
129            user_id = %self.ctx.user_id,
130            provider = %self.ctx.provider,
131            model = %self.ctx.model,
132            error,
133            "Gateway audit: request failed"
134        );
135        Ok(())
136    }
137}
systemprompt_api/services/gateway/audit/mod.rs

systemprompt_api/services/gateway/audit/
mod.rs
