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