superfluid_daemon/
api.rs

1//! api-native.
2
3use std::collections::{HashMap, HashSet};
4use std::io::{Read, Write};
5use std::os::unix::net::{UnixListener, UnixStream};
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::sync::{Arc, Mutex};
8
9use serde::{Deserialize, Serialize};
10
11use crate::bus::{BusMsg, EndReason};
12use crate::store::CommittedEvent;
13use crate::wal::{EventBody, GenParams};
14use crate::{Daemon, DaemonError};
15
16#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
17pub struct EventMsg {
18    pub event_id: u64,
19    pub epoch: u64,
20    pub ts_unix_ms: u64,
21    pub body: EventBody,
22}
23
24impl From<CommittedEvent> for EventMsg {
25    fn from(e: CommittedEvent) -> EventMsg {
26        EventMsg {
27            event_id: e.event_id,
28            epoch: e.epoch,
29            ts_unix_ms: e.ts_unix_ms,
30            body: e.body,
31        }
32    }
33}
34
35// postcard encodes a variant as its index on the socket, so new variants of Request and
36// Response go at the end.
37#[derive(Debug, Serialize, Deserialize)]
38pub enum Request {
39    Create {
40        parent: Option<u64>,
41        params: GenParams,
42    },
43    Append {
44        session: u64,
45        text: Option<String>,
46        span: Vec<u32>,
47    },
48    AppendMessage {
49        session: u64,
50        role: u32,
51        text: String,
52    },
53    Generate {
54        session: u64,
55        max_tokens: u32,
56    },
57    Read {
58        session: u64,
59        cursor: u64,
60    },
61    AppendToolResult {
62        session: u64,
63        call_id: u64,
64        content: String,
65    },
66    OpenToolCalls {
67        session: u64,
68    },
69    List,
70    Cancel {
71        session: u64,
72    },
73    Subscribe {
74        session: u64,
75        cursor: u64,
76        provisional: bool,
77    },
78    Unsubscribe {
79        sub: u64,
80    },
81    Fork {
82        session: u64,
83        at_event: u64,
84        params: Option<GenParams>,
85    },
86    Tree,
87    Sessions {
88        include_archived: bool,
89    },
90    SetMeta {
91        session: u64,
92        expected_version: u64,
93        title: Option<String>,
94        archived: Option<bool>,
95    },
96    Rebase {
97        session: u64,
98        edits: Vec<crate::wal::RebaseEdit>,
99        params: Option<GenParams>,
100    },
101    Purge {
102        session: u64,
103        expected_generation: u64,
104        mode: crate::wal::PurgeMode,
105    },
106    SetQos {
107        session: u64,
108        class: u8,
109        batch_invariant: bool,
110    },
111    SetBackgroundRate {
112        divisor: u32,
113    },
114    Inspect {
115        session: u64,
116    },
117    SetLogLevel {
118        directive: String,
119    },
120    Metrics,
121    AppendBlock {
122        session: u64,
123        role: u32,
124        kind: u32,
125        payload: String,
126    },
127    RequestPermission {
128        session: u64,
129        call_id: u64,
130        text: String,
131    },
132    RespondPermission {
133        session: u64,
134        request_id: u64,
135        granted: bool,
136    },
137    RequestToolCancel {
138        session: u64,
139        call_id: u64,
140    },
141    AppendToolOutcome {
142        session: u64,
143        call_id: u64,
144        outcome: u8,
145        note: String,
146    },
147    Ledger {
148        session: u64,
149    },
150    PutMedia {
151        bytes: Vec<u8>,
152        mime: String,
153    },
154    AppendImage {
155        session: u64,
156        role: u32,
157        blob: String,
158        pre_text: String,
159        post_text: String,
160    },
161    GcMedia,
162    Complete {
163        prefix: String,
164        suffix: String,
165        mode: u8,
166        max_tokens: u32,
167    },
168    Trim {
169        session: u64,
170        to_event: u64,
171    },
172    Pin {
173        session: u64,
174        ttl_ms: u64,
175    },
176    AppendSystem {
177        session: u64,
178        text: Option<String>,
179        tools: Vec<String>,
180    },
181}
182
183#[derive(Debug, Serialize, Deserialize)]
184pub enum Response {
185    Created {
186        session: u64,
187    },
188    Delta {
189        event: EventMsg,
190    },
191    Cancelling,
192    Committed {
193        event: EventMsg,
194    },
195    Generated {
196        events: Vec<EventMsg>,
197        tokens_generated: u32,
198        finish: u32,
199        warm_prefix: u64,
200    },
201    Events {
202        events: Vec<EventMsg>,
203    },
204    Sessions {
205        ids: Vec<u64>,
206    },
207    ToolCalls {
208        calls: Vec<(u64, String, String)>,
209    },
210    Subscribed {
211        sub: u64,
212    },
213    SubEvent {
214        sub: u64,
215        event: EventMsg,
216    },
217    SubProvisional {
218        sub: u64,
219        channel: u32,
220        text: String,
221    },
222    SubEnded {
223        sub: u64,
224        reason: String,
225    },
226    Unsubscribed,
227    Err {
228        message: String,
229    },
230    Tree {
231        nodes: Vec<(u64, Option<u64>, u64)>,
232    },
233    SessionList {
234        sessions: Vec<crate::SessionSummary>,
235    },
236    Meta {
237        version: u64,
238    },
239    Purged {
240        sessions: Vec<u64>,
241    },
242    Ack,
243    Inspection {
244        inspection: crate::Inspection,
245    },
246    Text {
247        text: String,
248    },
249    Ledger {
250        entries: Vec<(u64, crate::store::LedgerEntry)>,
251    },
252    Media {
253        hash: String,
254    },
255    Gc {
256        kept: u64,
257        removed: u64,
258    },
259    Completion {
260        text: String,
261        tokens: Vec<u32>,
262        cached: bool,
263        expired: bool,
264    },
265    Pinned {
266        deadline_unix_ms: u64,
267    },
268    Throttled {
269        retry_after_ms: u64,
270    },
271}
272
273fn write_frame<T: Serialize>(w: &mut impl Write, msg: &T) -> Result<(), DaemonError> {
274    let payload = postcard::to_stdvec(msg)?;
275    w.write_all(&(payload.len() as u32).to_le_bytes())?;
276    w.write_all(&payload)?;
277    Ok(())
278}
279
280fn read_frame<T: for<'de> Deserialize<'de>>(r: &mut impl Read) -> Result<Option<T>, DaemonError> {
281    let mut len_buf = [0u8; 4];
282    match r.read_exact(&mut len_buf) {
283        Ok(()) => {}
284        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None),
285        Err(e) => return Err(e.into()),
286    }
287    let len = u32::from_le_bytes(len_buf) as usize;
288    const MAX_FRAME: usize = 64 << 20;
289    if len > MAX_FRAME {
290        return Err(DaemonError::Protocol("api frame exceeds limit"));
291    }
292    let mut payload = vec![0u8; len];
293    r.read_exact(&mut payload)?;
294    Ok(Some(postcard::from_bytes(&payload)?))
295}
296
297pub fn handle_web(d: &Daemon, req: Request) -> Response {
298    match req {
299        Request::Cancel { session } => cancel(&d.cancel_registry(), &d.active_registry(), session),
300        Request::Generate { .. }
301        | Request::Subscribe { .. }
302        | Request::Unsubscribe { .. } => Response::Err {
303            message: "streaming verbs use the WebSocket surface (/web/ws) or SSE (/web/stream), \
304                      not /web/rpc"
305                .into(),
306        },
307        other => handle(d, other),
308    }
309}
310
311fn cancel(cancels: &crate::CancelSet, active: &Mutex<HashSet<u64>>, session: u64) -> Response {
312    let mut c = cancels.lock().expect("cancel registry");
313    if active.lock().expect("active set").contains(&session) {
314        c.insert(session);
315        // The running tick ends at its next step, not when it would have.
316        cancels.wake();
317        Response::Cancelling
318    } else {
319        Response::Err {
320            message: format!("session {session} has no generation in flight"),
321        }
322    }
323}
324
325fn handle(d: &Daemon, req: Request) -> Response {
326    let result: Result<Response, DaemonError> = (|| match req {
327        Request::Create { parent, params } => {
328            let session = d.create(parent, params)?;
329            Ok(Response::Created { session })
330        }
331        Request::Append {
332            session,
333            text,
334            span,
335        } => {
336            let event = d.append(session, text, span)?;
337            Ok(Response::Committed {
338                event: event.into(),
339            })
340        }
341        Request::AppendMessage {
342            session,
343            role,
344            text,
345        } => {
346            let event = d.append_message(session, role, text)?;
347            Ok(Response::Committed {
348                event: event.into(),
349            })
350        }
351        Request::Read { session, cursor } => {
352            let events = d.read(session, cursor)?;
353            Ok(Response::Events {
354                events: events.into_iter().map(Into::into).collect(),
355            })
356        }
357        Request::AppendToolResult {
358            session,
359            call_id,
360            content,
361        } => {
362            let event = d.append_tool_result(session, call_id, content)?;
363            Ok(Response::Committed {
364                event: event.into(),
365            })
366        }
367        Request::OpenToolCalls { session } => Ok(Response::ToolCalls {
368            calls: d.open_tool_calls(session)?,
369        }),
370        Request::List => Ok(Response::Sessions {
371            ids: d.session_ids(),
372        }),
373        Request::Fork {
374            session,
375            at_event,
376            params,
377        } => {
378            let child = d.fork(session, at_event, params)?;
379            Ok(Response::Created { session: child })
380        }
381        Request::Tree => Ok(Response::Tree { nodes: d.tree() }),
382        Request::Sessions { include_archived } => Ok(Response::SessionList {
383            sessions: d.sessions(include_archived),
384        }),
385        Request::SetMeta {
386            session,
387            expected_version,
388            title,
389            archived,
390        } => Ok(Response::Meta {
391            version: d.set_meta(session, expected_version, title, archived)?,
392        }),
393        Request::Rebase {
394            session,
395            edits,
396            params,
397        } => {
398            let child = d.rebase(session, edits, params)?;
399            Ok(Response::Created { session: child })
400        }
401        Request::Purge {
402            session,
403            expected_generation,
404            mode,
405        } => Ok(Response::Purged {
406            sessions: d.purge(session, expected_generation, mode)?,
407        }),
408        Request::SetQos {
409            session,
410            class,
411            batch_invariant,
412        } => Ok(Response::Committed {
413            event: d.set_qos(session, class, batch_invariant)?.into(),
414        }),
415        Request::SetBackgroundRate { divisor } => {
416            d.set_background_tick_divisor(divisor);
417            Ok(Response::Ack)
418        }
419        Request::Inspect { session } => Ok(Response::Inspection {
420            inspection: d.inspect(session)?,
421        }),
422        Request::SetLogLevel { directive } => {
423            crate::telemetry::set_filter(&directive).map_err(DaemonError::Generation)?;
424            Ok(Response::Ack)
425        }
426        Request::Metrics => Ok(Response::Text {
427            text: d.metrics_text(),
428        }),
429        Request::AppendBlock {
430            session,
431            role,
432            kind,
433            payload,
434        } => Ok(Response::Committed {
435            event: d.append_block(session, role, kind, payload)?.into(),
436        }),
437        Request::RequestPermission {
438            session,
439            call_id,
440            text,
441        } => Ok(Response::Committed {
442            event: d.request_permission(session, call_id, text)?.into(),
443        }),
444        Request::RespondPermission {
445            session,
446            request_id,
447            granted,
448        } => Ok(Response::Committed {
449            event: d.respond_permission(session, request_id, granted)?.into(),
450        }),
451        Request::RequestToolCancel { session, call_id } => Ok(Response::Committed {
452            event: d.request_tool_cancel(session, call_id)?.into(),
453        }),
454        Request::AppendToolOutcome {
455            session,
456            call_id,
457            outcome,
458            note,
459        } => Ok(Response::Committed {
460            event: d.append_tool_outcome(session, call_id, outcome, note)?.into(),
461        }),
462        Request::Ledger { session } => Ok(Response::Ledger {
463            entries: d.ledger(session)?,
464        }),
465        Request::PutMedia { bytes, mime: _ } => Ok(Response::Media {
466            hash: d.put_media(&bytes)?,
467        }),
468        Request::AppendImage {
469            session,
470            role,
471            blob,
472            pre_text,
473            post_text,
474        } => Ok(Response::Committed {
475            event: d.append_image(session, role, &blob, &pre_text, &post_text)?.into(),
476        }),
477        Request::GcMedia => {
478            let (kept, removed) = d.gc_media()?;
479            Ok(Response::Gc {
480                kept: kept as u64,
481                removed: removed as u64,
482            })
483        }
484        Request::Trim { session, to_event } => Ok(Response::Created {
485            session: d.trim(session, to_event)?,
486        }),
487        Request::Pin { session, ttl_ms } => Ok(Response::Pinned {
488            deadline_unix_ms: d.pin(session, ttl_ms)?,
489        }),
490        Request::AppendSystem {
491            session,
492            text,
493            tools,
494        } => Ok(Response::Committed {
495            event: d.append_system(session, text, tools)?.into(),
496        }),
497        Request::Complete {
498            prefix,
499            suffix,
500            mode,
501            max_tokens,
502        } => {
503            let c = d.complete(&prefix, &suffix, mode, max_tokens)?;
504            Ok(Response::Completion {
505                text: c.text,
506                tokens: c.tokens,
507                cached: c.cached,
508                expired: c.expired,
509            })
510        }
511        Request::Generate { .. }
512        | Request::Cancel { .. }
513        | Request::Subscribe { .. }
514        | Request::Unsubscribe { .. } => {
515            Err(DaemonError::Protocol("request routed to the wrong handler"))
516        }
517    })();
518    result.unwrap_or_else(|e| Response::Err {
519        message: e.to_string(),
520    })
521}
522
523pub fn serve(listener: UnixListener, daemon: Arc<Daemon>) -> Result<(), DaemonError> {
524    let cancels = daemon.cancel_registry();
525    let active = daemon.active_registry();
526    for conn in listener.incoming() {
527        let stream = conn?;
528        let daemon = Arc::clone(&daemon);
529        let cancels = Arc::clone(&cancels);
530        let active = Arc::clone(&active);
531        std::thread::spawn(move || {
532            let _ = serve_conn(stream, daemon, &cancels, &active);
533        });
534    }
535    Ok(())
536}
537
538pub(crate) type Sink = Arc<dyn Fn(&Response) -> Result<(), DaemonError> + Send + Sync>;
539
540fn send(w: &Sink, msg: &Response) -> Result<(), DaemonError> {
541    w(msg)
542}
543
544struct ConnSub {
545    session: u64,
546    bus_id: u64,
547    detached: Arc<AtomicBool>,
548}
549
550fn serve_conn(
551    mut stream: UnixStream,
552    daemon: Arc<Daemon>,
553    cancels: &crate::CancelSet,
554    active: &Mutex<HashSet<u64>>,
555) -> Result<(), DaemonError> {
556    let out = Mutex::new(stream.try_clone()?);
557    let writer: Sink = Arc::new(move |msg: &Response| {
558        write_frame(&mut *out.lock().expect("api writer"), msg)
559    });
560    let requests = std::iter::from_fn(move || read_frame::<Request>(&mut stream).transpose());
561    serve_requests(requests, writer, daemon, cancels, active)
562}
563
564pub(crate) fn serve_requests(
565    requests: impl Iterator<Item = Result<Request, DaemonError>>,
566    writer: Sink,
567    daemon: Arc<Daemon>,
568    cancels: &crate::CancelSet,
569    active: &Mutex<HashSet<u64>>,
570) -> Result<(), DaemonError> {
571    let mut subs: HashMap<u64, ConnSub> = HashMap::new();
572    let bucket_cfg = daemon.completion_bucket();
573    let mut fim_bucket =
574        crate::completion_bucket::TokenBucket::new(bucket_cfg, std::time::Instant::now());
575    let result = (|| -> Result<(), DaemonError> {
576        for req in requests {
577            match req? {
578                Request::Generate {
579                    session,
580                    max_tokens,
581                } => {
582                    let w = Arc::clone(&writer);
583                    let result = daemon.generate_streaming(session, max_tokens, |ev| {
584                        send(
585                            &w,
586                            &Response::Delta {
587                                event: ev.clone().into(),
588                            },
589                        )
590                    });
591                    let res = match result {
592                        Ok(out) => Response::Generated {
593                            events: out.events.into_iter().map(Into::into).collect(),
594                            tokens_generated: out.tokens_generated,
595                            finish: out.finish,
596                            warm_prefix: out.warm_prefix,
597                        },
598                        Err(e) => Response::Err {
599                            message: e.to_string(),
600                        },
601                    };
602                    send(&writer, &res)?;
603                }
604                Request::Cancel { session } => {
605                    send(&writer, &cancel(cancels, active, session))?;
606                }
607                Request::Subscribe {
608                    session,
609                    cursor,
610                    provisional,
611                } => {
612                    let res = subscribe(&daemon, &writer, &mut subs, session, cursor, provisional);
613                    if let Err(e) = res {
614                        send(
615                            &writer,
616                            &Response::Err {
617                                message: e.to_string(),
618                            },
619                        )?;
620                    }
621                }
622                Request::Unsubscribe { sub } => {
623                    let res = match subs.remove(&sub) {
624                        Some(cs) => {
625                            cs.detached.store(true, Ordering::Release);
626                            daemon.bus().unsubscribe(cs.session, cs.bus_id);
627                            Response::Unsubscribed
628                        }
629                        None => Response::Err {
630                            message: format!("no subscription {sub} on this connection"),
631                        },
632                    };
633                    send(&writer, &res)?;
634                }
635                req @ Request::Complete { .. } => {
636                    let res = match fim_bucket.try_take(bucket_cfg, std::time::Instant::now()) {
637                        Ok(()) => handle(&daemon, req),
638                        Err(wait) => {
639                            daemon.note_completion_throttled();
640                            Response::Throttled {
641                                retry_after_ms: crate::completion_bucket::retry_after_ms(wait),
642                            }
643                        }
644                    };
645                    send(&writer, &res)?;
646                }
647                other => {
648                    let res = handle(&daemon, other);
649                    send(&writer, &res)?;
650                }
651            }
652        }
653        Ok(())
654    })();
655    for (_, cs) in subs.drain() {
656        cs.detached.store(true, Ordering::Release);
657        daemon.bus().unsubscribe(cs.session, cs.bus_id);
658    }
659    result
660}
661
662fn subscribe(
663    daemon: &Arc<Daemon>,
664    writer: &Sink,
665    subs: &mut HashMap<u64, ConnSub>,
666    session: u64,
667    cursor: u64,
668    provisional: bool,
669) -> Result<(), DaemonError> {
670    let sub = daemon.bus().subscribe(session, provisional);
671    let backlog = match daemon.read(session, cursor) {
672        Ok(evs) => evs,
673        Err(e) => {
674            daemon.bus().unsubscribe(session, sub.id);
675            return Err(e);
676        }
677    };
678    let sub_id = sub.id;
679    let detached = Arc::new(AtomicBool::new(false));
680    subs.insert(
681        sub_id,
682        ConnSub {
683            session,
684            bus_id: sub.id,
685            detached: Arc::clone(&detached),
686        },
687    );
688    send(writer, &Response::Subscribed { sub: sub_id })?;
689    let w = Arc::clone(writer);
690    std::thread::Builder::new()
691        .name(format!("superfluid-sub-{sub_id}"))
692        .spawn(move || {
693            let mut next_id = cursor;
694            for ev in backlog {
695                next_id = ev.event_id + 1;
696                if send(
697                    &w,
698                    &Response::SubEvent {
699                        sub: sub_id,
700                        event: ev.into(),
701                    },
702                )
703                .is_err()
704                {
705                    return;
706                }
707            }
708            loop {
709                if detached.load(Ordering::Acquire) {
710                    let _ = send(
711                        &w,
712                        &Response::SubEnded {
713                            sub: sub_id,
714                            reason: "unsubscribed".into(),
715                        },
716                    );
717                    return;
718                }
719                match sub.rx.recv() {
720                    Ok(msg) if detached.load(Ordering::Acquire) => {
721                        let _ = msg;
722                        let _ = send(
723                            &w,
724                            &Response::SubEnded {
725                                sub: sub_id,
726                                reason: "unsubscribed".into(),
727                            },
728                        );
729                        return;
730                    }
731                    Ok(BusMsg::Committed(ev)) => {
732                        if ev.event_id < next_id {
733                            continue;
734                        }
735                        next_id = ev.event_id + 1;
736                        if send(
737                            &w,
738                            &Response::SubEvent {
739                                sub: sub_id,
740                                event: ev.into(),
741                            },
742                        )
743                        .is_err()
744                        {
745                            return;
746                        }
747                    }
748                    Ok(BusMsg::Provisional { channel, text }) => {
749                        if send(
750                            &w,
751                            &Response::SubProvisional {
752                                sub: sub_id,
753                                channel,
754                                text,
755                            },
756                        )
757                        .is_err()
758                        {
759                            return;
760                        }
761                    }
762                    Err(_) => {
763                        let reason = if detached.load(Ordering::Acquire) {
764                            "unsubscribed"
765                        } else {
766                            match sub.end_reason() {
767                                EndReason::Lagged => "lagged",
768                                EndReason::PublisherClosed => "publisher closed",
769                                EndReason::Purged => "purged",
770                            }
771                        };
772                        let _ = send(
773                            &w,
774                            &Response::SubEnded {
775                                sub: sub_id,
776                                reason: reason.into(),
777                            },
778                        );
779                        return;
780                    }
781                }
782            }
783        })
784        .expect("spawn subscription delivery");
785    Ok(())
786}
787
788pub struct NativeClient {
789    stream: UnixStream,
790}
791
792impl NativeClient {
793    pub fn connect(path: &std::path::Path) -> Result<NativeClient, DaemonError> {
794        Ok(NativeClient {
795            stream: UnixStream::connect(path)?,
796        })
797    }
798
799    pub fn request(&mut self, req: &Request) -> Result<Response, DaemonError> {
800        write_frame(&mut self.stream, req)?;
801        read_frame(&mut self.stream)?.ok_or(DaemonError::Protocol("daemon closed the connection"))
802    }
803
804    pub fn generate_stream(
805        &mut self,
806        session: u64,
807        max_tokens: u32,
808        mut on_delta: impl FnMut(EventMsg),
809    ) -> Result<Response, DaemonError> {
810        write_frame(
811            &mut self.stream,
812            &Request::Generate {
813                session,
814                max_tokens,
815            },
816        )?;
817        loop {
818            let res: Response = read_frame(&mut self.stream)?
819                .ok_or(DaemonError::Protocol("daemon closed mid-stream"))?;
820            match res {
821                Response::Delta { event } => on_delta(event),
822                terminal => return Ok(terminal),
823            }
824        }
825    }
826
827    pub fn cancel(&mut self, session: u64) -> Result<(), DaemonError> {
828        match self.request(&Request::Cancel { session })? {
829            Response::Cancelling => Ok(()),
830            _ => Err(DaemonError::Protocol("unexpected cancel response")),
831        }
832    }
833
834    pub fn subscribe(
835        &mut self,
836        session: u64,
837        cursor: u64,
838        provisional: bool,
839    ) -> Result<u64, DaemonError> {
840        match self.request(&Request::Subscribe {
841            session,
842            cursor,
843            provisional,
844        })? {
845            Response::Subscribed { sub } => Ok(sub),
846            Response::Err { message } => Err(DaemonError::Generation(message)),
847            _ => Err(DaemonError::Protocol("unexpected subscribe response")),
848        }
849    }
850
851    pub fn next_frame(&mut self) -> Result<Response, DaemonError> {
852        read_frame(&mut self.stream)?.ok_or(DaemonError::Protocol("daemon closed the connection"))
853    }
854}
855
856#[cfg(test)]
857mod tests {
858    use super::*;
859
860    #[test]
861    fn socket_variant_indices_are_stable() {
862        let enc = |r: &Request| postcard::to_stdvec(r).unwrap();
863        assert_eq!(enc(&Request::List), [7]);
864        assert_eq!(
865            enc(&Request::AppendMessage { session: 1, role: 1, text: "hi".into() }),
866            [2, 1, 1, 2, b'h', b'i']
867        );
868        assert_eq!(
869            enc(&Request::AppendToolResult { session: 1, call_id: 4, content: "x".into() }),
870            [5, 1, 4, 1, b'x']
871        );
872        assert_eq!(enc(&Request::Pin { session: 1, ttl_ms: 5 }), [33, 1, 5]);
873        assert_eq!(
874            enc(&Request::AppendSystem { session: 1, text: None, tools: vec!["{}".into()] }),
875            [34, 1, 0, 1, 2, b'{', b'}']
876        );
877        let res = |r: &Response| postcard::to_stdvec(r).unwrap();
878        assert_eq!(res(&Response::Err { message: "no".into() }), [13, 2, b'n', b'o']);
879        assert_eq!(res(&Response::Throttled { retry_after_ms: 9 }), [26, 9]);
880        let back: Request = postcard::from_bytes(&[3, 1, 8]).unwrap();
881        assert!(matches!(back, Request::Generate { session: 1, max_tokens: 8 }));
882    }
883}
superfluid_daemon/api.rs

superfluid_daemon/
api.rs
