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

sim_lib_openai_server/routes/
responses.rs
