llm.bend
bend-kit-llm@0.1.0.0 · 667 lines · 0 definitions
1# LLM client for the Anthropic Messages and OpenAI Chat Completions APIs, with SSE streaming. Source: https://github.com/paymog/bend-kit/tree/main/llm
2import Base
3import bend-kit-llm@0.1.0.0/sse.bend as Sse
4import bend-kit-hairpin@0.1.0.0/hairpin.bend as Hairpin
5import bend-kit-http@0.23.0.1/http.bend as Http
6# By hash, as http imports them: bend-kit-json@0.5.0.1, -bytes@0.3.0.0, -encoding@0.3.0.0.
7import bend-kit-json@0.5.0.1/json.bend as Json
8import bend-kit-bytes@0.3.0.0/bytes.bend as Bytes
9import bend-kit-encoding@0.3.0.0/encoding.bend as Enc
10
11# Anthropic: POST {base}/v1/messages. OpenAI: POST {base}/chat/completions, which many other providers serve too.
12type Api is Data:
13 Anthropic{}
14 OpenAI{}
15
16type Role is Data:
17 User{}
18 Assistant{}
19
20type Msg is Data:
21 Msg{role: Role, text: String}
22
23# system: "" for none. max: the most tokens to write.
24type Req is Data:
25 Req{model: String, system: String, msgs: List<&2, Msg>, max: U32}
26
27# stop: the API's reason, as "end_turn" or "stop". input and output count tokens.
28type Reply is Data:
29 Reply{id: String, model: String, text: String, stop: String, input: U32, output: U32}
30
31# ErrNet: the request did not complete. ErrApi: the API refused, with its status and error text; an error
32# event in a stream has status 200. ErrReply: a reply that is not the API's JSON, with its text.
33type Err is Data:
34 ErrNet{e: Http.Err}
35 ErrApi{status: U32, body: String}
36 ErrReply{body: String}
37
38# headers go on every request. ms: the step timeout. retries: extra tries after a retryable failure.
39type Client is Type:
40 Client{api: Api, base: String, headers: Map<&2, List<&2, String>>, ms: U32, retries: U32, http: Hairpin.Client}
41
42# ---- JSON
43
44def str(s: String) -> Json.Val:
45 Json.Str{Json.utf8(s)}
46
47def kv(k: String, v: Json.Val) -> Bytes.Bytes & Json.Val:
48 (Json.utf8(k), v)
49
50def num(+n: U32) -> Json.Val:
51 Json.Num{Bytes.from_string(U32.show(n))}
52
53def obj(kvs: List<&1, Bytes.Bytes & Json.Val>) -> Json.Val:
54 Json.Obj{kvs}
55
56def when(on: Bool, x: Bytes.Bytes & Json.Val, xs: List<&1, Bytes.Bytes & Json.Val>) -> List<&1, Bytes.Bytes & Json.Val>:
57 match on:
58 case True{}:
59 Con{x, xs}
60 case False{}:
61 xs
62
63def role(r: Role) -> String:
64 match r:
65 case User{}:
66 "user"
67 case Assistant{}:
68 "assistant"
69
70def msg(+r: String, t: String) -> Json.Val:
71 obj([kv("role", str(r)), kv("content", str(t))])
72
73def msgs(xs: List<&2, Msg>) -> List<&1, Json.Val>:
74 match xs:
75 case Nil{}:
76 Nil{}
77 case Con{Msg{r, t}, rest}:
78 Con{msg(role(r), t), msgs(rest)}
79
80def sys(on: Bool, s: String, xs: List<&1, Json.Val>) -> List<&1, Json.Val>:
81 match on:
82 case True{}:
83 Con{msg("system", s), xs}
84 case False{}:
85 xs
86
87# The request body for api. stream asks for SSE.
88def body(api: Api, req: Req, stream: Bool) -> Json.Val:
89 match api:
90 case Anthropic{}:
91 Req{model, +system, ms, max} = req
92 obj(Con{kv("model", str(model)), Con{kv("max_tokens", num(max)),
93 when(Bool.not(String.is_empty(system)), kv("system", str(system)),
94 Con{kv("messages", Json.Arr{msgs(ms)}), when(stream, kv("stream", Json.Flag{True{}}), Nil{})})}})
95 case OpenAI{}:
96 Req{model, +system, ms, max} = req
97 obj(Con{kv("model", str(model)), Con{kv("max_completion_tokens", num(max)),
98 Con{kv("messages", Json.Arr{sys(Bool.not(String.is_empty(system)), system, msgs(ms))}),
99 when(stream, kv("stream", Json.Flag{True{}}), Nil{})}}})
100
101# A JSON leaf: its path, as "content.0.text", and its text.
102type Leaf is Data:
103 Leaf{p: String, v: String}
104
105def path(+p: String, +k: String) -> String:
106 Bool.pick(String, String.is_empty(p), k, p ++ "." ++ k)
107
108def push.arr(xs: List<&1, Json.Val>, +p: String, +i: U32, t: List<&1, String & Json.Val>) -> List<&1, String & Json.Val>:
109 match xs:
110 case Nil{}:
111 t
112 case Con{x, rest}:
113 Con{(path(p, U32.show(i)), x), push.arr(rest, p, (i + 1 : U32), t)}
114
115def push.obj(kvs: List<&1, Bytes.Bytes & Json.Val>, +p: String, t: List<&1, String & Json.Val>) -> List<&1, String & Json.Val>:
116 match kvs:
117 case Nil{}:
118 t
119 case Con{(k, x), rest}:
120 Con{(path(p, Enc.utf8.decode(k)), x), push.obj(rest, p, t)}
121
122# The leaves under the stack's values, last first. f: at least the values left.
123def flat(f: Nat, stack: List<&1, String & Json.Val>, acc: List<&2, Leaf>) -> List<&2, Leaf>:
124 match f:
125 case 0n:
126 acc
127 case 1n+g:
128 match stack:
129 case Nil{}:
130 acc
131 case Con{(p, v), t}:
132 match v:
133 case Json.Null{}:
134 flat(g, t, acc)
135 case Json.Flag{on}:
136 flat(g, t, Con{Leaf{p, Bool.pick(String, on, "true", "false")}, acc})
137 case Json.Num{s}:
138 flat(g, t, Con{Leaf{p, Bytes.to_string(s)}, acc})
139 case Json.Str{s}:
140 flat(g, t, Con{Leaf{p, Enc.utf8.decode(s)}, acc})
141 case Json.Arr{xs}:
142 flat(g, push.arr(xs, p, 0, t), acc)
143 case Json.Obj{kvs}:
144 flat(g, push.obj(kvs, p, t), acc)
145
146def doc.of(+n: U32, m: Maybe<&1, Json.Val>) -> Maybe<&2, List<&2, Leaf>>:
147 match m:
148 case None{}:
149 None{}
150 case Some{v}:
151 Some{List.reverse(&2, Leaf, flat(Nat.add(U32.to_nat(n), 1n), [("", v)], Nil{}))}
152
153# Every non-null leaf of a JSON text in document order, keyed by its path, as ("content.0.text", "Hi").
154# Strings are decoded; numbers keep their text. None when b is not JSON. Each value takes a byte, so n + 1 steps suffice.
155def doc(b: Bytes.Bytes) -> Maybe<&2, List<&2, Leaf>>:
156 Bytes.Bytes{+n, buf} = b
157 doc.of(n, Json.parse.bytes(Bytes.Bytes{n, buf}))
158
159# The first value at path k, or "".
160def look(d: List<&2, Leaf>, +k: String) -> String:
161 match d:
162 case Nil{}:
163 ""
164 case Con{Leaf{p, v}, t}:
165 Bool.pick(String, String.eq(p, k), v, look(t, k))
166
167def count(s: String) -> U32:
168 Maybe.default(&2, U32, Json.u32.num(s), 0)
169
170def dots(s: String, +n: U32) -> U32:
171 match s:
172 case SNil{}:
173 n
174 case SCon{Chr{+c}, t}:
175 dots(t, Bool.pick(U32, U32.is_eq(c, 46), (n + 1 : U32), n))
176
177# Is p the text of a content block, "content.<i>.text"?
178def block(+p: String) -> Bool:
179 Bool.and(Bool.and(String.starts_with(p, "content."), String.ends_with(p, ".text")), U32.is_eq(dots(p, 0), 2))
180
181# The text blocks of an Anthropic message, joined.
182def blocks(d: List<&2, Leaf>) -> String:
183 match d:
184 case Nil{}:
185 ""
186 case Con{Leaf{+p, v}, t}:
187 Bool.pick(String, block(p), v, "") ++ blocks(t)
188
189def reply.of(api: Api, +d: List<&2, Leaf>) -> Reply:
190 match api:
191 case Anthropic{}:
192 Reply{look(d, "id"), look(d, "model"), blocks(d), look(d, "stop_reason"),
193 count(look(d, "usage.input_tokens")), count(look(d, "usage.output_tokens"))}
194 case OpenAI{}:
195 Reply{look(d, "id"), look(d, "model"), look(d, "choices.0.message.content"), look(d, "choices.0.finish_reason"),
196 count(look(d, "usage.prompt_tokens")), count(look(d, "usage.completion_tokens"))}
197
198def reply.got(api: Api, b: Bytes.Bytes, m: Maybe<&2, List<&2, Leaf>>) -> Result<&1, &1, Err, Reply>:
199 match m:
200 case None{}:
201 Fail{ErrReply{Enc.utf8.decode(b)}}
202 case Some{d}:
203 Done{reply.of(api, d)}
204
205def reply.copy(api: Api, r: Bytes.Bytes & Bytes.Bytes) -> Result<&1, &1, Err, Reply>:
206 (b, c) = r
207 reply.got(api, b, doc(c))
208
209# A 2xx response body as a Reply.
210def reply(api: Api, b: Bytes.Bytes) -> Result<&1, &1, Err, Reply>:
211 reply.copy(api, Bytes.slice(b, 0, 4294967295))
212
213# ---- stream events
214
215# A text piece, the end, an error, or an event with no text.
216type Ev is Data:
217 EText{s: String}
218 EEnd{}
219 EFail{e: Err}
220 ESkip{}
221
222def ev.text(+s: String) -> Ev:
223 Bool.pick(Ev, String.is_empty(s), ESkip{}, EText{s})
224
225def ev.doc(m: Maybe<&2, List<&2, Leaf>>) -> List<&2, Leaf>:
226 Maybe.default(&2, List<&2, Leaf>, m, Nil{})
227
228def ev.anthropic(+name: String, data: Bytes.Bytes) -> Ev:
229 +d = ev.doc(doc(data))
230 Bool.pick(Ev, String.eq(name, "content_block_delta"), ev.text(look(d, "delta.text")),
231 Bool.pick(Ev, String.eq(name, "message_stop"), EEnd{},
232 Bool.pick(Ev, String.eq(name, "error"), EFail{ErrApi{200, look(d, "error.message")}}, ESkip{})))
233
234def ev.openai(+s: String) -> Ev:
235 +d = ev.doc(doc(Bytes.from_string(s)))
236 +why = look(d, "error.message")
237 Bool.pick(Ev, String.eq(s, "[DONE]"), EEnd{},
238 Bool.pick(Ev, String.is_empty(why), ev.text(look(d, "choices.0.delta.content")), EFail{ErrApi{200, why}}))
239
240# What one SSE event of api means.
241def ev(api: Api, e: Sse.Message) -> Ev:
242 match api:
243 case Anthropic{}:
244 Sse.Message{name, data, id} = e
245 ev.anthropic(name, data)
246 case OpenAI{}:
247 Sse.Message{name, data, id} = e
248 ev.openai(Bytes.to_string(data))
249
250# ---- requests
251
252def ok(+s: U32) -> Bool:
253 Bool.and(U32.is_le(200, s), U32.is_lt(s, 300))
254
255# Statuses a later try may clear: 408, 409, 429, and 5xx, which holds Anthropic's 529.
256def again.status(+s: U32) -> Bool:
257 Bool.or(Bool.or(U32.is_eq(s, 408), U32.is_eq(s, 409)), Bool.or(U32.is_eq(s, 429), U32.is_le(500, s)))
258
259# A refused connect or a timeout may clear.
260def again.err(+e: Http.Err) -> Maybe<&2, String>:
261 match e:
262 case Http.ErrConnect{code, why}:
263 Some{""}
264 case Http.ErrTimeout{}:
265 Some{""}
266 case other:
267 None{}
268
269def again.of(+status: U32, +h: Map<&2, List<&2, String>>) -> Maybe<&2, String>:
270 Bool.pick(Maybe<&2, String>, again.status(status), Some{Http.header(h, "retry-after")}, None{})
271
272# The result, with Some{Retry-After} when another try may help.
273def again(r: Result<&1, &1, Http.Err, Http.Res>) -> Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>:
274 match r:
275 case Fail{+e}:
276 (Fail{e}, again.err(e))
277 case Done{res}:
278 Http.Res{+status, +h, b} = res
279 (Done{Http.Res{status, h, b}}, again.of(status, h))
280
281def post.judged(h: Hairpin.Client, body: Bytes.Bytes, j: Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>) ->
282 Hairpin.Client & Bytes.Bytes & Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>:
283 (r, m) = j
284 (h, body, r, m)
285
286def post.kept(body: Bytes.Bytes, x: Hairpin.Client & Result<&1, &1, Http.Err, Http.Res>) ->
287 Hairpin.Client & Bytes.Bytes & Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>:
288 (h, r) = x
289 post.judged(h, body, again(r))
290
291def post.once(+url: String, +headers: Map<&2, List<&2, String>>, h: Hairpin.Client, r: Bytes.Bytes & Bytes.Bytes) ->
292 IO(Hairpin.Client & Bytes.Bytes & Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>):
293 (body, spare) = r
294 do IO<Hairpin.Client & Bytes.Bytes & Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>>:
295 x : Hairpin.Client & Result<&1, &1, Http.Err, Http.Res> <- Hairpin.request(h, "POST", url, headers, spare)
296 return post.kept(body, x)
297
298# left: retries still allowed. n: the retry to wait for next, 0 first. The wait is Retry-After, else Http's backoff.
299def post.go(left: Nat, +n: Nat, +url: String, +headers: Map<&2, List<&2, String>>,
300 t: Hairpin.Client & Bytes.Bytes & Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>) ->
301 IO(Hairpin.Client & Result<&1, &1, Http.Err, Http.Res>):
302 match left:
303 case 0n:
304 (h, body, r, m) = t
305 IO.pure(Hairpin.Client & Result<&1, &1, Http.Err, Http.Res>, (h, r))
306 case 1n+rest:
307 (h, body, r, m) = t
308 match m:
309 case None{}:
310 IO.pure(Hairpin.Client & Result<&1, &1, Http.Err, Http.Res>, (h, r))
311 case Some{v}:
312 do IO<Hairpin.Client & Result<&1, &1, Http.Err, Http.Res>>:
313 Http.retry.sleep(v, n)
314 tried : Hairpin.Client & Bytes.Bytes & Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String> <-
315 post.once(url, headers, h, Bytes.slice(body, 0, 4294967295))
316 post.go(rest, 1n+n, url, headers, tried)
317
318# POST body, then retry. Hairpin retries only idempotent methods, and these POSTs are safe to repeat.
319def post(+retries: U32, +url: String, +headers: Map<&2, List<&2, String>>, h: Hairpin.Client, body: Bytes.Bytes) ->
320 IO(Hairpin.Client & Result<&1, &1, Http.Err, Http.Res>):
321 do IO<Hairpin.Client & Result<&1, &1, Http.Err, Http.Res>>:
322 first : Hairpin.Client & Bytes.Bytes & Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String> <-
323 post.once(url, headers, h, Bytes.slice(body, 0, 4294967295))
324 post.go(U32.to_nat(retries), 0n, url, headers, first)
325
326def body.ok(good: Bool, +status: U32, b: Bytes.Bytes) -> Result<&1, &1, Err, Bytes.Bytes>:
327 match good:
328 case True{}:
329 Done{b}
330 case False{}:
331 Fail{ErrApi{status, Enc.utf8.decode(b)}}
332
333# The 2xx body, or why there is none.
334def body.of(r: Result<&1, &1, Http.Err, Http.Res>) -> Result<&1, &1, Err, Bytes.Bytes>:
335 match r:
336 case Fail{e}:
337 Fail{ErrNet{e}}
338 case Done{res}:
339 Http.Res{+status, h, b} = res
340 body.ok(ok(status), status, b)
341
342def url(api: Api, +base: String) -> String:
343 match api:
344 case Anthropic{}:
345 base ++ "/v1/messages"
346 case OpenAI{}:
347 base ++ "/chat/completions"
348
349# ---- the client
350
351def new(+api: Api, base: String, headers: Map<&2, List<&2, String>>) -> Client:
352 Client{api, base, headers, 600000, 2, Hairpin.timeout(Hairpin.new(), 600000)}
353
354def json(h: Map<&2, List<&2, String>>) -> Map<&2, List<&2, String>>:
355 Http.set(h, "content-type", "application/json")
356
357# base https://api.anthropic.com, 10 min steps, 2 retries.
358def anthropic(key: String) -> Client:
359 new(Anthropic{}, "https://api.anthropic.com", json(Http.set(Http.set(Http.empty(), "x-api-key", key), "anthropic-version", "2023-06-01")))
360
361# base https://api.openai.com/v1, 10 min steps, 2 retries.
362def openai(key: String) -> Client:
363 new(OpenAI{}, "https://api.openai.com/v1", json(Http.set(Http.empty(), "authorization", "Bearer " ++ key)))
364
365# Another base URL, as for a provider that serves the same API.
366def base(c: Client, url: String) -> Client:
367 Client{api, old, h, ms, n, http} = c
368 Client{api, url, h, ms, n, http}
369
370# A header on every request, as anthropic-beta. It replaces one of the same name.
371def header(c: Client, k: String, v: String) -> Client:
372 Client{api, b, h, ms, n, http} = c
373 Client{api, b, Http.set(h, String.to_lower(k), v), ms, n, http}
374
375def timeout(c: Client, +ms: U32) -> Client:
376 Client{api, b, h, old, n, http} = c
377 Client{api, b, h, ms, n, Hairpin.timeout(http, ms)}
378
379def retry(c: Client, n: U32) -> Client:
380 Client{api, b, h, ms, old, http} = c
381 Client{api, b, h, ms, n, http}
382
383def close(c: Client) -> IO(Unit):
384 Client{api, b, h, ms, n, http} = c
385 Hairpin.close(http)
386
387def call.done(+api: Api, base: String, headers: Map<&2, List<&2, String>>, +ms: U32, +n: U32,
388 x: Hairpin.Client & Result<&1, &1, Http.Err, Http.Res>) -> Client & Result<&1, &1, Err, Bytes.Bytes>:
389 (h, r) = x
390 (Client{api, base, headers, ms, n, h}, body.of(r))
391
392# POST body to the API, with retries; the 2xx body comes back.
393def call(c: Client, body: Bytes.Bytes) -> IO(Client & Result<&1, &1, Err, Bytes.Bytes>):
394 Client{+api, +base, +headers, +ms, +n, h} = c
395 do IO<Client & Result<&1, &1, Err, Bytes.Bytes>>:
396 x : Hairpin.Client & Result<&1, &1, Http.Err, Http.Res> <- post(n, url(api, base), headers, h, body)
397 return call.done(api, base, headers, ms, n, x)
398
399def raw.got(b: Bytes.Bytes, m: Maybe<&1, Json.Val>) -> Result<&1, &1, Err, Json.Val>:
400 match m:
401 case None{}:
402 Fail{ErrReply{Enc.utf8.decode(b)}}
403 case Some{v}:
404 Done{v}
405
406def raw.copy(r: Bytes.Bytes & Bytes.Bytes) -> Result<&1, &1, Err, Json.Val>:
407 (b, c) = r
408 raw.got(b, Json.parse.bytes(c))
409
410def raw.of(r: Result<&1, &1, Err, Bytes.Bytes>) -> Result<&1, &1, Err, Json.Val>:
411 match r:
412 case Fail{e}:
413 Fail{e}
414 case Done{b}:
415 raw.copy(Bytes.slice(b, 0, 4294967295))
416
417def raw.done(x: Client & Result<&1, &1, Err, Bytes.Bytes>) -> Client & Result<&1, &1, Err, Json.Val>:
418 (c, r) = x
419 (c, raw.of(r))
420
421# Any request body, as for tools or images, and the reply as JSON.
422def send.raw(c: Client, v: Json.Val) -> IO(Client & Result<&1, &1, Err, Json.Val>):
423 do IO<Client & Result<&1, &1, Err, Json.Val>>:
424 x : Client & Result<&1, &1, Err, Bytes.Bytes> <- call(c, Json.encode.bytes(v))
425 return raw.done(x)
426
427def send.of(r: Result<&1, &1, Err, Bytes.Bytes>, +api: Api) -> Result<&1, &1, Err, Reply>:
428 match r:
429 case Fail{e}:
430 Fail{e}
431 case Done{b}:
432 reply(api, b)
433
434def send.done(+api: Api, x: Client & Result<&1, &1, Err, Bytes.Bytes>) -> Client & Result<&1, &1, Err, Reply>:
435 (c, r) = x
436 (c, send.of(r, api))
437
438# One whole reply.
439def send(c: Client, req: Req) -> IO(Client & Result<&1, &1, Err, Reply>):
440 Client{+api, b, h, ms, n, http} = c
441 do IO<Client & Result<&1, &1, Err, Reply>>:
442 x : Client & Result<&1, &1, Err, Bytes.Bytes> <- call(Client{api, b, h, ms, n, http}, Json.encode.bytes(body(api, req, False{})))
443 return send.done(api, x)
444
445# ---- streams
446
447type Stream is Type:
448 Stream{api: Api, h: Http.Stream, rd: Sse.Reader}
449
450def open.res(x: Http.Stream & Http.Res) -> Result<&1, &1, Http.Err, Http.Stream> & Maybe<&2, String>:
451 (st, res) = x
452 Http.Res{+status, +h, b} = res
453 (Done{st}, again.of(status, h))
454
455def open.judge(r: Result<&1, &1, Http.Err, Http.Stream>) -> Result<&1, &1, Http.Err, Http.Stream> & Maybe<&2, String>:
456 match r:
457 case Fail{+e}:
458 (Fail{e}, again.err(e))
459 case Done{st}:
460 open.res(Http.stream.res(st))
461
462def open.kept(body: Bytes.Bytes, j: Result<&1, &1, Http.Err, Http.Stream> & Maybe<&2, String>) -> Bytes.Bytes & Result<&1, &1, Http.Err, Http.Stream> & Maybe<&2, String>:
463 (r, m) = j
464 (body, r, m)
465
466def open.once(+url: String, +headers: Map<&2, List<&2, String>>, +ms: U32, r: Bytes.Bytes & Bytes.Bytes) ->
467 IO(Bytes.Bytes & Result<&1, &1, Http.Err, Http.Stream> & Maybe<&2, String>):
468 (body, spare) = r
469 do IO<Bytes.Bytes & Result<&1, &1, Http.Err, Http.Stream> & Maybe<&2, String>>:
470 s : Result<&1, &1, Http.Err, Http.Stream> <- Http.open.with("POST", url, headers, spare, ms)
471 return open.kept(body, open.judge(s))
472
473def open.drop(r: Result<&1, &1, Http.Err, Http.Stream>) -> IO(Unit):
474 match r:
475 case Fail{e}:
476 IO.pure(Unit, Unit{})
477 case Done{st}:
478 Http.stream.close(st)
479
480# As post.go, but a refused stream closes before the next try.
481def open.go(left: Nat, +n: Nat, +url: String, +headers: Map<&2, List<&2, String>>, +ms: U32,
482 t: Bytes.Bytes & Result<&1, &1, Http.Err, Http.Stream> & Maybe<&2, String>) -> IO(Result<&1, &1, Http.Err, Http.Stream>):
483 match left:
484 case 0n:
485 (body, r, m) = t
486 IO.pure(Result<&1, &1, Http.Err, Http.Stream>, r)
487 case 1n+rest:
488 (body, r, m) = t
489 match m:
490 case None{}:
491 IO.pure(Result<&1, &1, Http.Err, Http.Stream>, r)
492 case Some{v}:
493 do IO<Result<&1, &1, Http.Err, Http.Stream>>:
494 open.drop(r)
495 Http.retry.sleep(v, n)
496 tried : Bytes.Bytes & Result<&1, &1, Http.Err, Http.Stream> & Maybe<&2, String> <- open.once(url, headers, ms, Bytes.slice(body, 0, 4294967295))
497 open.go(rest, 1n+n, url, headers, ms, tried)
498
499# A refused stream's error body, at most f more pieces, then the stream closed.
500def drain(f: Nat, x: Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>>, acc: Bytes.Bytes) -> IO(Bytes.Bytes):
501 match f:
502 case 0n:
503 (h, r) = x
504 do IO<Bytes.Bytes>:
505 Http.stream.close(h)
506 return acc
507 case 1n+g:
508 (h, r) = x
509 match r:
510 case Fail{e}:
511 do IO<Bytes.Bytes>:
512 Http.stream.close(h)
513 return acc
514 case Done{m}:
515 match m:
516 case None{}:
517 do IO<Bytes.Bytes>:
518 Http.stream.close(h)
519 return acc
520 case Some{b}:
521 do IO<Bytes.Bytes>:
522 y : Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>> <- Http.stream.read(h)
523 drain(g, y, Bytes.append(acc, b))
524
525def start.refused(+status: U32, h: Http.Stream) -> IO(Result<&1, &1, Err, Http.Stream>):
526 do IO<Result<&1, &1, Err, Http.Stream>>:
527 y : Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>> <- Http.stream.read(h)
528 b : Bytes.Bytes <- drain(64n, y, Bytes.new(0))
529 return Fail{ErrApi{status, Enc.utf8.decode(b)}}
530
531def start.ok(good: Bool, +status: U32, h: Http.Stream) -> IO(Result<&1, &1, Err, Http.Stream>):
532 match good:
533 case True{}:
534 IO.pure(Result<&1, &1, Err, Http.Stream>, Done{h})
535 case False{}:
536 start.refused(status, h)
537
538def start.res(x: Http.Stream & Http.Res) -> IO(Result<&1, &1, Err, Http.Stream>):
539 (h, res) = x
540 Http.Res{+status, hd, b} = res
541 start.ok(ok(status), status, h)
542
543def start(r: Result<&1, &1, Http.Err, Http.Stream>) -> IO(Result<&1, &1, Err, Http.Stream>):
544 match r:
545 case Fail{e}:
546 IO.pure(Result<&1, &1, Err, Http.Stream>, Fail{ErrNet{e}})
547 case Done{h}:
548 start.res(Http.stream.res(h))
549
550def stream.of(r: Result<&1, &1, Err, Http.Stream>, +api: Api) -> Result<&1, &1, Err, Stream>:
551 match r:
552 case Fail{e}:
553 Fail{e}
554 case Done{h}:
555 Done{Stream{api, h, Sse.reader()}}
556
557# Any request body that asks for SSE, as for tools. A stream opens its own connection, outside the pool.
558def stream.raw(c: Client, v: Json.Val) -> IO(Client & Result<&1, &1, Err, Stream>):
559 Client{+api, +base, +headers, +ms, +n, h} = c
560 do IO<Client & Result<&1, &1, Err, Stream>>:
561 first : Bytes.Bytes & Result<&1, &1, Http.Err, Http.Stream> & Maybe<&2, String> <- open.once(url(api, base), headers, ms, Bytes.slice(Json.encode.bytes(v), 0, 4294967295))
562 r : Result<&1, &1, Http.Err, Http.Stream> <- open.go(U32.to_nat(n), 0n, url(api, base), headers, ms, first)
563 s : Result<&1, &1, Err, Http.Stream> <- start(r)
564 return (Client{api, base, headers, ms, n, h}, stream.of(s, api))
565
566# The reply as it is written: read it with next.
567def stream(c: Client, req: Req) -> IO(Client & Result<&1, &1, Err, Stream>):
568 Client{+api, b, h, ms, n, http} = c
569 stream.raw(Client{api, b, h, ms, n, http}, body(api, req, True{}))
570
571def stream.close(s: Stream) -> IO(Unit):
572 Stream{api, h, rd} = s
573 Http.stream.close(h)
574
575# The reader after the next event, or the stream's result.
576type Rn is Type:
577 RnNext{h: Http.Stream, r: Sse.Reader & Sse.Next}
578 RnOut{h: Http.Stream, rd: Sse.Reader, r: Result<&1, &1, Err, Maybe<&1, Sse.Message>>}
579
580def event.read(rd: Sse.Reader, x: Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>>) -> Rn:
581 (h, r) = x
582 match r:
583 case Fail{e}:
584 RnOut{h, rd, Fail{ErrNet{e}}}
585 case Done{m}:
586 match m:
587 case None{}:
588 RnOut{h, rd, Done{None{}}}
589 case Some{b}:
590 RnNext{h, Sse.next(Sse.feed(rd, b))}
591
592# f: pieces left to read before the stream counts as stalled.
593def event.loop(f: Nat, st: Rn) -> IO(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>):
594 match f:
595 case 0n:
596 match st:
597 case RnNext{h, r}:
598 (rd, n) = r
599 IO.pure(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>, (h, rd, Fail{ErrReply{"too many pieces without an event"}}))
600 case RnOut{h, rd, r}:
601 IO.pure(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>, (h, rd, r))
602 case 1n+g:
603 match st:
604 case RnNext{h, r}:
605 (rd, n) = r
606 match n:
607 case Sse.Got{e}:
608 IO.pure(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>, (h, rd, Done{Some{e}}))
609 case Sse.Want{}:
610 do IO<Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>>:
611 x : Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>> <- Http.stream.read(h)
612 event.loop(g, event.read(rd, x))
613 case RnOut{h, rd, r}:
614 IO.pure(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>, (h, rd, r))
615
616def event.done(+api: Api, x: Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>) -> Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>>:
617 (h, rd, r) = x
618 (Stream{api, h, rd}, r)
619
620# The next SSE event, or None when the stream ends. An event the end cuts off is dropped.
621def event(s: Stream) -> IO(Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>>):
622 Stream{+api, h, rd} = s
623 do IO<Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>>>:
624 x : Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>> <- event.loop(Nat.mul(65536n, 256n), RnNext{h, Sse.next(rd)})
625 return event.done(api, x)
626
627def next.ev(r: Result<&1, &1, Err, Maybe<&1, Sse.Message>>, +api: Api) -> Ev:
628 match r:
629 case Fail{e}:
630 EFail{e}
631 case Done{m}:
632 match m:
633 case None{}:
634 EEnd{}
635 case Some{e}:
636 ev(api, e)
637
638def next.of(x: Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>>) -> Stream & Ev:
639 (s, r) = x
640 Stream{+api, h, rd} = s
641 (Stream{api, h, rd}, next.ev(r, api))
642
643# f: events left to skip.
644def next.loop(f: Nat, x: Stream & Ev) -> IO(Stream & Result<&1, &1, Err, Maybe<&2, String>>):
645 match f:
646 case 0n:
647 (s, e) = x
648 IO.pure(Stream & Result<&1, &1, Err, Maybe<&2, String>>, (s, Fail{ErrReply{"too many events without text"}}))
649 case 1n+g:
650 (s, e) = x
651 match e:
652 case EText{t}:
653 IO.pure(Stream & Result<&1, &1, Err, Maybe<&2, String>>, (s, Done{Some{t}}))
654 case EEnd{}:
655 IO.pure(Stream & Result<&1, &1, Err, Maybe<&2, String>>, (s, Done{None{}}))
656 case EFail{err}:
657 IO.pure(Stream & Result<&1, &1, Err, Maybe<&2, String>>, (s, Fail{err}))
658 case ESkip{}:
659 do IO<Stream & Result<&1, &1, Err, Maybe<&2, String>>>:
660 y : Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>> <- event(s)
661 next.loop(g, next.of(y))
662
663# The next piece of reply text, or None at the end.
664def next(s: Stream) -> IO(Stream & Result<&1, &1, Err, Maybe<&2, String>>):
665 do IO<Stream & Result<&1, &1, Err, Maybe<&2, String>>>:
666 y : Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>> <- event(s)
667 next.loop(Nat.mul(65536n, 256n), next.of(y))