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::{SystemWallClock, WallClock},
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-shaped `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 clock = SystemWallClock;
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            if let Err(error) = stamp_response_owner(state, &fabric, request) {
70                return error.into_response();
71            }
72            response.response().clone()
73        }
74        Err(err) => OpenAiRouteError::internal_message(format!("gateway realize failed: {err}"))
75            .into_response(),
76    }
77}
78
79fn stamp_response_owner(
80    state: &GatewayRouteState,
81    fabric: &OpenAiGatewayFabric,
82    request: &GatewayRequest,
83) -> RouteResult<()> {
84    let owner_key_id = state
85        .keys()
86        .key_for_request(request)
87        .map_err(OpenAiRouteError::internal)?
88        .map(|key| key.id().to_owned());
89    let Some(execution) = fabric
90        .last_execution()
91        .map_err(OpenAiRouteError::internal)?
92    else {
93        return Ok(());
94    };
95    let Some(response_id) = execution.response_id() else {
96        return Ok(());
97    };
98    let mut store = state.store().lock().map_err(|err| {
99        OpenAiRouteError::internal_message(format!("gateway store lock failed: {err}"))
100    })?;
101    let Some(mut record) = store.response_object(response_id) else {
102        return Ok(());
103    };
104    record.owner_key_id = owner_key_id;
105    store
106        .put_response_object(record)
107        .map_err(OpenAiRouteError::internal)
108}
109
110/// Handles `GET /v1/responses/{id}`, returning the stored response for the id
111/// in the path.
112pub fn handle_response_retrieval(
113    request: &GatewayRequest,
114    state: &GatewayRouteState,
115) -> GatewayResponse {
116    let Some(response_id) = response_id_from_path(request.path()) else {
117        return OpenAiRouteError::not_found("response").into_response();
118    };
119    match state.store().lock() {
120        Ok(store) => retrieve_response(&*store, response_id),
121        Err(err) => OpenAiRouteError::internal_message(format!("gateway store lock failed: {err}"))
122            .into_response(),
123    }
124}
125
126/// Returns the stored response object for `response_id`, or a not-found error
127/// response if no such response is stored.
128pub fn retrieve_response<S>(store: &S, response_id: &str) -> GatewayResponse
129where
130    S: GatewayResponseObjectStore,
131{
132    store
133        .response_object(response_id)
134        .map(|record| record.response().clone())
135        .unwrap_or_else(|| OpenAiRouteError::not_found(response_id).into_response())
136}
137
138/// Executes a `/v1/responses` request with a fresh plan cache and no runner or
139/// federation targets, returning the [`ResponseExecution`] outcome.
140pub fn execute_response_request<S, C>(
141    cx: &mut Cx,
142    store: &mut S,
143    ids: &mut ResponseIdGenerators,
144    clock: &mut C,
145    request: &GatewayRequest,
146) -> ResponseExecution
147where
148    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
149    C: WallClock,
150{
151    let mut cache = OpenAiPlanCache::new();
152    execute_response_request_with_cache(cx, store, &mut cache, ids, clock, request)
153}
154
155/// Executes a `/v1/responses` request reusing the given plan cache, with no
156/// runner or federation targets.
157pub fn execute_response_request_with_cache<S, C>(
158    cx: &mut Cx,
159    store: &mut S,
160    cache: &mut OpenAiPlanCache,
161    ids: &mut ResponseIdGenerators,
162    clock: &mut C,
163    request: &GatewayRequest,
164) -> ResponseExecution
165where
166    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
167    C: WallClock,
168{
169    match try_execute_response_request(cx, store, cache, ids, clock, request, None) {
170        Ok(execution) => execution,
171        Err(error) => ResponseExecution::error(error),
172    }
173}
174
175/// Executes a `/v1/responses` request against the given runner registry,
176/// using a fresh plan cache and no federation.
177pub fn execute_response_request_with_runners<S, C>(
178    cx: &mut Cx,
179    store: &mut S,
180    ids: &mut ResponseIdGenerators,
181    clock: &mut C,
182    request: &GatewayRequest,
183    runners: &OpenAiRunnerRegistry,
184) -> ResponseExecution
185where
186    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
187    C: WallClock,
188{
189    let mut cache = OpenAiPlanCache::new();
190    execute_response_request_with_cache_and_runners(
191        cx, store, &mut cache, ids, clock, request, runners,
192    )
193}
194
195/// Executes a `/v1/responses` request against the given runner registry,
196/// reusing the supplied plan cache and using no federation.
197pub fn execute_response_request_with_cache_and_runners<S, C>(
198    cx: &mut Cx,
199    store: &mut S,
200    cache: &mut OpenAiPlanCache,
201    ids: &mut ResponseIdGenerators,
202    clock: &mut C,
203    request: &GatewayRequest,
204    runners: &OpenAiRunnerRegistry,
205) -> ResponseExecution
206where
207    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
208    C: WallClock,
209{
210    match try_execute_response_request(
211        cx,
212        store,
213        cache,
214        ids,
215        clock,
216        request,
217        Some(ResponseRuntimeTargets::runners(runners)),
218    ) {
219        Ok(execution) => execution,
220        Err(error) => ResponseExecution::error(error),
221    }
222}
223
224/// Executes a `/v1/responses` request reusing the given plan cache and
225/// dispatching to the supplied runner and federation [`ResponseRuntimeTargets`].
226pub fn execute_response_request_with_cache_runners_and_federation<S, C>(
227    cx: &mut Cx,
228    store: &mut S,
229    cache: &mut OpenAiPlanCache,
230    ids: &mut ResponseIdGenerators,
231    clock: &mut C,
232    request: &GatewayRequest,
233    targets: ResponseRuntimeTargets<'_>,
234) -> ResponseExecution
235where
236    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
237    C: WallClock,
238{
239    match try_execute_response_request(cx, store, cache, ids, clock, request, Some(targets)) {
240        Ok(execution) => execution,
241        Err(error) => ResponseExecution::error(error),
242    }
243}
244
245fn try_execute_response_request<S, C>(
246    cx: &mut Cx,
247    store: &mut S,
248    cache: &mut OpenAiPlanCache,
249    ids: &mut ResponseIdGenerators,
250    clock: &mut C,
251    request: &GatewayRequest,
252    targets: Option<ResponseRuntimeTargets<'_>>,
253) -> RouteResult<ResponseExecution>
254where
255    S: GatewayStore + GatewayResponseObjectStore + GatewayStateStore,
256    C: WallClock,
257{
258    let normalized = normalize_response_request(store, request)?;
259    let object = normalized.object;
260    let model = required_string(&object, "model")?.to_owned();
261    require_input(&object)?;
262    let store_response = object
263        .get("store")
264        .and_then(Value::as_bool)
265        .unwrap_or(false);
266    let stream_response = object
267        .get("stream")
268        .and_then(Value::as_bool)
269        .unwrap_or(false);
270
271    let request_expr = decode_openai_request(Input::Bytes(normalized.request.body().to_vec()))
272        .map_err(OpenAiRouteError::bad_request_from_error)?;
273    let plan = parse_plan(&model).map_err(OpenAiRouteError::bad_model_from_error)?;
274    check_plan(&plan).map_err(OpenAiRouteError::bad_model_from_error)?;
275
276    let RunPrologue {
277        recorded_request,
278        request_content_id,
279        run_id,
280        run_content_id,
281    } = begin_run(store, ids, clock, &normalized.request, None, store_response)?;
282
283    let mut event_log = EventLog::default();
284    let mut sequence = 0;
285    append_event(
286        store,
287        ids,
288        clock,
289        &run_id,
290        EventInput::new(sequence, "request-start", recorded_request.to_expr()),
291        store_response,
292        &mut event_log,
293    )?;
294    sequence += 1;
295    append_event(
296        store,
297        ids,
298        clock,
299        &run_id,
300        EventInput::new(sequence, "plan-start", plan.clone()),
301        store_response,
302        &mut event_log,
303    )?;
304    sequence += 1;
305    append_event(
306        store,
307        ids,
308        clock,
309        &run_id,
310        EventInput::new(sequence, "model-start", Expr::String(model.clone())),
311        store_response,
312        &mut event_log,
313    )?;
314    sequence += 1;
315
316    let initial_plan_report = match targets {
317        Some(targets) => {
318            if let Some(federation) = targets.federation_ref() {
319                eval_plan_report_with_cache_runners_and_federation(
320                    cx,
321                    &plan,
322                    &request_expr,
323                    cache,
324                    targets.runners_ref(),
325                    federation,
326                )
327            } else {
328                eval_plan_report_with_cache_and_runners(
329                    cx,
330                    &plan,
331                    &request_expr,
332                    cache,
333                    targets.runners_ref(),
334                )
335            }
336        }
337        None => eval_plan_report_with_cache(cx, &plan, &request_expr, cache),
338    }
339    .map_err(|err| OpenAiRouteError::model(err, &model))?;
340    let plan_report = run_tool_loop_with_cache(
341        cx,
342        &plan,
343        &request_expr,
344        &object,
345        cache,
346        initial_plan_report,
347    )
348    .map_err(|err| OpenAiRouteError::model(err, &model))?;
349    for event in &plan_report.events {
350        append_event(
351            store,
352            ids,
353            clock,
354            &run_id,
355            EventInput::from_symbol(sequence, event.kind.clone(), event.payload.clone()),
356            store_response,
357            &mut event_log,
358        )?;
359        sequence += 1;
360    }
361    let model_response = plan_report.response;
362    for delta in response_delta_chunks(&model_response, stream_response)? {
363        append_event(
364            store,
365            ids,
366            clock,
367            &run_id,
368            EventInput::new(sequence, "delta", Expr::String(delta)),
369            store_response,
370            &mut event_log,
371        )?;
372        sequence += 1;
373    }
374    append_event(
375        store,
376        ids,
377        clock,
378        &run_id,
379        EventInput::new(sequence, "usage", response_usage_expr(&model_response)),
380        store_response,
381        &mut event_log,
382    )?;
383    sequence += 1;
384    append_event(
385        store,
386        ids,
387        clock,
388        &run_id,
389        EventInput::new(sequence, "final", model_response.clone()),
390        store_response,
391        &mut event_log,
392    )?;
393
394    let response_id = ids.response.next_id().map_err(OpenAiRouteError::internal)?;
395    let response_created_at = clock.now_ms().map_err(OpenAiRouteError::internal)?;
396    let response_body =
397        encode_openai_responses_response(&model_response, &response_id, response_created_at)
398            .map_err(OpenAiRouteError::internal)?;
399    let final_response = GatewayResponse::json(200, response_body);
400    let response_content_id = if store_response {
401        let id =
402            content_id_for_expr(&final_response.to_expr()).map_err(OpenAiRouteError::internal)?;
403        let mut record =
404            StoredGatewayResponse::new(response_id.clone(), id.clone(), final_response.clone());
405        record.request_content_id = Some(request_content_id.clone());
406        record.run_content_id = Some(run_content_id.clone());
407        record.event_content_ids = event_log.content_ids.clone();
408        store
409            .put_response_object(record)
410            .map_err(OpenAiRouteError::internal)?;
411        Some(id)
412    } else {
413        None
414    };
415    let response = if stream_response {
416        GatewayResponse::sse(
417            200,
418            encode_gateway_events_sse(
419                &event_log.events,
420                OpenAiSseSurface::Responses,
421                &response_id,
422                response_created_at,
423            )
424            .map_err(OpenAiRouteError::internal)?,
425        )
426    } else {
427        final_response
428    };
429
430    Ok(ResponseExecution {
431        response,
432        request_content_id: Some(request_content_id),
433        run_content_id: Some(run_content_id),
434        event_content_ids: event_log.content_ids,
435        events: event_log.events,
436        response_id: Some(response_id),
437        response_created_at_ms: Some(response_created_at),
438        response_content_id,
439    })
440}
441
442fn response_id_from_path(path: &str) -> Option<&str> {
443    super::path::id_from_path(path, RESPONSE_RETRIEVAL_PREFIX)
444}
445
446use crate::routes::request_json::required_string;
447
448fn require_input(object: &Map<String, Value>) -> RouteResult<()> {
449    if object.contains_key("input") || object.contains_key("messages") {
450        Ok(())
451    } else {
452        Err(OpenAiRouteError::missing_required("input"))
453    }
454}
sim_lib_openai_server/routes/responses.rs

sim_lib_openai_server/routes/
responses.rs
