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

sim_lib_openai_server/runtime/
fabric.rs
