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.13.0.1/hairpin.bend as Hairpin
5	import bend-kit-retry@0.2.0.0/retry.bend as Retry
6	import bend-kit-http@0.38.1.1/http.bend as Http
7	import bend-kit-json@0.7.0.0/json.bend as Json
8	import bend-kit-bytes@0.3.3.0/bytes.bend as Bytes
9	import bend-kit-encoding@0.5.0.0/encoding.bend as Enc
10	
11	# Anthropic: POST {base}/v1/messages. OpenAI: POST {base}/chat/completions, which many other providers serve too.
12	type Api is Data:
13	  Anthropic{}
14	  OpenAI{}
15	
16	type Role is Data:
17	  User{}
18	  Assistant{}
19	
20	type Msg is Data:
21	  Msg{role: Role, text: String}
22	
23	# system: "" for none. max: the most tokens to write.
24	type 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.
28	type 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, or Hairpin's breaker admitted no attempt. ErrApi: the API refused, with
32	# 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.
33	type Err is Data:
34	  ErrNet{e: Hairpin.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.
39	type 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	
44	def str(s: String) -> Json.Val:
45	  Json.Str{Json.utf8(s)}
46	
47	def kv(k: String, v: Json.Val) -> Bytes.Bytes & Json.Val:
48	  (Json.utf8(k), v)
49	
50	def num(+n: U32) -> Json.Val:
51	  Json.Num{Bytes.from_string(U32.show(n))}
52	
53	def obj(kvs: List<&1, Bytes.Bytes & Json.Val>) -> Json.Val:
54	  Json.Obj{kvs}
55	
56	def 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	
63	def role(r: Role) -> String:
64	  match r:
65	    case User{}:
66	      "user"
67	    case Assistant{}:
68	      "assistant"
69	
70	def msg(+r: String, t: String) -> Json.Val:
71	  obj([kv("role", str(r)), kv("content", str(t))])
72	
73	def 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	
80	def 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.
88	def 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.
102	type Leaf is Data:
103	  Leaf{p: String, v: String}
104	
105	def path(+p: String, +k: String) -> String:
106	  Bool.pick(String, String.is_empty(p), k, p ++ "." ++ k)
107	
108	def 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	
115	def 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.
123	def 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	
146	def 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.
155	def 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 "".
160	def 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	
167	def count(s: String) -> U32:
168	  Maybe.default(&2, U32, Json.u32.num(s), 0)
169	
170	def 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"?
178	def 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.
182	def 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	
189	def 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	
198	def 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	
205	def 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.
210	def 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.
216	type Ev is Data:
217	  EText{s: String}
218	  EEnd{}
219	  EFail{e: Err}
220	  ESkip{}
221	
222	def ev.text(+s: String) -> Ev:
223	  Bool.pick(Ev, String.is_empty(s), ESkip{}, EText{s})
224	
225	def ev.doc(m: Maybe<&2, List<&2, Leaf>>) -> List<&2, Leaf>:
226	  Maybe.default(&2, List<&2, Leaf>, m, Nil{})
227	
228	def 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	
234	def 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.
241	def 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	
252	def 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.
256	def 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.
260	def 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	
269	def 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	# Again when another try may help; otherwise settled, as Hairpin.judge would settle it.
273	def verdict(m: Maybe<&2, String>, settled: Hairpin.Verdict) -> Hairpin.Verdict:
274	  match m:
275	    case Some{after}:
276	      Hairpin.VerdictAgain{after}
277	    case None{}:
278	      settled
279	
280	# Hairpin's judge for every call.
281	def again(r: Result<&1, &1, Http.Err, Http.Res>) -> Result<&1, &1, Http.Err, Http.Res> & Hairpin.Verdict:
282	  match r:
283	    case Fail{+e}:
284	      (Fail{e}, verdict(again.err(e), Hairpin.VerdictTarget{}))
285	    case Done{res}:
286	      Http.Res{+status, +h, b} = res
287	      (Done{Http.Res{status, h, b}}, verdict(again.of(status, h), Hairpin.judge.status(status)))
288	
289	def body.ok(good: Bool, +status: U32, b: Bytes.Bytes) -> Result<&1, &1, Err, Bytes.Bytes>:
290	  match good:
291	    case True{}:
292	      Done{b}
293	    case False{}:
294	      Fail{ErrApi{status, Enc.utf8.decode(b)}}
295	
296	# The 2xx body, or why there is none.
297	def body.of(r: Result<&1, &1, Hairpin.Err, Http.Res>) -> Result<&1, &1, Err, Bytes.Bytes>:
298	  match r:
299	    case Fail{e}:
300	      Fail{ErrNet{e}}
301	    case Done{res}:
302	      Http.Res{+status, h, b} = res
303	      body.ok(ok(status), status, b)
304	
305	def url(api: Api, +base: String) -> String:
306	  match api:
307	    case Anthropic{}:
308	      base ++ "/v1/messages"
309	    case OpenAI{}:
310	      base ++ "/chat/completions"
311	
312	# ---- the client
313	
314	def new(+api: Api, base: String, headers: Map<&2, List<&2, String>>) -> Client:
315	  Client{api, base, headers, 600000, 2, Hairpin.timeout(Hairpin.new(), 600000)}
316	
317	def json(h: Map<&2, List<&2, String>>) -> Map<&2, List<&2, String>>:
318	  Http.set(h, "content-type", "application/json")
319	
320	# base https://api.anthropic.com, 10 min steps, 2 retries.
321	def anthropic(key: String) -> Client:
322	  new(Anthropic{}, "https://api.anthropic.com", json(Http.set(Http.set(Http.empty(), "x-api-key", key), "anthropic-version", "2023-06-01")))
323	
324	# base https://api.openai.com/v1, 10 min steps, 2 retries.
325	def openai(key: String) -> Client:
326	  new(OpenAI{}, "https://api.openai.com/v1", json(Http.set(Http.empty(), "authorization", "Bearer " ++ key)))
327	
328	# Another base URL, as for a provider that serves the same API.
329	def base(c: Client, url: String) -> Client:
330	  Client{api, old, h, ms, n, http} = c
331	  Client{api, url, h, ms, n, http}
332	
333	# A header on every request, as anthropic-beta. It replaces one of the same name.
334	def header(c: Client, k: String, v: String) -> Client:
335	  Client{api, b, h, ms, n, http} = c
336	  Client{api, b, Http.set(h, String.to_lower(k), v), ms, n, http}
337	
338	def timeout(c: Client, +ms: U32) -> Client:
339	  Client{api, b, h, old, n, http} = c
340	  Client{api, b, h, ms, n, Hairpin.timeout(http, ms)}
341	
342	def retry(c: Client, n: U32) -> Client:
343	  Client{api, b, h, ms, old, http} = c
344	  Client{api, b, h, ms, n, http}
345	
346	def close(c: Client) -> IO(Unit):
347	  Client{api, b, h, ms, n, http} = c
348	  Hairpin.close(http)
349	
350	def call.done(+api: Api, base: String, headers: Map<&2, List<&2, String>>, +ms: U32, +n: U32,
351	    x: Hairpin.Client & Retry.Budget & Result<&1, &1, Hairpin.Err, Http.Res>) -> Client & Result<&1, &1, Err, Bytes.Bytes>:
352	  (h, b, r) = x
353	  (Client{api, base, headers, ms, n, h}, body.of(r))
354	
355	def call.budget(+api: Api, +base: String, +headers: Map<&2, List<&2, String>>, body: Bytes.Bytes, y: Hairpin.Client & Retry.Budget) ->
356	    IO(Hairpin.Client & Retry.Budget & Result<&1, &1, Hairpin.Err, Http.Res>):
357	  (h, b) = y
358	  do IO<Hairpin.Client & Retry.Budget & Result<&1, &1, Hairpin.Err, Http.Res>>:
359	    enc : String <- Http.codings()
360	    Hairpin.request.as(~again, h, b, True{}, enc, "POST", url(api, base), headers, body)
361	
362	# POST body to the API with n retries on a budget of n + 1; the 2xx body comes back.
363	# These POSTs are safe to repeat, so Hairpin retries them as again judges.
364	def call(c: Client, body: Bytes.Bytes) -> IO(Client & Result<&1, &1, Err, Bytes.Bytes>):
365	  Client{+api, +base, +headers, +ms, +n, h} = c
366	  do IO<Client & Result<&1, &1, Err, Bytes.Bytes>>:
367	    y : Hairpin.Client & Retry.Budget <- Hairpin.budget(Hairpin.retry(h, n))
368	    x : Hairpin.Client & Retry.Budget & Result<&1, &1, Hairpin.Err, Http.Res> <- call.budget(api, base, headers, body, y)
369	    return call.done(api, base, headers, ms, n, x)
370	
371	def raw.got(b: Bytes.Bytes, m: Maybe<&1, Json.Val>) -> Result<&1, &1, Err, Json.Val>:
372	  match m:
373	    case None{}:
374	      Fail{ErrReply{Enc.utf8.decode(b)}}
375	    case Some{v}:
376	      Done{v}
377	
378	def raw.copy(r: Bytes.Bytes & Bytes.Bytes) -> Result<&1, &1, Err, Json.Val>:
379	  (b, c) = r
380	  raw.got(b, Json.parse.bytes(c))
381	
382	def raw.of(r: Result<&1, &1, Err, Bytes.Bytes>) -> Result<&1, &1, Err, Json.Val>:
383	  match r:
384	    case Fail{e}:
385	      Fail{e}
386	    case Done{b}:
387	      raw.copy(Bytes.slice(b, 0, 4294967295))
388	
389	def raw.done(x: Client & Result<&1, &1, Err, Bytes.Bytes>) -> Client & Result<&1, &1, Err, Json.Val>:
390	  (c, r) = x
391	  (c, raw.of(r))
392	
393	# Any request body, as for tools or images, and the reply as JSON.
394	def send.raw(c: Client, v: Json.Val) -> IO(Client & Result<&1, &1, Err, Json.Val>):
395	  do IO<Client & Result<&1, &1, Err, Json.Val>>:
396	    x : Client & Result<&1, &1, Err, Bytes.Bytes> <- call(c, Json.encode.bytes(v))
397	    return raw.done(x)
398	
399	def send.of(r: Result<&1, &1, Err, Bytes.Bytes>, +api: Api) -> Result<&1, &1, Err, Reply>:
400	  match r:
401	    case Fail{e}:
402	      Fail{e}
403	    case Done{b}:
404	      reply(api, b)
405	
406	def send.done(+api: Api, x: Client & Result<&1, &1, Err, Bytes.Bytes>) -> Client & Result<&1, &1, Err, Reply>:
407	  (c, r) = x
408	  (c, send.of(r, api))
409	
410	# One whole reply.
411	def send(c: Client, req: Req) -> IO(Client & Result<&1, &1, Err, Reply>):
412	  Client{+api, b, h, ms, n, http} = c
413	  do IO<Client & Result<&1, &1, Err, Reply>>:
414	    x : Client & Result<&1, &1, Err, Bytes.Bytes> <- call(Client{api, b, h, ms, n, http}, Json.encode.bytes(body(api, req, False{})))
415	    return send.done(api, x)
416	
417	# ---- streams
418	
419	type Stream is Type:
420	  Stream{api: Api, h: Http.Stream, rd: Sse.Reader}
421	
422	def open.res(x: Http.Stream & Http.Res) -> Result<&1, &1, Http.Err, Http.Stream> & Hairpin.Verdict:
423	  (st, res) = x
424	  Http.Res{+status, +h, b} = res
425	  (Done{st}, verdict(again.of(status, h), Hairpin.judge.status(status)))
426	
427	def open.judge(r: Result<&1, &1, Http.Err, Http.Stream>) -> Result<&1, &1, Http.Err, Http.Stream> & Hairpin.Verdict:
428	  match r:
429	    case Fail{+e}:
430	      (Fail{e}, verdict(again.err(e), Hairpin.VerdictTarget{}))
431	    case Done{st}:
432	      open.res(Http.stream.res(st))
433	
434	# A refused stream's error body, at most f more pieces, then the stream closed.
435	def drain(f: Nat, x: Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>>, acc: Bytes.Bytes) -> IO(Bytes.Bytes):
436	  match f:
437	    case 0n:
438	      (h, r) = x
439	      do IO<Bytes.Bytes>:
440	        Http.stream.close(h)
441	        return acc
442	    case 1n+g:
443	      (h, r) = x
444	      match r:
445	        case Fail{e}:
446	          do IO<Bytes.Bytes>:
447	            Http.stream.close(h)
448	            return acc
449	        case Done{m}:
450	          match m:
451	            case None{}:
452	              do IO<Bytes.Bytes>:
453	                Http.stream.close(h)
454	                return acc
455	            case Some{b}:
456	              do IO<Bytes.Bytes>:
457	                y : Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>> <- Http.stream.read(h)
458	                drain(g, y, Bytes.append(acc, b))
459	
460	def start.refused(+status: U32, h: Http.Stream) -> IO(Result<&1, &1, Err, Http.Stream>):
461	  do IO<Result<&1, &1, Err, Http.Stream>>:
462	    y : Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>> <- Http.stream.read(h)
463	    b : Bytes.Bytes <- drain(64n, y, Bytes.new(0))
464	    return Fail{ErrApi{status, Enc.utf8.decode(b)}}
465	
466	def start.ok(good: Bool, +status: U32, h: Http.Stream) -> IO(Result<&1, &1, Err, Http.Stream>):
467	  match good:
468	    case True{}:
469	      IO.pure(Result<&1, &1, Err, Http.Stream>, Done{h})
470	    case False{}:
471	      start.refused(status, h)
472	
473	def start.res(x: Http.Stream & Http.Res) -> IO(Result<&1, &1, Err, Http.Stream>):
474	  (h, res) = x
475	  Http.Res{+status, hd, b} = res
476	  start.ok(ok(status), status, h)
477	
478	def start(r: Result<&1, &1, Hairpin.Err, Http.Stream>) -> IO(Result<&1, &1, Err, Http.Stream>):
479	  match r:
480	    case Fail{e}:
481	      IO.pure(Result<&1, &1, Err, Http.Stream>, Fail{ErrNet{e}})
482	    case Done{h}:
483	      start.res(Http.stream.res(h))
484	
485	def stream.of(r: Result<&1, &1, Err, Http.Stream>, +api: Api) -> Result<&1, &1, Err, Stream>:
486	  match r:
487	    case Fail{e}:
488	      Fail{e}
489	    case Done{h}:
490	      Done{Stream{api, h, Sse.reader()}}
491	
492	def stream.done(+api: Api, base: String, headers: Map<&2, List<&2, String>>, +ms: U32, +n: U32,
493	    x: Hairpin.Client & Retry.Budget & Result<&1, &1, Hairpin.Err, Http.Stream>) -> IO(Client & Result<&1, &1, Err, Stream>):
494	  (h, b, r) = x
495	  do IO<Client & Result<&1, &1, Err, Stream>>:
496	    s : Result<&1, &1, Err, Http.Stream> <- start(r)
497	    return (Client{api, base, headers, ms, n, h}, stream.of(s, api))
498	
499	def open.budget(+api: Api, +base: String, +headers: Map<&2, List<&2, String>>, body: Bytes.Bytes, y: Hairpin.Client & Retry.Budget) ->
500	    IO(Hairpin.Client & Retry.Budget & Result<&1, &1, Hairpin.Err, Http.Stream>):
501	  (h, b) = y
502	  do IO<Hairpin.Client & Retry.Budget & Result<&1, &1, Hairpin.Err, Http.Stream>>:
503	    enc : String <- Http.codings()
504	    Hairpin.open.as(~open.judge, h, b, True{}, enc, "POST", url(api, base), headers, body)
505	
506	# Any request body that asks for SSE, as for tools. The open retries as call does, on a budget of n + 1.
507	# A stream opens on the client's pool. Close it with stream.close to give its connection back.
508	def stream.raw(c: Client, v: Json.Val) -> IO(Client & Result<&1, &1, Err, Stream>):
509	  Client{+api, +base, +headers, +ms, +n, h} = c
510	  do IO<Client & Result<&1, &1, Err, Stream>>:
511	    y : Hairpin.Client & Retry.Budget <- Hairpin.budget(Hairpin.retry(h, n))
512	    x : Hairpin.Client & Retry.Budget & Result<&1, &1, Hairpin.Err, Http.Stream> <- open.budget(api, base, headers, Json.encode.bytes(v), y)
513	    stream.done(api, base, headers, ms, n, x)
514	
515	# The reply as it is written: read it with next.
516	def stream(c: Client, req: Req) -> IO(Client & Result<&1, &1, Err, Stream>):
517	  Client{+api, b, h, ms, n, http} = c
518	  stream.raw(Client{api, b, h, ms, n, http}, body(api, req, True{}))
519	
520	# Close a stream. A stream read to its end gives its connection to the client's pool.
521	def stream.close(c: Client, s: Stream) -> IO(Client):
522	  Client{api, base, headers, ms, n, h} = c
523	  Stream{sa, st, rd} = s
524	  do IO<Client>:
525	    p : Hairpin.Client <- Hairpin.stream.close(h, st)
526	    return Client{api, base, headers, ms, n, p}
527	
528	# The reader after the next event, or the stream's result.
529	type Rn is Type:
530	  RnNext{h: Http.Stream, r: Sse.Reader & Sse.Next}
531	  RnOut{h: Http.Stream, rd: Sse.Reader, r: Result<&1, &1, Err, Maybe<&1, Sse.Message>>}
532	
533	def event.read(rd: Sse.Reader, x: Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>>) -> Rn:
534	  (h, r) = x
535	  match r:
536	    case Fail{e}:
537	      RnOut{h, rd, Fail{ErrNet{Hairpin.ErrHttp{e}}}}
538	    case Done{m}:
539	      match m:
540	        case None{}:
541	          RnOut{h, rd, Done{None{}}}
542	        case Some{b}:
543	          RnNext{h, Sse.next(Sse.feed(rd, b))}
544	
545	# f: pieces left to read before the stream counts as stalled.
546	def event.loop(f: Nat, st: Rn) -> IO(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>):
547	  match f:
548	    case 0n:
549	      match st:
550	        case RnNext{h, r}:
551	          (rd, n) = r
552	          IO.pure(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>, (h, rd, Fail{ErrReply{"too many pieces without an event"}}))
553	        case RnOut{h, rd, r}:
554	          IO.pure(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>, (h, rd, r))
555	    case 1n+g:
556	      match st:
557	        case RnNext{h, r}:
558	          (rd, n) = r
559	          match n:
560	            case Sse.Got{e}:
561	              IO.pure(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>, (h, rd, Done{Some{e}}))
562	            case Sse.Want{}:
563	              do IO<Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>>:
564	                x : Http.Stream & Result<&1, &1, Http.Err, Maybe<&1, Bytes.Bytes>> <- Http.stream.read(h)
565	                event.loop(g, event.read(rd, x))
566	        case RnOut{h, rd, r}:
567	          IO.pure(Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>>, (h, rd, r))
568	
569	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>>:
570	  (h, rd, r) = x
571	  (Stream{api, h, rd}, r)
572	
573	# The next SSE event, or None when the stream ends. An event the end cuts off is dropped.
574	def event(s: Stream) -> IO(Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>>):
575	  Stream{+api, h, rd} = s
576	  do IO<Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>>>:
577	    x : Http.Stream & Sse.Reader & Result<&1, &1, Err, Maybe<&1, Sse.Message>> <- event.loop(Nat.mul(65536n, 256n), RnNext{h, Sse.next(rd)})
578	    return event.done(api, x)
579	
580	def next.ev(r: Result<&1, &1, Err, Maybe<&1, Sse.Message>>, +api: Api) -> Ev:
581	  match r:
582	    case Fail{e}:
583	      EFail{e}
584	    case Done{m}:
585	      match m:
586	        case None{}:
587	          EEnd{}
588	        case Some{e}:
589	          ev(api, e)
590	
591	def next.of(x: Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>>) -> Stream & Ev:
592	  (s, r) = x
593	  Stream{+api, h, rd} = s
594	  (Stream{api, h, rd}, next.ev(r, api))
595	
596	# f: events left to skip.
597	def next.loop(f: Nat, x: Stream & Ev) -> IO(Stream & Result<&1, &1, Err, Maybe<&2, String>>):
598	  match f:
599	    case 0n:
600	      (s, e) = x
601	      IO.pure(Stream & Result<&1, &1, Err, Maybe<&2, String>>, (s, Fail{ErrReply{"too many events without text"}}))
602	    case 1n+g:
603	      (s, e) = x
604	      match e:
605	        case EText{t}:
606	          IO.pure(Stream & Result<&1, &1, Err, Maybe<&2, String>>, (s, Done{Some{t}}))
607	        case EEnd{}:
608	          IO.pure(Stream & Result<&1, &1, Err, Maybe<&2, String>>, (s, Done{None{}}))
609	        case EFail{err}:
610	          IO.pure(Stream & Result<&1, &1, Err, Maybe<&2, String>>, (s, Fail{err}))
611	        case ESkip{}:
612	          do IO<Stream & Result<&1, &1, Err, Maybe<&2, String>>>:
613	            y : Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>> <- event(s)
614	            next.loop(g, next.of(y))
615	
616	# The next piece of reply text, or None at the end.
617	def next(s: Stream) -> IO(Stream & Result<&1, &1, Err, Maybe<&2, String>>):
618	  do IO<Stream & Result<&1, &1, Err, Maybe<&2, String>>>:
619	    y : Stream & Result<&1, &1, Err, Maybe<&1, Sse.Message>> <- event(s)
620	    next.loop(Nat.mul(65536n, 256n), next.of(y))