sim_lib_openai_server/runtime/
fabric.rs

1use std::sync::{Arc, Mutex};
2
3use serde_json::{Map, Number, Value as JsonValue};
4use sim_citizen_derive::non_citizen;
5use sim_kernel::{
6    Consistency, Cx, Error, EvalFabric, EvalMode, EvalReply, EvalRequest, Expr, Object,
7    ObjectCompat, Result, Symbol, Value, eval_remote_capability,
8};
9
10use crate::{
11    clock::{DeterministicGatewayClock, GatewayClock, SystemGatewayClock},
12    objects::{GatewayRequest, GatewayResponseValue},
13    routes::responses::{RESPONSES_PATH, ResponseExecution, ResponseIdGenerators},
14    server::GatewayRouteState,
15};
16
17/// Location-transparent eval fabric exposing the OpenAI gateway as an `EvalFabric`.
18///
19/// Decodes an [`EvalRequest`] into a `/v1/responses` gateway request, runs it
20/// against the shared route state (store, cache, runners, federation), and
21/// returns the gateway response wrapped as a runtime value. Server code targets
22/// this surface rather than a transport-specific API.
23#[non_citizen(
24    reason = "live OpenAI gateway fabric handle; reconstruct route data via openai/GatewayRequest and openai/Plan descriptors",
25    kind = "handle"
26)]
27pub struct OpenAiGatewayFabric {
28    state: GatewayRouteState,
29    runtime: Mutex<GatewayFabricRuntime>,
30    codecs: Vec<Symbol>,
31}
32
33struct GatewayFabricRuntime {
34    ids: ResponseIdGenerators,
35    clock: GatewayFabricClock,
36    last_execution: Option<ResponseExecution>,
37}
38
39enum GatewayFabricClock {
40    System(SystemGatewayClock),
41    Deterministic(DeterministicGatewayClock),
42}
43
44impl OpenAiGatewayFabric {
45    /// Returns a fabric over fresh in-memory state and the system clock.
46    pub fn memory() -> Self {
47        Self::with_state_system(GatewayRouteState::memory(), 1)
48    }
49
50    /// Returns a fabric with deterministic ids and a fixed-step simulated clock.
51    ///
52    /// Suitable for tests and replay: ids count from `id_start` and the clock
53    /// starts at `clock_start_ms`, advancing `clock_step_ms` per read.
54    pub fn deterministic(id_start: u64, clock_start_ms: u64, clock_step_ms: u64) -> Self {
55        Self::with_state_clock(
56            GatewayRouteState::memory(),
57            ResponseIdGenerators::deterministic(id_start),
58            GatewayFabricClock::Deterministic(DeterministicGatewayClock::new(
59                clock_start_ms,
60                clock_step_ms,
61            )),
62        )
63    }
64
65    /// Returns a fabric over the supplied `state` using the system clock.
66    ///
67    /// Response ids are seeded deterministically from `id_seed`.
68    pub fn with_state_system(state: GatewayRouteState, id_seed: u64) -> Self {
69        Self::with_state_clock(
70            state,
71            ResponseIdGenerators::deterministic(id_seed),
72            GatewayFabricClock::System(SystemGatewayClock),
73        )
74    }
75
76    fn with_state_clock(
77        state: GatewayRouteState,
78        ids: ResponseIdGenerators,
79        clock: GatewayFabricClock,
80    ) -> Self {
81        Self {
82            state,
83            runtime: Mutex::new(GatewayFabricRuntime {
84                ids,
85                clock,
86                last_execution: None,
87            }),
88            codecs: vec![
89                Symbol::qualified("codec", "binary"),
90                Symbol::qualified("codec", "openai"),
91            ],
92        }
93    }
94
95    /// Returns the local-first eval request that carries `request`'s body for realization.
96    pub fn eval_request_for_gateway_request(request: &GatewayRequest) -> EvalRequest {
97        EvalRequest {
98            expr: Expr::Bytes(request.body().to_vec()),
99            result_shape: None,
100            required_capabilities: Vec::new(),
101            deadline: None,
102            consistency: Consistency::LocalFirst,
103            mode: EvalMode::Eval,
104            answer_limit: None,
105            stream_buffer: None,
106            stream: false,
107            trace: false,
108        }
109    }
110
111    /// Returns the most recent response execution, or `None` before any run.
112    pub fn last_execution(&self) -> Result<Option<ResponseExecution>> {
113        self.runtime
114            .lock()
115            .map_err(|_| Error::PoisonedLock("openai gateway fabric runtime"))
116            .map(|runtime| runtime.last_execution.clone())
117    }
118}
119
120impl Object for OpenAiGatewayFabric {
121    fn display(&self, _cx: &mut Cx) -> Result<String> {
122        Ok("#<openai-gateway-fabric>".to_owned())
123    }
124
125    fn as_any(&self) -> &dyn std::any::Any {
126        self
127    }
128}
129
130impl ObjectCompat for OpenAiGatewayFabric {
131    fn as_table(&self, cx: &mut Cx) -> Result<Value> {
132        cx.factory().table(vec![
133            (
134                Symbol::new("kind"),
135                cx.factory().symbol(Symbol::new("openai-gateway"))?,
136            ),
137            (
138                Symbol::new("address"),
139                cx.factory().symbol(Symbol::new("local"))?,
140            ),
141            (
142                Symbol::new("codecs"),
143                cx.factory().list(
144                    self.codecs
145                        .iter()
146                        .cloned()
147                        .map(|codec| cx.factory().symbol(codec))
148                        .collect::<Result<Vec<_>>>()?,
149                )?,
150            ),
151        ])
152    }
153
154    fn as_eval_fabric(&self) -> Option<&dyn EvalFabric> {
155        Some(self)
156    }
157}
158
159impl EvalFabric for OpenAiGatewayFabric {
160    fn realize(&self, cx: &mut Cx, request: EvalRequest) -> Result<EvalReply> {
161        if matches!(request.consistency, Consistency::RemoteOnly) {
162            return Err(Error::CapabilityDenied {
163                capability: eval_remote_capability(),
164            });
165        }
166        if !matches!(request.mode, EvalMode::Eval) {
167            return Err(Error::Eval(
168                "openai gateway fabric only supports eval mode".to_owned(),
169            ));
170        }
171        for capability in &request.required_capabilities {
172            cx.require(capability)?;
173        }
174
175        let gateway_request = gateway_request_from_expr(&request.expr, request.stream)?;
176        let mut runtime = self
177            .runtime
178            .lock()
179            .map_err(|_| Error::PoisonedLock("openai gateway fabric runtime"))?;
180        let mut store = self
181            .state
182            .store()
183            .lock()
184            .map_err(|_| Error::PoisonedLock("openai gateway fabric store"))?;
185        let mut cache = self
186            .state
187            .cache()
188            .lock()
189            .map_err(|_| Error::PoisonedLock("openai gateway fabric cache"))?;
190        let runtime = &mut *runtime;
191        let execution =
192            crate::routes::responses::execute_response_request_with_cache_runners_and_federation(
193                cx,
194                &mut *store,
195                &mut cache,
196                &mut runtime.ids,
197                &mut runtime.clock,
198                &gateway_request,
199                crate::routes::responses::ResponseRuntimeTargets::with_federation(
200                    self.state.runners(),
201                    self.state.federation(),
202                ),
203            );
204        let response = execution.response().clone();
205        runtime.last_execution = Some(execution);
206        Ok(EvalReply {
207            value: cx
208                .factory()
209                .opaque(Arc::new(GatewayResponseValue::new(response)))?,
210            diagnostics: cx.take_diagnostics(),
211            trace: request
212                .trace
213                .then(|| cx.factory().symbol(Symbol::new("openai-gateway")))
214                .transpose()?,
215        })
216    }
217}
218
219#[cfg(feature = "http")]
220impl sim_lib_server::EvalSite for OpenAiGatewayFabric {
221    fn site_kind(&self) -> &'static str {
222        "openai-gateway"
223    }
224
225    fn address(&self) -> &sim_lib_server::ServerAddress {
226        static ADDRESS: sim_lib_server::ServerAddress = sim_lib_server::ServerAddress::Local;
227        &ADDRESS
228    }
229
230    fn codecs(&self) -> &[Symbol] {
231        &self.codecs
232    }
233
234    fn answer(
235        &self,
236        cx: &mut Cx,
237        frame: sim_lib_server::ServerFrame,
238    ) -> Result<sim_lib_server::ServerFrame> {
239        let consistency = frame.envelope.consistency;
240        let reply_codec = frame
241            .envelope
242            .reply_codec_hint
243            .clone()
244            .filter(|hint| self.codecs.iter().any(|codec| codec == hint))
245            .unwrap_or_else(|| self.codecs[0].clone());
246        let request = sim_lib_server::eval_request_from_frame(cx, &frame)?;
247        let reply = self.realize(cx, request)?;
248        sim_lib_server::server_frame_from_reply(cx, &reply_codec, reply, consistency)
249    }
250
251    fn as_eval_fabric(&self) -> Option<&dyn EvalFabric> {
252        Some(self)
253    }
254
255    fn as_any(&self) -> &dyn std::any::Any {
256        self
257    }
258}
259
260impl GatewayClock for GatewayFabricClock {
261    fn now_ms(&mut self) -> Result<u64> {
262        match self {
263            Self::System(clock) => clock.now_ms(),
264            Self::Deterministic(clock) => clock.now_ms(),
265        }
266    }
267}
268
269fn gateway_request_from_expr(expr: &Expr, stream: bool) -> Result<GatewayRequest> {
270    let mut body = match expr {
271        Expr::Bytes(bytes) => bytes.clone(),
272        Expr::String(text) => text.as_bytes().to_vec(),
273        Expr::Map(_) => {
274            let mut json = json_from_expr(expr)?;
275            if stream && let Some(object) = json.as_object_mut() {
276                object.insert("stream".to_owned(), JsonValue::Bool(true));
277            }
278            serde_json::to_vec(&json).map_err(|err| {
279                Error::Eval(format!(
280                    "failed to encode OpenAI gateway request JSON: {err}"
281                ))
282            })?
283        }
284        _ => {
285            return Err(Error::TypeMismatch {
286                expected: "OpenAI gateway request bytes, JSON string, or map",
287                found: "non-request",
288            });
289        }
290    };
291    if stream && !matches!(expr, Expr::Map(_)) {
292        body = ensure_streaming_body(body)?;
293    }
294    Ok(GatewayRequest::new(
295        "POST",
296        RESPONSES_PATH,
297        vec![("Content-Type".to_owned(), "application/json".to_owned())],
298        body,
299    ))
300}
301
302fn ensure_streaming_body(body: Vec<u8>) -> Result<Vec<u8>> {
303    let mut json = serde_json::from_slice::<JsonValue>(&body)
304        .map_err(|err| Error::Eval(format!("invalid OpenAI gateway request JSON: {err}")))?;
305    let Some(object) = json.as_object_mut() else {
306        return Err(Error::Eval(
307            "OpenAI gateway request JSON must be an object".to_owned(),
308        ));
309    };
310    object.insert("stream".to_owned(), JsonValue::Bool(true));
311    serde_json::to_vec(&json).map_err(|err| {
312        Error::Eval(format!(
313            "failed to encode streaming OpenAI gateway request JSON: {err}"
314        ))
315    })
316}
317
318fn json_from_expr(expr: &Expr) -> Result<JsonValue> {
319    Ok(match expr {
320        Expr::Nil => JsonValue::Null,
321        Expr::Bool(flag) => JsonValue::Bool(*flag),
322        Expr::Number(number) => number
323            .canonical
324            .parse::<i64>()
325            .ok()
326            .map(Number::from)
327            .map(JsonValue::Number)
328            .unwrap_or_else(|| JsonValue::String(number.canonical.clone())),
329        Expr::String(text) => JsonValue::String(text.clone()),
330        Expr::Symbol(symbol) | Expr::Local(symbol) => JsonValue::String(symbol.to_string()),
331        Expr::List(items) | Expr::Vector(items) => JsonValue::Array(
332            items
333                .iter()
334                .map(json_from_expr)
335                .collect::<Result<Vec<_>>>()?,
336        ),
337        Expr::Map(entries) => {
338            let mut object = Map::new();
339            for (key, value) in entries {
340                object.insert(json_key(key)?, json_from_expr(value)?);
341            }
342            JsonValue::Object(object)
343        }
344        _ => {
345            return Err(Error::TypeMismatch {
346                expected: "JSON-compatible request expression",
347                found: "non-json",
348            });
349        }
350    })
351}
352
353fn json_key(expr: &Expr) -> Result<String> {
354    match expr {
355        Expr::String(text) => Ok(text.clone()),
356        Expr::Symbol(symbol) | Expr::Local(symbol) if symbol.namespace.is_none() => {
357            Ok(symbol.name.as_ref().to_owned())
358        }
359        Expr::Symbol(symbol) | Expr::Local(symbol) => Ok(symbol.to_string()),
360        _ => Err(Error::TypeMismatch {
361            expected: "JSON object key",
362            found: "non-key",
363        }),
364    }
365}
sim_lib_openai_server/runtime/fabric.rs

sim_lib_openai_server/runtime/
fabric.rs
