sim_lib_openai_server/routes/
execution_record.rs

1//! Shared substrate for the gateway execution-record engines (OVERLAP9.04).
2//!
3//! The `/v1/responses`, `/v1/embeddings`, and run_record routes each drive the
4//! same content-addressed ledger: mint a request id, redact and record the
5//! request; open a run; append a sequence of events; store a final response.
6//! Before OVERLAP9.04 that machinery was forked three ways (three id-generator
7//! trios, three `EventLog`s, three `append_event` bodies, three request+run
8//! prologues, three outcome structs). This module owns the ONE copy behind the
9//! OVERLAP9.03 goldens.
10//!
11//! What stays route-local (and must, to keep the goldens byte-identical): the
12//! object-vs-bare response storage split (`put_response_object` for responses
13//! vs `put_response` for embeddings/run_record), the per-route event sequence
14//! and kinds, and each route's `store` default.
15
16use sim_kernel::{ContentId, Expr, Symbol};
17
18use crate::{
19    clock::GatewayClock,
20    content_id::{content_id_for_expr, request_content_id},
21    ids::GatewayIdGenerator,
22    objects::{GatewayEvent, GatewayRequest, GatewayResponse, GatewayRun},
23    runtime::redacted_gateway_request,
24    storage::GatewayStore,
25};
26
27use super::errors::OpenAiRouteError;
28
29type RouteResult<T> = std::result::Result<T, OpenAiRouteError>;
30
31/// Per-kind id generators for a single gateway run.
32///
33/// All three engines seed the request/run/event generators identically
34/// (`gwreq`/`gwrun`/`gwevt` from the same start), so one struct serves them
35/// all. The extra `response` generator (`resp`) is only advanced by the
36/// `/v1/responses` engine, which mints a public response id; the other engines
37/// simply never touch it.
38#[derive(Clone, Debug)]
39pub struct GatewayRunIdGenerators {
40    request: GatewayIdGenerator,
41    run: GatewayIdGenerator,
42    /// Response id generator, used only by the `/v1/responses` engine.
43    pub(crate) response: GatewayIdGenerator,
44    event: GatewayIdGenerator,
45}
46
47impl GatewayRunIdGenerators {
48    /// Builds id generators seeded deterministically from `start`, so a given
49    /// seed always yields the same id sequence.
50    pub fn deterministic(start: u64) -> Self {
51        Self {
52            request: GatewayIdGenerator::deterministic("gwreq", start),
53            run: GatewayIdGenerator::deterministic("gwrun", start),
54            response: GatewayIdGenerator::deterministic("resp", start),
55            event: GatewayIdGenerator::deterministic("gwevt", start),
56        }
57    }
58
59    pub(crate) fn next_event_id(&mut self) -> sim_kernel::Result<String> {
60        self.event.next_id()
61    }
62}
63
64/// One event appended to a run: its sequence, kind, and payload.
65pub(crate) struct EventInput {
66    sequence: u64,
67    kind: Symbol,
68    payload: Expr,
69}
70
71impl EventInput {
72    pub(crate) fn new(sequence: u64, kind: &'static str, payload: Expr) -> Self {
73        Self::from_symbol(sequence, Symbol::new(kind), payload)
74    }
75
76    pub(crate) fn from_symbol(sequence: u64, kind: Symbol, payload: Expr) -> Self {
77        Self {
78            sequence,
79            kind,
80            payload,
81        }
82    }
83}
84
85/// Accumulates the content ids and events produced while a run executes.
86#[derive(Default)]
87pub(crate) struct EventLog {
88    pub(crate) content_ids: Vec<ContentId>,
89    pub(crate) events: Vec<GatewayEvent>,
90}
91
92/// Mints an event id, builds the [`GatewayEvent`], records it under its content
93/// id when `store_event` is set, and appends both to `event_log`.
94pub(crate) fn append_event<S, C>(
95    store: &mut S,
96    ids: &mut GatewayRunIdGenerators,
97    clock: &mut C,
98    run_id: &str,
99    input: EventInput,
100    store_event: bool,
101    event_log: &mut EventLog,
102) -> RouteResult<()>
103where
104    S: GatewayStore,
105    C: GatewayClock,
106{
107    let event = GatewayEvent::new(
108        ids.next_event_id().map_err(OpenAiRouteError::internal)?,
109        run_id,
110        input.sequence,
111        input.kind,
112        input.payload,
113        clock.now_ms().map_err(OpenAiRouteError::internal)?,
114    );
115    let id = content_id_for_expr(&event.to_expr()).map_err(OpenAiRouteError::internal)?;
116    if store_event {
117        store
118            .put_event(id.clone(), event.clone())
119            .map_err(OpenAiRouteError::internal)?;
120    }
121    event_log.content_ids.push(id);
122    event_log.events.push(event);
123    Ok(())
124}
125
126/// The recorded request and open run produced by [`begin_run`].
127pub(crate) struct RunPrologue {
128    pub(crate) recorded_request: GatewayRequest,
129    pub(crate) request_content_id: ContentId,
130    pub(crate) run_id: String,
131    pub(crate) run_content_id: ContentId,
132}
133
134/// Runs the shared request+run prologue: redact `request`, stamp it with a
135/// fresh request id and timestamp, record it (when `store_record`), then open a
136/// run pointing at it and record that too.
137///
138/// `run_status` is applied to the run before its content id is computed; the
139/// run_record engine passes `completed`, while responses/embeddings pass
140/// `None` and leave the run status at its default. Keeping this optional is
141/// what preserves each route's run wire form.
142pub(crate) fn begin_run<S, C>(
143    store: &mut S,
144    ids: &mut GatewayRunIdGenerators,
145    clock: &mut C,
146    request: &GatewayRequest,
147    run_status: Option<Symbol>,
148    store_record: bool,
149) -> RouteResult<RunPrologue>
150where
151    S: GatewayStore,
152    C: GatewayClock,
153{
154    let recorded_request = redacted_gateway_request(request).with_metadata(
155        ids.request.next_id().map_err(OpenAiRouteError::internal)?,
156        clock.now_ms().map_err(OpenAiRouteError::internal)?,
157    );
158    let request_content_id =
159        request_content_id(&recorded_request).map_err(OpenAiRouteError::internal)?;
160    if store_record {
161        store
162            .put_request(request_content_id.clone(), recorded_request.clone())
163            .map_err(OpenAiRouteError::internal)?;
164    }
165
166    let run_id = ids.run.next_id().map_err(OpenAiRouteError::internal)?;
167    let mut run = GatewayRun::new(
168        run_id.clone(),
169        request_content_id.clone(),
170        clock.now_ms().map_err(OpenAiRouteError::internal)?,
171    );
172    if let Some(status) = run_status {
173        run = run.with_status(status);
174    }
175    let run_content_id = content_id_for_expr(&run.to_expr()).map_err(OpenAiRouteError::internal)?;
176    if store_record {
177        store
178            .put_run(run_content_id.clone(), run)
179            .map_err(OpenAiRouteError::internal)?;
180    }
181
182    Ok(RunPrologue {
183        recorded_request,
184        request_content_id,
185        run_id,
186        run_content_id,
187    })
188}
189
190/// The outcome of executing a gateway run: the wire response plus the
191/// content-addressed ledger ids and events it produced.
192///
193/// `response_id`/`response_created_at_ms` are only populated by the
194/// `/v1/responses` engine (which mints a public response object); the other
195/// engines leave them `None`.
196#[derive(Clone, Debug)]
197pub struct GatewayRunExecution {
198    pub(crate) response: GatewayResponse,
199    pub(crate) request_content_id: Option<ContentId>,
200    pub(crate) run_content_id: Option<ContentId>,
201    pub(crate) event_content_ids: Vec<ContentId>,
202    pub(crate) events: Vec<GatewayEvent>,
203    pub(crate) response_id: Option<String>,
204    pub(crate) response_created_at_ms: Option<u64>,
205    pub(crate) response_content_id: Option<ContentId>,
206}
207
208impl GatewayRunExecution {
209    /// Returns the wire response produced by the execution.
210    pub fn response(&self) -> &GatewayResponse {
211        &self.response
212    }
213
214    /// Returns the content id of the stored request, if it was recorded.
215    pub fn request_content_id(&self) -> Option<&ContentId> {
216        self.request_content_id.as_ref()
217    }
218
219    /// Returns the content id of the stored run, if it was recorded.
220    pub fn run_content_id(&self) -> Option<&ContentId> {
221        self.run_content_id.as_ref()
222    }
223
224    /// Returns the content ids of the stored events, in sequence order.
225    pub fn event_content_ids(&self) -> &[ContentId] {
226        &self.event_content_ids
227    }
228
229    /// Returns the events emitted during the execution.
230    pub fn events(&self) -> &[GatewayEvent] {
231        &self.events
232    }
233
234    /// Returns the generated response id, if a response object was produced.
235    pub fn response_id(&self) -> Option<&str> {
236        self.response_id.as_deref()
237    }
238
239    /// Returns the response creation timestamp in milliseconds, if set.
240    pub fn response_created_at_ms(&self) -> Option<u64> {
241        self.response_created_at_ms
242    }
243
244    /// Returns the content id of the stored response, if it was recorded.
245    pub fn response_content_id(&self) -> Option<&ContentId> {
246        self.response_content_id.as_ref()
247    }
248
249    pub(crate) fn error(error: OpenAiRouteError) -> Self {
250        Self {
251            response: error.into_response(),
252            request_content_id: None,
253            run_content_id: None,
254            event_content_ids: Vec::new(),
255            events: Vec::new(),
256            response_id: None,
257            response_created_at_ms: None,
258            response_content_id: None,
259        }
260    }
261}
262
263/// Extracts the `usage` field of a model response expression, defaulting to nil.
264pub(crate) fn response_usage_expr(response: &Expr) -> Expr {
265    response_field(response, "usage")
266        .cloned()
267        .unwrap_or(Expr::Nil)
268}
269
270fn response_field<'a>(expr: &'a Expr, name: &str) -> Option<&'a Expr> {
271    let Expr::Map(entries) = expr else {
272        return None;
273    };
274    entries.iter().find_map(|(key, value)| match key {
275        Expr::Symbol(symbol) if symbol.namespace.is_none() && symbol.name.as_ref() == name => {
276            Some(value)
277        }
278        _ => None,
279    })
280}
sim_lib_openai_server/routes/execution_record.rs

sim_lib_openai_server/routes/
execution_record.rs
