sim_lib_openai_server/routes/
responses.rs

1use std::sync::Arc;
2
3use serde_json::{Map, Value};
4use sim_codec::Input;
5use sim_kernel::{Cx, DefaultFactory, EvalFabric, Expr, NoopEvalPolicy};
6
7use crate::{
8    clock::{GatewayClock, SystemGatewayClock},
9    codec_openai::{
10        OpenAiSseSurface, decode_openai_request, encode_gateway_events_sse,
11        encode_openai_responses_response,
12    },
13    content_id::{content_id_for_expr, request_content_id},
14    objects::{GatewayRequest, GatewayResponse, GatewayResponseValue, GatewayRun},
15    plan::{
16        check_plan, eval_plan_report_with_cache, eval_plan_report_with_cache_and_runners,
17        eval_plan_report_with_cache_runners_and_federation, parse_plan,
18    },
19    runtime::{
20        OpenAiGatewayFabric, OpenAiPlanCache, OpenAiRunnerRegistry, redacted_gateway_request,
21        run_tool_loop_with_cache,
22    },
23    server::GatewayRouteState,
24    storage::{GatewayResponseObjectStore, GatewayStateStore, GatewayStore, StoredGatewayResponse},
25};
26
27pub use super::response_runtime::{
28    ResponseExecution, ResponseIdGenerators, ResponseRuntimeTargets,
29};
30use super::{
31    errors::OpenAiRouteError,
32    response_log::{EventInput, EventLog, append_event, response_usage_expr},
33    response_text::response_delta_chunks,
34    thread_context::normalize_response_request,
35};
36
37/// Route path for the OpenAI-compatible `POST /v1/responses` endpoint.
38pub const RESPONSES_PATH: &str = "/v1/responses";
39/// Path prefix stripped to extract a response id from a retrieval request.
40pub const RESPONSE_RETRIEVAL_PREFIX: &str = "/v1/responses/";
41/// Route template for retrieving a single stored response by id.
42pub const RESPONSE_RETRIEVAL_ROUTE: &str = "/v1/responses/{id}";
43
44type RouteResult<T> = std::result::Result<T, OpenAiRouteError>;
45
46/// Handles `POST /v1/responses`, realizing the request through the gateway
47/// eval fabric under the caller's effective capabilities.
48pub fn handle_responses(request: &GatewayRequest, state: &GatewayRouteState) -> GatewayResponse {
49    let mut clock = SystemGatewayClock;
50    let seed = clock.now_ms().unwrap_or(1);
51    let mut cx = Cx::new(Arc::new(NoopEvalPolicy), Arc::new(DefaultFactory));
52    let fabric = OpenAiGatewayFabric::with_state_system(state.clone(), seed);
53    match state
54        .keys()
55        .with_effective_capabilities(&mut cx, request, |cx| {
56            fabric.realize(
57                cx,
58                OpenAiGatewayFabric::eval_request_for_gateway_request(request),
59            )
60        }) {
61        Ok(reply) => {
62            let Some(response) = reply.value.object().downcast_ref::<GatewayResponseValue>() else {
63                return OpenAiRouteError::internal_message(
64                    "openai gateway fabric returned a non-response value",
65                )
66                .into_response();
67            };
68            response.response().clone()
69        }
70        Err(err) => OpenAiRouteError::internal_message(format!("gateway realize failed: {err}"))
71            .into_response(),
72    }
73}
74
75/// Handles `GET /v1/responses/{id}`, returning the stored response for the id
76/// in the path.
77pub fn handle_response_retrieval(
78    request: &GatewayRequest,
79    state: &GatewayRouteState,
80) -> GatewayResponse {
81    let Some(response_id) = response_id_from_path(request.path()) else {
82        return OpenAiRouteError::not_found("response").into_response();
83    };
84    match state.store().lock() {
85        Ok(store) => retrieve_response(&*store, response_id),
86        Err(err) => OpenAiRouteError::internal_message(format!("gateway store lock failed: {err}"))
87            .into_response(),
88    }
89}
90
91/// Returns the stored response object for `response_id`, or a not-found error
92/// response if no such response is stored.
93pub fn retrieve_response<S>(store: &S, response_id: &str) -> GatewayResponse
94where
95    S: GatewayResponseObjectStore,
96{
97    store
98        .response_object(response_id)
99        .map(|record| record.response().clone())
100        .unwrap_or_else(|| OpenAiRouteError::not_found(response_id).into_response())
101}
102
103/// Executes a `/v1/responses` request with a fresh plan cache and no runner or
104/// federation targets, returning the [`ResponseExecution`] outcome.
105pub fn execute_response_request<S, C>(
106    cx: &mut Cx,
107    store: &mut S,
108    ids: &mut ResponseIdGenerators,
109    clock: &mut C,
110    request: &GatewayRequest,
111) -> ResponseExecution
112where
113    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
114    C: GatewayClock,
115{
116    let mut cache = OpenAiPlanCache::new();
117    execute_response_request_with_cache(cx, store, &mut cache, ids, clock, request)
118}
119
120/// Executes a `/v1/responses` request reusing the given plan cache, with no
121/// runner or federation targets.
122pub fn execute_response_request_with_cache<S, C>(
123    cx: &mut Cx,
124    store: &mut S,
125    cache: &mut OpenAiPlanCache,
126    ids: &mut ResponseIdGenerators,
127    clock: &mut C,
128    request: &GatewayRequest,
129) -> ResponseExecution
130where
131    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
132    C: GatewayClock,
133{
134    match try_execute_response_request(cx, store, cache, ids, clock, request, None) {
135        Ok(execution) => execution,
136        Err(error) => ResponseExecution::error(error),
137    }
138}
139
140/// Executes a `/v1/responses` request against the given runner registry,
141/// using a fresh plan cache and no federation.
142pub fn execute_response_request_with_runners<S, C>(
143    cx: &mut Cx,
144    store: &mut S,
145    ids: &mut ResponseIdGenerators,
146    clock: &mut C,
147    request: &GatewayRequest,
148    runners: &OpenAiRunnerRegistry,
149) -> ResponseExecution
150where
151    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
152    C: GatewayClock,
153{
154    let mut cache = OpenAiPlanCache::new();
155    execute_response_request_with_cache_and_runners(
156        cx, store, &mut cache, ids, clock, request, runners,
157    )
158}
159
160/// Executes a `/v1/responses` request against the given runner registry,
161/// reusing the supplied plan cache and using no federation.
162pub fn execute_response_request_with_cache_and_runners<S, C>(
163    cx: &mut Cx,
164    store: &mut S,
165    cache: &mut OpenAiPlanCache,
166    ids: &mut ResponseIdGenerators,
167    clock: &mut C,
168    request: &GatewayRequest,
169    runners: &OpenAiRunnerRegistry,
170) -> ResponseExecution
171where
172    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
173    C: GatewayClock,
174{
175    match try_execute_response_request(
176        cx,
177        store,
178        cache,
179        ids,
180        clock,
181        request,
182        Some(ResponseRuntimeTargets::runners(runners)),
183    ) {
184        Ok(execution) => execution,
185        Err(error) => ResponseExecution::error(error),
186    }
187}
188
189/// Executes a `/v1/responses` request reusing the given plan cache and
190/// dispatching to the supplied runner and federation [`ResponseRuntimeTargets`].
191pub fn execute_response_request_with_cache_runners_and_federation<S, C>(
192    cx: &mut Cx,
193    store: &mut S,
194    cache: &mut OpenAiPlanCache,
195    ids: &mut ResponseIdGenerators,
196    clock: &mut C,
197    request: &GatewayRequest,
198    targets: ResponseRuntimeTargets<'_>,
199) -> ResponseExecution
200where
201    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
202    C: GatewayClock,
203{
204    match try_execute_response_request(cx, store, cache, ids, clock, request, Some(targets)) {
205        Ok(execution) => execution,
206        Err(error) => ResponseExecution::error(error),
207    }
208}
209
210fn try_execute_response_request<S, C>(
211    cx: &mut Cx,
212    store: &mut S,
213    cache: &mut OpenAiPlanCache,
214    ids: &mut ResponseIdGenerators,
215    clock: &mut C,
216    request: &GatewayRequest,
217    targets: Option<ResponseRuntimeTargets<'_>>,
218) -> RouteResult<ResponseExecution>
219where
220    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
221    C: GatewayClock,
222{
223    let normalized = normalize_response_request(store, request)?;
224    let object = normalized.object;
225    let model = required_string(&object, "model")?.to_owned();
226    require_input(&object)?;
227    let store_response = object
228        .get("store")
229        .and_then(Value::as_bool)
230        .unwrap_or(false);
231    let stream_response = object
232        .get("stream")
233        .and_then(Value::as_bool)
234        .unwrap_or(false);
235
236    let request_expr = decode_openai_request(Input::Bytes(normalized.request.body().to_vec()))
237        .map_err(OpenAiRouteError::bad_request_from_error)?;
238    let plan = parse_plan(&model).map_err(OpenAiRouteError::bad_model_from_error)?;
239    check_plan(&plan).map_err(OpenAiRouteError::bad_model_from_error)?;
240
241    let recorded_request = redacted_gateway_request(&normalized.request).with_metadata(
242        ids.request.next_id().map_err(OpenAiRouteError::internal)?,
243        clock.now_ms().map_err(OpenAiRouteError::internal)?,
244    );
245    let request_content_id =
246        request_content_id(&recorded_request).map_err(OpenAiRouteError::internal)?;
247    if store_response {
248        store
249            .put_request(request_content_id.clone(), recorded_request.clone())
250            .map_err(OpenAiRouteError::internal)?;
251    }
252
253    let run_id = ids.run.next_id().map_err(OpenAiRouteError::internal)?;
254    let run = GatewayRun::new(
255        run_id.clone(),
256        request_content_id.clone(),
257        clock.now_ms().map_err(OpenAiRouteError::internal)?,
258    );
259    let run_content_id = content_id_for_expr(&run.to_expr()).map_err(OpenAiRouteError::internal)?;
260    if store_response {
261        store
262            .put_run(run_content_id.clone(), run.clone())
263            .map_err(OpenAiRouteError::internal)?;
264    }
265
266    let mut event_log = EventLog::default();
267    let mut sequence = 0;
268    append_event(
269        store,
270        ids,
271        clock,
272        &run_id,
273        EventInput::new(sequence, "request-start", recorded_request.to_expr()),
274        store_response,
275        &mut event_log,
276    )?;
277    sequence += 1;
278    append_event(
279        store,
280        ids,
281        clock,
282        &run_id,
283        EventInput::new(sequence, "plan-start", plan.clone()),
284        store_response,
285        &mut event_log,
286    )?;
287    sequence += 1;
288    append_event(
289        store,
290        ids,
291        clock,
292        &run_id,
293        EventInput::new(sequence, "model-start", Expr::String(model.clone())),
294        store_response,
295        &mut event_log,
296    )?;
297    sequence += 1;
298
299    let initial_plan_report = match targets {
300        Some(targets) => {
301            if let Some(federation) = targets.federation_ref() {
302                eval_plan_report_with_cache_runners_and_federation(
303                    cx,
304                    &plan,
305                    &request_expr,
306                    cache,
307                    targets.runners_ref(),
308                    federation,
309                )
310            } else {
311                eval_plan_report_with_cache_and_runners(
312                    cx,
313                    &plan,
314                    &request_expr,
315                    cache,
316                    targets.runners_ref(),
317                )
318            }
319        }
320        None => eval_plan_report_with_cache(cx, &plan, &request_expr, cache),
321    }
322    .map_err(|err| OpenAiRouteError::model(err, &model))?;
323    let plan_report = run_tool_loop_with_cache(
324        cx,
325        &plan,
326        &request_expr,
327        &object,
328        cache,
329        initial_plan_report,
330    )
331    .map_err(|err| OpenAiRouteError::model(err, &model))?;
332    for event in &plan_report.events {
333        append_event(
334            store,
335            ids,
336            clock,
337            &run_id,
338            EventInput::from_symbol(sequence, event.kind.clone(), event.payload.clone()),
339            store_response,
340            &mut event_log,
341        )?;
342        sequence += 1;
343    }
344    let model_response = plan_report.response;
345    for delta in response_delta_chunks(&model_response, stream_response)? {
346        append_event(
347            store,
348            ids,
349            clock,
350            &run_id,
351            EventInput::new(sequence, "delta", Expr::String(delta)),
352            store_response,
353            &mut event_log,
354        )?;
355        sequence += 1;
356    }
357    append_event(
358        store,
359        ids,
360        clock,
361        &run_id,
362        EventInput::new(sequence, "usage", response_usage_expr(&model_response)),
363        store_response,
364        &mut event_log,
365    )?;
366    sequence += 1;
367    append_event(
368        store,
369        ids,
370        clock,
371        &run_id,
372        EventInput::new(sequence, "final", model_response.clone()),
373        store_response,
374        &mut event_log,
375    )?;
376
377    let response_id = ids.response.next_id().map_err(OpenAiRouteError::internal)?;
378    let response_created_at = clock.now_ms().map_err(OpenAiRouteError::internal)?;
379    let response_body =
380        encode_openai_responses_response(&model_response, &response_id, response_created_at)
381            .map_err(OpenAiRouteError::internal)?;
382    let final_response = GatewayResponse::json(200, response_body);
383    let response_content_id = if store_response {
384        let id =
385            content_id_for_expr(&final_response.to_expr()).map_err(OpenAiRouteError::internal)?;
386        let mut record =
387            StoredGatewayResponse::new(response_id.clone(), id.clone(), final_response.clone());
388        record.request_content_id = Some(request_content_id.clone());
389        record.run_content_id = Some(run_content_id.clone());
390        record.event_content_ids = event_log.content_ids.clone();
391        store
392            .put_response_object(record)
393            .map_err(OpenAiRouteError::internal)?;
394        Some(id)
395    } else {
396        None
397    };
398    let response = if stream_response {
399        GatewayResponse::sse(
400            200,
401            encode_gateway_events_sse(
402                &event_log.events,
403                OpenAiSseSurface::Responses,
404                &response_id,
405                response_created_at,
406            )
407            .map_err(OpenAiRouteError::internal)?,
408        )
409    } else {
410        final_response
411    };
412
413    Ok(ResponseExecution {
414        response,
415        request_content_id: Some(request_content_id),
416        run_content_id: Some(run_content_id),
417        event_content_ids: event_log.content_ids,
418        events: event_log.events,
419        response_id: Some(response_id),
420        response_created_at_ms: Some(response_created_at),
421        response_content_id,
422    })
423}
424
425fn response_id_from_path(path: &str) -> Option<&str> {
426    super::path::id_from_path(path, RESPONSE_RETRIEVAL_PREFIX)
427}
428
429use crate::routes::request_json::required_string;
430
431fn require_input(object: &Map<String, Value>) -> RouteResult<()> {
432    if object.contains_key("input") || object.contains_key("messages") {
433        Ok(())
434    } else {
435        Err(OpenAiRouteError::missing_required("input"))
436    }
437}
sim_lib_openai_server/routes/responses.rs

sim_lib_openai_server/routes/
responses.rs
