sim_lib_openai_server/routes/
execution_record.rs

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

sim_lib_openai_server/routes/
execution_record.rs
