sse.bend

bend-kit-llm@0.13.0.1 · 292 lines · 43 definitions

1# Server-Sent Events parser (WHATWG HTML §9.2.6), fed in pieces. Source: https://github.com/paymog/bend-kit/tree/main/llm
2import Base
3import bend-kit-bytes@0.3.3.0/bytes.bend as Bytes
4
5# name: the event type, "message" when the stream sets none. data: the data lines joined by LF.
6# id: the last event ID the stream set. name and id hold one Char per octet.
7type Message is Type:
8  Message{name: String, data: Bytes.Bytes, id: String}
9
10# The event so far: its type, its data lines each ended by LF, the last event ID, and the last retry in ms.
11type St is Type:
12  St{name: String, data: Bytes.Bytes, id: String, retry: Maybe<&2, U32>}
13
14# buf from pos: bytes not yet read as lines. No CR or LF is between pos and seen.
15# cr: the last line ended with a CR at the end of buf, so an LF first belongs to it.
16type Reader is Type:
17  Reader{buf: Bytes.Bytes, pos: U32, seen: U32, cr: Bool, st: St}
18
19type Next is Type:
20  Got{e: Message}
21  Want{}
22
23# ponytail: a leading U+FEFF is not dropped; the LLM APIs never send one.
24def reader() -> Reader:
25  Reader{Bytes.new(0), 0, 0, False{}, St{"", Bytes.new(0), "", None{}}}
26
27def feed.at(zero: Bool, buf: Bytes.Bytes, +pos: U32, +seen: U32, cr: Bool, st: St, chunk: Bytes.Bytes) -> Reader:
28  match zero:
29    case True{}:
30      Reader{Bytes.append(buf, chunk), 0, seen, cr, st}
31    case False{}:
32      Reader{Bytes.append(Bytes.snd(Bytes.slice(buf, pos, 4294967295)), chunk), 0, (U32.max(seen, pos) - pos : U32), cr, st}
33
34# Drops the lines already read, then adds chunk.
35def feed(r: Reader, chunk: Bytes.Bytes) -> Reader:
36  Reader{buf, +pos, +seen, cr, st} = r
37  feed.at(U32.is_zero(pos), buf, pos, seen, cr, st, chunk)
38
39# The last retry field, in ms.
40def retry(r: Reader) -> Reader & Maybe<&2, U32>:
41  Reader{buf, pos, seen, cr, St{name, data, id, +m}} = r
42  (Reader{buf, pos, seen, cr, St{name, data, id, m}}, m)
43
44# The first CR or LF at or after i, or the end. f counts the bytes from i to the end; r holds byte i.
45def eol(f: Nat, r: Array<U32> & U32, +i: U32) -> Array<U32> & U32:
46  match f:
47    case 0n:
48      (a, c) = r
49      (a, i)
50    case 1n+p:
51      (a, +c) = r
52      match c:
53        case 10:
54          (a, i)
55        case 13:
56          (a, i)
57        case _:
58          eol(p, Bytes.peek(a, (i + 1 : U32)), (i + 1 : U32))
59
60# The first colon at or after i, or the line's end. f counts the bytes from i to the line's end; r holds byte i.
61def colon(f: Nat, r: Array<U32> & U32, +i: U32) -> Array<U32> & U32:
62  match f:
63    case 0n:
64      (a, c) = r
65      (a, i)
66    case 1n+p:
67      (a, +c) = r
68      match c:
69        case 58:
70          (a, i)
71        case _:
72          colon(p, Bytes.peek(a, (i + 1 : U32)), (i + 1 : U32))
73
74# Base-ten digits, at most 9 of them; None for any other text.
75def digits(s: String, +acc: U32, +n: U32, bad: Bool) -> Maybe<&2, U32>:
76  match s:
77    case SNil{}:
78      Bool.pick(Maybe<&2, U32>, Bool.or(bad, Bool.or(U32.is_zero(n), U32.is_lt(9, n))), None{}, Some{acc})
79    case SCon{Chr{+c}, t}:
80      digits(t, (acc * 10 + (c - 48 : U32) : U32), (n + 1 : U32), Bool.or(bad, Bool.or(U32.is_lt(c, 48), U32.is_lt(57, c))))
81
82def has_nul(s: String) -> Bool:
83  match s:
84    case SNil{}:
85      False{}
86    case SCon{Chr{+c}, t}:
87      Bool.or(U32.is_zero(c), has_nul(t))
88
89def lf() -> Bytes.Bytes:
90  Bytes.from_string("\n")
91
92def kind.is(+k: U32, r: Array<U32> & Bool) -> Array<U32> & U32:
93  (a, +hit) = r
94  (a, Bool.pick(U32, hit, k, 0))
95
96def kind.five(+s: U32, r: Array<U32> & Bool) -> Array<U32> & U32:
97  (a, hit) = r
98  match hit:
99    case True{}:
100      (a, 2)
101    case False{}:
102      kind.is(4, Bytes.at(a, "retry", s))
103
104# The kind of the field whose n-byte name starts at s: 1 data, 2 event, 3 id, 4 retry, 0 any other.
105def kind(+n: U32, a: Array<U32>, +s: U32) -> Array<U32> & U32:
106  match n:
107    case 2:
108      kind.is(3, Bytes.at(a, "id", s))
109    case 4:
110      kind.is(1, Bytes.at(a, "data", s))
111    case 5:
112      kind.five(s, Bytes.at(a, "event", s))
113    case _:
114      (a, 0)
115
116def set.retry(m: Maybe<&2, U32>, old: Maybe<&2, U32>) -> Maybe<&2, U32>:
117  match m:
118    case None{}:
119      old
120    case Some{x}:
121      Some{x}
122
123# The event after a field of kind k with value v. An id with NUL and a retry that is not digits change nothing.
124def set(+k: U32, v: Bytes.Bytes, st: St) -> St:
125  match k:
126    case 1:
127      St{name, data, id, retry} = st
128      St{name, Bytes.append(Bytes.append(data, v), lf()), id, retry}
129    case 2:
130      St{old, data, id, retry} = st
131      St{Bytes.to_string(v), data, id, retry}
132    case 3:
133      St{name, data, old, retry} = st
134      +s = Bytes.to_string(v)
135      St{name, data, Bool.pick(String, has_nul(s), old, s), retry}
136    case 4:
137      St{name, data, id, old} = st
138      St{name, data, id, set.retry(digits(Bytes.to_string(v), 0, 0, False{}), old)}
139    case _:
140      st
141
142def field.put(+k: U32, st: St, r: Bytes.Bytes & Bytes.Bytes) -> Array<U32> & St:
143  (Bytes.Bytes{l, a}, v) = r
144  (a, set(k, v, st))
145
146def field.set(+len: U32, +v: U32, +e: U32, st: St, r: Array<U32> & U32) -> Array<U32> & St:
147  (a, +k) = r
148  field.put(k, st, Bytes.slice(Bytes.Bytes{len, a}, v, (e - v : U32)))
149
150# r holds the byte after the colon at c. The value starts after the colon and one space, if there is one.
151def field.at(+s: U32, +c: U32, +e: U32, +len: U32, st: St, r: Array<U32> & U32) -> Array<U32> & St:
152  (a, +b) = r
153  +v = Bool.pick(U32, U32.is_eq(c, e), e, Bool.pick(U32, Bool.and(U32.is_lt((c + 1 : U32), e), U32.is_eq(b, 32)), (c + 2 : U32), (c + 1 : U32)))
154  field.set(len, v, e, st, kind((c - s : U32), a, s))
155
156def field.colon(+s: U32, +e: U32, +len: U32, st: St, r: Array<U32> & U32) -> Array<U32> & St:
157  (a, +c) = r
158  field.at(s, c, e, len, st, Bytes.peek(a, (c + 1 : U32)))
159
160# The line from s to e as a field: the name up to the first colon, the value after it. A comment's name is empty.
161def field(a: Array<U32>, +len: U32, +s: U32, +e: U32, st: St) -> Array<U32> & St:
162  field.colon(s, e, len, st, colon(U32.to_nat((e - s : U32)), Bytes.peek(a, s), s))
163
164# Drops the last LF. Bytes past len stay 0.
165def trim(b: Bytes.Bytes) -> Bytes.Bytes:
166  Bytes.Bytes{+n, buf} = b
167  Bytes.Bytes{(n - 1 : U32), Bytes.poke(buf, (n - 1 : U32), 0)}
168
169def dispatch.of(empty: Bool, +name: String, data: Bytes.Bytes, +id: String, retry: Maybe<&2, U32>) -> St & Maybe<&1, Message>:
170  match empty:
171    case True{}:
172      (St{"", data, id, retry}, None{})
173    case False{}:
174      (St{"", Bytes.new(0), id, retry}, Some{Message{Bool.pick(String, String.is_empty(name), "message", name), trim(data), id}})
175
176# A blank line: the event, when it has data. Either way the event type starts over.
177def dispatch(st: St) -> St & Maybe<&1, Message>:
178  St{name, Bytes.Bytes{+n, buf}, id, retry} = st
179  dispatch.of(U32.is_zero(n), name, Bytes.Bytes{n, buf}, id, retry)
180
181# A line was read and more may follow; the bytes ran out first; or a line ended an event.
182type Line is Type:
183  LMore{a: Array<U32>, pos: U32, cr: Bool, st: St}
184  LWant{a: Array<U32>, pos: U32, cr: Bool, st: St}
185  LGot{a: Array<U32>, pos: U32, cr: Bool, st: St, e: Message}
186
187def line.got(a: Array<U32>, +next: U32, cr: Bool, r: St & Maybe<&1, Message>) -> Line:
188  (st, m) = r
189  match m:
190    case None{}:
191      LMore{a, next, cr, st}
192    case Some{e}:
193      LGot{a, next, cr, st, e}
194
195def line.set(+next: U32, cr: Bool, r: Array<U32> & St) -> Line:
196  (a, st) = r
197  LMore{a, next, cr, st}
198
199# The line from s to e; the next line starts at next.
200def line.body(blank: Bool, a: Array<U32>, +len: U32, +s: U32, +e: U32, +next: U32, cr: Bool, st: St) -> Line:
201  match blank:
202    case True{}:
203      line.got(a, next, cr, dispatch(st))
204    case False{}:
205      line.set(next, cr, field(a, len, s, e, st))
206
207def line.crlf(+len: U32, +s: U32, +e: U32, st: St, r: Array<U32> & U32) -> Line:
208  (a, +c) = r
209  line.body(U32.is_eq(s, e), a, len, s, e, Bool.pick(U32, U32.is_eq(c, 10), (e + 2 : U32), (e + 1 : U32)), False{}, st)
210
211# A CR at e. With nothing after it yet, an LF may still come.
212def line.cr(more: Bool, a: Array<U32>, +len: U32, +s: U32, +e: U32, st: St) -> Line:
213  match more:
214    case True{}:
215      line.crlf(len, s, e, st, Bytes.peek(a, (e + 1 : U32)))
216    case False{}:
217      line.body(U32.is_eq(s, e), a, len, s, e, (e + 1 : U32), True{}, st)
218
219def line.term(+len: U32, +s: U32, +e: U32, st: St, r: Array<U32> & U32) -> Line:
220  (a, +t) = r
221  match t:
222    case 13:
223      line.cr(U32.is_lt((e + 1 : U32), len), a, len, s, e, st)
224    case _:
225      line.body(U32.is_eq(s, e), a, len, s, e, (e + 1 : U32), False{}, st)
226
227def line.end(want: Bool, a: Array<U32>, +len: U32, +s: U32, +e: U32, cr: Bool, st: St) -> Line:
228  match want:
229    case True{}:
230      LWant{a, s, cr, st}
231    case False{}:
232      line.term(len, s, e, st, Bytes.peek(a, e))
233
234def line.eol(+len: U32, +s: U32, cr: Bool, st: St, r: Array<U32> & U32) -> Line:
235  (a, +e) = r
236  line.end(U32.is_eq(e, len), a, len, s, e, cr, st)
237
238# The line that starts at s. No CR or LF lies between s and seen.
239def line(a: Array<U32>, +len: U32, +s: U32, +seen: U32, cr: Bool, st: St) -> Line:
240  +i = U32.max(s, seen)
241  line.eol(len, s, cr, st, eol(U32.to_nat((len - i : U32)), Bytes.peek(a, i), i))
242
243def stop(l: Line, +len: U32) -> Reader & Next:
244  match l:
245    case LMore{a, +pos, cr, st}:
246      (Reader{Bytes.Bytes{len, a}, pos, pos, cr, st}, Want{})
247    case LWant{a, +pos, cr, st}:
248      (Reader{Bytes.Bytes{len, a}, pos, len, cr, st}, Want{})
249    case LGot{a, +pos, cr, st, e}:
250      (Reader{Bytes.Bytes{len, a}, pos, pos, cr, st}, Got{e})
251
252# f: at least the lines left, plus one.
253def loop(f: Nat, l: Line, +len: U32) -> Reader & Next:
254  match f:
255    case 0n:
256      stop(l, len)
257    case 1n+p:
258      match l:
259        case LMore{a, +pos, cr, st}:
260          loop(p, line(a, len, pos, pos, cr, st), len)
261        case LWant{a, +pos, cr, st}:
262          stop(LWant{a, pos, cr, st}, len)
263        case LGot{a, +pos, cr, st, e}:
264          stop(LGot{a, pos, cr, st, e}, len)
265
266def skip.byte(+pos: U32, r: Array<U32> & U32) -> Array<U32> & U32 & Bool:
267  (a, +c) = r
268  (a, Bool.pick(U32, U32.is_eq(c, 10), (pos + 1 : U32), pos), False{})
269
270def skip.at(some: Bool, a: Array<U32>, +pos: U32) -> Array<U32> & U32 & Bool:
271  match some:
272    case True{}:
273      skip.byte(pos, Bytes.peek(a, pos))
274    case False{}:
275      (a, pos, True{})
276
277# After a CR that ended the last piece, an LF first belongs to that CR.
278def skip(cr: Bool, a: Array<U32>, +len: U32, +pos: U32) -> Array<U32> & U32 & Bool:
279  match cr:
280    case True{}:
281      skip.at(U32.is_lt(pos, len), a, pos)
282    case False{}:
283      (a, pos, False{})
284
285def next.at(+len: U32, +seen: U32, st: St, r: Array<U32> & U32 & Bool) -> Reader & Next:
286  (a, +pos, cr) = r
287  loop(Nat.add(U32.to_nat((len - pos : U32)), 2n), line(a, len, pos, seen, cr, st), len)
288
289# The next whole event, or Want when the bytes run out first. A blank line with no data before it gives no event.
290def next(r: Reader) -> Reader & Next:
291  Reader{Bytes.Bytes{+len, a}, +pos, +seen, cr, st} = r
292  next.at(len, seen, st, skip(cr, a, len, pos))