1	# ezmcp/src/io: the effects. Every decision is src/web.bend's, src/line.bend's,
2	# src/server.bend's or src/client.bend's; this module reads, runs the tool
3	# runner, and writes, in order (MCP-TRUST-1). It reaches ezhttp's socket
4	# effects, so no law file imports it.
5	import Base
6	import 0xa81e161008ecc2f297c0a3065cb9b188/main.bend as Json
7	import 0xa81e161008ecc2f297c0a3065cb9b188/src/value.bend as Value
8	import 0xedb2120a04a47dc18c3a756ec7c62750/main.bend as Ez
9	import 0xedb2120a04a47dc18c3a756ec7c62750/src/http.bend as Http
10	import 0xedb2120a04a47dc18c3a756ec7c62750/src/client.bend as EzClient
11	import 0xedb2120a04a47dc18c3a756ec7c62750/src/server.bend as EzServer
12	import ./rpc.bend as Rpc
13	import ./server.bend as Server
14	import ./web.bend as Web
15	import ./line.bend as Line
16	import ./client.bend as Client
17	
18	# --- running acts ---
19	
20	# one element of a batch, under revision `rev`: its reply, after running its
21	# tool when it is a tool call; none for a notification or a response
22	def act.reply(
23	  ~C: Data,
24	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
25	  +ctx: C,
26	  +rev: String,
27	  act: Server.Act
28	) -> IO(Maybe<&2, Value.Json>):
29	  match act:
30	    case Server.Quiet{}:
31	      IO.pure(Maybe<&2, Value.Json>, None{})
32	    case Server.Send{reply}:
33	      IO.pure(Maybe<&2, Value.Json>, Some{reply})
34	    case Server.Fault{reply}:
35	      IO.pure(Maybe<&2, Value.Json>, Some{reply})
36	    case Server.Run{id, name, args}:
37	      IO.bind(Rpc.Out, Maybe<&2, Value.Json>, run(ctx, name, args),
38	        out => IO.pure(Maybe<&2, Value.Json>, Some{Server.ran(rev, id, out)}))
39	    case Server.Batch{acts}:
40	      IO.pure(Maybe<&2, Value.Json>, None{})
41	
42	# every element of a batch, in order
43	def act.all(
44	  ~C: Data,
45	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
46	  +ctx: C,
47	  +rev: String,
48	  acts: List<&2, Server.Act>
49	) -> IO(List<&2, Maybe<&2, Value.Json>>):
50	  match acts:
51	    case []:
52	      IO.pure(List<&2, Maybe<&2, Value.Json>>, [])
53	    case a <> rest:
54	      IO.bind(Maybe<&2, Value.Json>, List<&2, Maybe<&2, Value.Json>>,
55	        act.reply(C, run, ctx, rev, a),
56	        r => IO.bind(List<&2, Maybe<&2, Value.Json>>,
57	          List<&2, Maybe<&2, Value.Json>>, act.all(C, run, ctx, rev, rest),
58	          rs => IO.pure(List<&2, Maybe<&2, Value.Json>>, r <> rs)))
59	
60	# --- streamable HTTP ---
61	
62	# act on the transport's decision: reply now, or run the tool and reply
63	# with its result, or run a batch's acts and reply with their replies
64	def http.web(
65	  ~C: Data,
66	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
67	  +ctx: C,
68	  wid: Web.Web
69	) -> IO(Http.Reply):
70	  match wid:
71	    case Web.Now{status, headers, body}:
72	      IO.pure(Http.Reply, Http.Reply{status, headers, body})
73	    case Web.Later{rev, id, name, args}:
74	      IO.bind(Rpc.Out, Http.Reply, run(ctx, name, args),
75	        out => IO.pure(Http.Reply, Web.reply(Web.ran(rev, id, out))))
76	    case Web.Many{+rev, acts}:
77	      IO.bind(List<&2, Maybe<&2, Value.Json>>, Http.Reply,
78	        act.all(C, run, ctx, rev, acts),
79	        rs => IO.pure(Http.Reply, Web.reply(Web.many(Server.batch.reply(rs)))))
80	
81	# one HTTP request to a server's endpoint at `path`
82	def http.handle_with(
83	  ~C: Data,
84	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
85	  ~srv: Server.Server,
86	  ~path: String,
87	  ~origins: List<&2, String>,
88	  +ctx: C,
89	  req: Http.Request
90	) -> IO(Http.Reply):
91	  http.web(C, run, ctx, Web.decide(srv, path, origins, req))
92	
93	# --- serving HTTP ---
94	#
95	# ezhttp serves each connection on its own fork, reads and parses the
96	# request, answers one it answers itself (a request that does not parse),
97	# and calls the handler with the context for the rest (MCP-TRUST-2); the
98	# handler below is the endpoint's
99	
100	# the endpoint's answer to one request ezhttp read, as its plain serves
101	# take one: the request as text in, a text reply out
102	def http.answer(
103	  ~C: Data,
104	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
105	  ~srv: Server.Server,
106	  ~path: String,
107	  ~origins: List<&2, String>,
108	  +ctx: C,
109	  req: EzServer.Req
110	) -> IO(EzServer.Answer):
111	  IO.bind(Http.Reply, EzServer.Answer,
112	    http.handle_with(C, run, srv, path, origins, ctx, EzServer.server.plain(req)),
113	    EzServer.server.text)
114	
115	# serve `limit` connections, at most `max` at once (0: no limit), then stop
116	def http.serve_with(
117	  ~C: Data,
118	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
119	  ~srv: Server.Server,
120	  ~path: String,
121	  ~origins: List<&2, String>,
122	  +ctx: C,
123	  host: String,
124	  port: U32,
125	  limit: U32,
126	  +max: U32
127	) -> IO(Unit):
128	  Ez.http.serve_io_with(~C,
129	    ~(c => req => http.answer(C, run, srv, path, origins, c, req)), ctx, host,
130	    port, limit, max)
131	
132	# serve until the process is stopped, at most `max` connections at once
133	# (0: no limit)
134	@unsafe def http.serve_forever_with(
135	  ~C: Data,
136	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
137	  ~srv: Server.Server,
138	  ~path: String,
139	  ~origins: List<&2, String>,
140	  +ctx: C,
141	  host: String,
142	  port: U32,
143	  +max: U32
144	) -> IO(Unit):
145	  Ez.http.serve_forever_with(~C,
146	    ~(c => req => http.answer(C, run, srv, path, origins, c, req)), ctx, host,
147	    port, max)
148	
149	# --- with the server in the context ---
150	#
151	# A server passed as `~srv` is a closed template too, so one built at run
152	# time (from a document read at startup, say) rides in the context: `srv`
153	# reads it from the context on every request
154	
155	# the endpoint's answer to one request ezhttp read, its server read from the
156	# context
157	def http.answer_from(
158	  ~C: Data,
159	  ~srv: C -> Server.Server,
160	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
161	  ~path: String,
162	  ~origins: List<&2, String>,
163	  +ctx: C,
164	  req: EzServer.Req
165	) -> IO(EzServer.Answer):
166	  IO.bind(Http.Reply, EzServer.Answer,
167	    http.web(C, run, ctx, Web.decide(srv(ctx), path, origins,
168	      EzServer.server.plain(req))),
169	    EzServer.server.text)
170	
171	# serve `limit` connections, at most `max` at once (0: no limit), then stop
172	def http.serve_from(
173	  ~C: Data,
174	  ~srv: C -> Server.Server,
175	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
176	  ~path: String,
177	  ~origins: List<&2, String>,
178	  +ctx: C,
179	  host: String,
180	  port: U32,
181	  limit: U32,
182	  +max: U32
183	) -> IO(Unit):
184	  Ez.http.serve_io_with(~C,
185	    ~(c => req => http.answer_from(C, srv, run, path, origins, c, req)), ctx,
186	    host, port, limit, max)
187	
188	# serve until the process is stopped, at most `max` connections at once
189	# (0: no limit)
190	@unsafe def http.serve_forever_from(
191	  ~C: Data,
192	  ~srv: C -> Server.Server,
193	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
194	  ~path: String,
195	  ~origins: List<&2, String>,
196	  +ctx: C,
197	  host: String,
198	  port: U32,
199	  +max: U32
200	) -> IO(Unit):
201	  Ez.http.serve_forever_with(~C,
202	    ~(c => req => http.answer_from(C, srv, run, path, origins, c, req)), ctx,
203	    host, port, max)
204	
205	# --- with no context ---
206	
207	# one HTTP request to a server's endpoint at `path`
208	def http.handle(
209	  ~run: String -> Value.Json -> IO(Rpc.Out),
210	  ~srv: Server.Server,
211	  ~path: String,
212	  ~origins: List<&2, String>,
213	  req: Http.Request
214	) -> IO(Http.Reply):
215	  http.handle_with(~Unit, ~(_c => n => a => run(n, a)), srv, path, origins,
216	    Unit{}, req)
217	
218	# serve `limit` connections, as many at once as arrive, then stop
219	def http.serve(
220	  ~run: String -> Value.Json -> IO(Rpc.Out),
221	  ~srv: Server.Server,
222	  ~path: String,
223	  ~origins: List<&2, String>,
224	  host: String,
225	  port: U32,
226	  limit: U32
227	) -> IO(Unit):
228	  http.serve_with(~Unit, ~(_c => n => a => run(n, a)), srv, path, origins,
229	    Unit{}, host, port, limit, 0)
230	
231	# serve until the process is stopped, as many connections at once as arrive
232	@unsafe def http.serve_forever(
233	  ~run: String -> Value.Json -> IO(Rpc.Out),
234	  ~srv: Server.Server,
235	  ~path: String,
236	  ~origins: List<&2, String>,
237	  host: String,
238	  port: U32
239	) -> IO(Unit):
240	  http.serve_forever_with(~Unit, ~(_c => n => a => run(n, a)), srv, path,
241	    origins, Unit{}, host, port, 0)
242	
243	# --- stdio ---
244	
245	# up to `max` bytes from file descriptor 0, as one read(2) returns them;
246	# empty at the end of input (MCP-TRUST-5)
247	def ezmcp.stdin(max: U32) -> IO(Result<&1, &1, U32 & String, List<&2, U32>>):
248	  import "./effs/stdin.c"
249	  import "./effs/stdin.js"
250	
251	# bytes one read of stdin asks for
252	def stdio.chunk() -> U32:
253	  65536
254	
255	# a batch's replies: nothing when there are none, else one line
256	def stdio.batch(replies: Maybe<&2, Value.Json>) -> IO(Unit):
257	  match replies:
258	    case None{}:
259	      IO.pure(Unit, Unit{})
260	    case Some{batch}:
261	      IO.write(Line.frame(batch))
262	
263	# act on one line's act, read by revision `rev`: write a reply as one line,
264	# or run the tool and write its reply, or run a batch's acts and write the
265	# array of their replies, or write nothing
266	def stdio.act(
267	  ~C: Data,
268	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
269	  +ctx: C,
270	  +rev: String,
271	  act: Server.Act
272	) -> IO(Unit):
273	  match act:
274	    case Server.Quiet{}:
275	      IO.pure(Unit, Unit{})
276	    case Server.Send{reply}:
277	      IO.write(Line.frame(reply))
278	    case Server.Fault{reply}:
279	      IO.write(Line.frame(reply))
280	    case Server.Run{id, name, args}:
281	      IO.bind(Rpc.Out, Unit, run(ctx, name, args),
282	        out => IO.write(Line.frame(Server.ran(rev, id, out))))
283	    case Server.Batch{acts}:
284	      IO.bind(List<&2, Maybe<&2, Value.Json>>, Unit, act.all(C, run, ctx, rev, acts),
285	        rs => stdio.batch(Server.batch.reply(rs)))
286	
287	# act on a turn read by revision `rev`, then go on with the revision after it
288	def stdio.turn(
289	  ~C: Data,
290	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
291	  +ctx: C,
292	  +rev: String,
293	  turn: Server.Turn,
294	  next: String -> IO(String)
295	) -> IO(String):
296	  match turn:
297	    case Server.Turn{act, after}:
298	      IO.bind(Unit, String, stdio.act(C, run, ctx, rev, act), _u => next(after))
299	
300	# every whole line, in order, each read by the revision the line before it
301	# left; answers the revision the last one left
302	def stdio.lines(
303	  ~C: Data,
304	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
305	  ~srv: Server.Server,
306	  +ctx: C,
307	  lines: List<&2, List<&2, U32>>,
308	  +rev: String
309	) -> IO(String):
310	  match lines:
311	    case []:
312	      IO.pure(String, rev)
313	    case line <> rest:
314	      stdio.turn(C, run, ctx, rev, Line.line(srv, rev, line),
315	        after => stdio.lines(C, run, srv, ctx, rest, after))
316	
317	# act on a cut's whole lines, then read on with its rest
318	def stdio.cut(
319	  ~C: Data,
320	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
321	  ~srv: Server.Server,
322	  +ctx: C,
323	  rev: String,
324	  again: List<&2, U32> -> String -> IO(Unit),
325	  cut: Line.Cut
326	) -> IO(Unit):
327	  match cut:
328	    case Line.Cut{lines, rest}:
329	      IO.bind(String, Unit, stdio.lines(C, run, srv, ctx, lines, rev),
330	        after => again(rest, after))
331	
332	# after a read: an empty one is the end of stdin, and the bytes after the
333	# last newline are the last message; any other is cut into lines
334	def stdio.chunked(
335	  ~C: Data,
336	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
337	  ~srv: Server.Server,
338	  +ctx: C,
339	  rev: String,
340	  buf: List<&2, U32>,
341	  bytes: List<&2, U32>,
342	  again: List<&2, U32> -> String -> IO(Unit),
343	  empty: Bool
344	) -> IO(Unit):
345	  match empty:
346	    case True{}:
347	      IO.bind(String, Unit, stdio.lines(C, run, srv, ctx, [buf], rev),
348	        _r => IO.pure(Unit, Unit{}))
349	    case False{}:
350	      stdio.cut(C, run, srv, ctx, rev, again,
351	        Line.cut(List.append(&2, U32, buf, bytes)))
352	
353	# after one read: a failure is logged to stderr and ends the loop
354	def stdio.got(
355	  ~C: Data,
356	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
357	  ~srv: Server.Server,
358	  +ctx: C,
359	  rev: String,
360	  buf: List<&2, U32>,
361	  again: List<&2, U32> -> String -> IO(Unit),
362	  res: Result<&1, &1, U32 & String, List<&2, U32>>
363	) -> IO(Unit):
364	  match res:
365	    case Fail{(_code, why)}:
366	      IO.print_err("ezmcp: stdin: " ++ why)
367	    case Done{+bytes}:
368	      stdio.chunked(C, run, srv, ctx, rev, buf, bytes, again,
369	        List.is_empty(&2, U32, bytes))
370	
371	# read, act on every whole line, and read again until stdin ends; `rev` is
372	# the revision the next line is read by
373	@unsafe def stdio.loop(
374	  ~C: Data,
375	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
376	  ~srv: Server.Server,
377	  +ctx: C,
378	  buf: List<&2, U32>,
379	  rev: String
380	) -> IO(Unit):
381	  IO.bind(Result<&1, &1, U32 & String, List<&2, U32>>, Unit,
382	    ezmcp.stdin(stdio.chunk()),
383	    res => stdio.got(C, run, srv, ctx, rev, buf,
384	      rest => after => stdio.loop(C, run, srv, ctx, rest, after), res))
385	
386	# serve newline-delimited JSON-RPC on stdin and stdout until stdin ends; a
387	# line is read by the latest revision until an initialize names another
388	@unsafe def stdio.serve_with(
389	  ~C: Data,
390	  ~run: C -> String -> Value.Json -> IO(Rpc.Out),
391	  ~srv: Server.Server,
392	  +ctx: C
393	) -> IO(Unit):
394	  stdio.loop(C, run, srv, ctx, [], Server.latest())
395	
396	# serve newline-delimited JSON-RPC on stdin and stdout until stdin ends,
397	# with no context
398	@unsafe def stdio.serve(
399	  ~run: String -> Value.Json -> IO(Rpc.Out),
400	  ~srv: Server.Server
401	) -> IO(Unit):
402	  stdio.serve_with(~Unit, ~(_c => n => a => run(n, a)), srv, Unit{})
403	
404	# --- client ---
405	
406	# what an ezhttp response to a request says
407	def client.took(res: EzClient.Response) -> Rpc.Reply:
408	  match res:
409	    case EzClient.Err{why}:
410	      Rpc.Unread{why}
411	    case EzClient.Response{status, +headers, reason, bytes}:
412	      Client.took(status, Http.headers.find(headers, "Content-Type"),
413	        Ez.body_text(bytes))
414	
415	# POST one request and read its reply
416	def client.ask(
417	  url: String,
418	  version: String,
419	  timeout_ms: U32,
420	  msg: Value.Json
421	) -> IO(Rpc.Reply):
422	  IO.bind(EzClient.Response, Rpc.Reply,
423	    Ez.http.post_opts(url, Client.headers(version), Ez.text_body(Json.print(msg)),
424	      timeout_ms),
425	    r => IO.pure(Rpc.Reply, client.took(r)))
426	
427	def client.sent(res: EzClient.Response) -> Bool:
428	  match res:
429	    case EzClient.Err{why}:
430	      False{}
431	    case EzClient.Response{status, headers, reason, bytes}:
432	      Client.accepted(status)
433	
434	# POST one notification; whether the server accepted it
435	def client.tell(
436	  url: String,
437	  version: String,
438	  timeout_ms: U32,
439	  msg: Value.Json
440	) -> IO(Bool):
441	  IO.bind(EzClient.Response, Bool,
442	    Ez.http.post_opts(url, Client.headers(version), Ez.text_body(Json.print(msg)),
443	      timeout_ms),
444	    r => IO.pure(Bool, client.sent(r)))