B-207: implement BEP-069 Read and Write interfaces

baml_language/crates/baml_builtins2/baml_std/ai/ns_mcp/mcp.baml

@@ -24,6 +24,8 @@ function rpc_line(id: int?, method: string, params: map<string, unknown>?) -> st
 class McpConnection {
     server_name: string,
     p: baml.sys.Process,
+    out: baml.sys.ReadPipeLines,
+    timeout_ms: int,
     next_id: int,
 
     // Spawn the server and perform the initialize handshake. The connection
@@ -37,13 +39,7 @@ class McpConnection {
         let p = baml.sys.start_process(
             command,
             args,
-            baml.sys.ProcessOptions {
-                cwd: null,
-                env: null,
-                timeout_ms: timeout_ms,
-                stdin: null,
-                keep_stdin_open: true,
-            },
+            baml.sys.ProcessOptions { cwd: null, env: null, timeout_ms: timeout_ms, stdin: null },
         ) catch_all (e) {
             _ => {
                 throw root.errors.NetworkFailure { status_code: null,
@@ -53,7 +49,13 @@ class McpConnection {
                 }
             },
         };
-        let conn = McpConnection { server_name: name, p: p, next_id: 1 };
+        let conn = McpConnection {
+            server_name: name,
+            p: p,
+            out: p.stdout.lines(),
+            timeout_ms: timeout_ms,
+            next_id: 1,
+        };
         let init: map<string, unknown> = {
             "protocolVersion": "2025-06-18",
             "capabilities": {},
@@ -69,7 +71,7 @@ class McpConnection {
         // lose the newline the protocol frames on; the newline is a quoted
         // string, appended before the single write.
         let framed = [line, "\n"].join("");
-        let _ = self.p.write_stdin(framed) catch_all (e) {
+        let _ = self.p.stdin.write(framed) catch_all (e) {
             _ => {
                 throw root.errors.NetworkFailure { status_code: null,
                     provider: `mcp/${self.server_name}`,
@@ -86,52 +88,62 @@ class McpConnection {
         let id = self.next_id;
         self.next_id += 1;
         self._send(rpc_line(id, method, params));
-        while (true) {
-            let next = self.p.stdout.next() catch_all (e) {
-                _ => {
-                    throw root.errors.NetworkFailure { status_code: null,
-                        provider: `mcp/${self.server_name}`,
-                        detail: `reading from the MCP server failed: ${e.to_string()}`,
-                        raw_body: null,
-                    }
-                },
-            };
-            match (next) {
-                let line: string => {
-                    let trimmed = line.trim();
-                    if (trimmed.length() > 0) {
-                        // stdout is protocol-only per the spec; tolerate stray lines
-                        let msg = baml.json.parse(trimmed) catch_all (e) {
-                            _ => null,
-                        };
-                        if (baml.json.path_or<int>(msg, ".id", -1) == id) {
-                            let err = baml.json.path_or<json>(msg, ".error", null);
-                            if (err != null) {
-                                throw root.errors.InvalidRequest {
-                                    provider: `mcp/${self.server_name}`,
-                                    status_code: baml.json.path_or<int?>(err, ".code", null),
-                                    detail: baml.json.path_or<string>(err, ".message", "MCP error"),
-                                    raw_body: baml.json.stringify(err),
+        baml.future.with_timeout(baml.time.Duration.from_milliseconds(self.timeout_ms), () -> {
+            while (true) {
+                let next = self.out.next() catch_all (e) {
+                    _ => {
+                        throw root.errors.NetworkFailure { status_code: null,
+                            provider: `mcp/${self.server_name}`,
+                            detail: `reading from the MCP server failed: ${e.to_string()}`,
+                            raw_body: null,
+                        }
+                    },
+                };
+                match (next) {
+                    let line: string => {
+                        let trimmed = line.trim();
+                        if (trimmed.length() > 0) {
+                            // stdout is protocol-only per the spec; tolerate stray lines
+                            let msg = baml.json.parse(trimmed) catch_all (e) {
+                                _ => null,
+                            };
+                            if (baml.json.path_or<int>(msg, ".id", -1) == id) {
+                                let err = baml.json.path_or<json>(msg, ".error", null);
+                                if (err != null) {
+                                    throw root.errors.InvalidRequest {
+                                        provider: `mcp/${self.server_name}`,
+                                        status_code: baml.json.path_or<int?>(err, ".code", null),
+                                        detail: baml.json.path_or<string>(err, ".message", "MCP error"),
+                                        raw_body: baml.json.stringify(err),
+                                    }
                                 }
+                                return baml.json.path_or<json>(msg, ".result", null);
                             }
-                            return baml.json.path_or<json>(msg, ".result", null);
                         }
-                    }
-                },
-                _ => {
-                    throw root.errors.NetworkFailure { status_code: null,
-                        provider: `mcp/${self.server_name}`,
-                        detail: `the MCP server stream ended before answering ${method}`,
-                        raw_body: null,
-                    }
-                },
-            };
-        }
-        // unreachable: the loop returns on a matching id or throws at EOF
-        throw root.errors.NetworkFailure { status_code: null,
-            provider: `mcp/${self.server_name}`,
-            detail: `no response for ${method}`,
-            raw_body: null,
+                    },
+                    _ => {
+                        throw root.errors.NetworkFailure { status_code: null,
+                            provider: `mcp/${self.server_name}`,
+                            detail: `the MCP server stream ended before answering ${method}`,
+                            raw_body: null,
+                        }
+                    },
+                };
+            }
+            // unreachable: the loop returns on a matching id or throws at EOF
+            throw root.errors.NetworkFailure { status_code: null,
+                provider: `mcp/${self.server_name}`,
+                detail: `no response for ${method}`,
+                raw_body: null,
+            }
+        }) catch (e) {
+            baml.errors.Timeout => {
+                throw root.errors.NetworkFailure { status_code: null,
+                    provider: `mcp/${self.server_name}`,
+                    detail: `reading from the MCP server timed out after ${self.timeout_ms}ms`,
+                    raw_body: null,
+                }
+            },
         }
     }
 

baml_language/crates/baml_builtins2/baml_std/baml/ns_csv/csv.baml

@@ -7,7 +7,7 @@
 //
 // The streaming reader is the only parser; every one-shot is a wrapper over
 // it. File IO goes through `baml.fs` handles: the reader pumps chunks via
-// `File.read_bytes` from thin BAML method bodies, while all tokenization,
+// `baml.io.Read` from thin BAML method bodies, while all tokenization,
 // decoding, and encoding is native.
 
 /// Classifies a `CsvError` for portable handling.
@@ -405,29 +405,21 @@ class CsvReader {
     /// (internal) Pull the next chunk from the backing file, or mark EOF for
     /// in-memory sources. IO errors are stream-scoped and fatal: the reader is
     /// marked exhausted before the throw, so subsequent `next()` calls return
-    /// `baml.iter.Done`.
-    function _read_chunk(self) -> null throws CsvError | root.errors.Io {
-        match (self._file) {
+    /// `baml.iter.Done`. A closed handle arrives as an IO error because
+    /// `baml.io.Read` has no separate closed-handle error channel.
+    function _read_chunk(self) -> void throws root.errors.Io {
+        let chunk = self._file?.read(65536) catch (e) {
+            root.errors.Io => {
+                self._mark_exhausted();
+                throw e
+            },
+        };
+        match (chunk) {
             null => {
                 self._feed_eof();
-                null
             },
-            let f: root.fs.File => {
-                let chunk = f.read_bytes(65536) catch (e) {
-                    root.errors.InvalidArgument { message } => {
-                        throw _error(CsvErrorKind.Closed, "underlying file handle is closed: " + message)
-                    },
-                    root.errors.Io => {
-                        self._mark_exhausted();
-                        throw e
-                    },
-                };
-                if (chunk.length() == 0) {
-                    self._feed_eof();
-                } else {
-                    self._feed(chunk);
-                }
-                null
+            let bytes: uint8array => {
+                self._feed(bytes);
             },
         }
     }
@@ -583,11 +575,7 @@ class CsvWriter {
             null => null,
             let f: root.fs.File => {
                 if (out.length() > 0) {
-                    f.write(out) catch (_) {
-                        root.errors.InvalidArgument { message } => {
-                            throw root.errors.Io { message: "underlying file handle is closed: " + message }
-                        },
-                    };
+                    f.write(out);
                 }
                 null
             },

baml_language/crates/baml_builtins2/baml_std/baml/ns_fs/fs.baml

@@ -1,37 +1,25 @@
 /// A handle to an open file. Use `baml.fs.open` to obtain one.
 ///
-/// Every `File.*` method can throw `InvalidArgument` when the underlying
-/// handle has already been closed (e.g. by a prior `close()`); read-string
-/// methods can additionally throw `ParseError` on non-UTF-8 input.
+/// Read and write operations on a closed handle throw `Io`. `text()`
+/// additionally throws `ParseError` when the remaining bytes are not valid
+/// UTF-8.
 class File {
     _handle: $rust_type,
 
     /// Reads the entire remaining file contents as a UTF-8 string.
-    function text(
-        self,
-    ) -> string throws root.errors.Io | root.errors.InvalidArgument | root.errors.ParseError {
-        $rust_io_function
+    function text(self) -> string throws root.errors.Io | root.errors.ParseError {
+        self.as<root.io.Read>.text()
     }
 
     /// Reads the entire remaining file contents as raw bytes.
-    function bytes(self) -> uint8array throws root.errors.Io | root.errors.InvalidArgument {
-        $rust_io_function
-    }
-
-    /// Reads up to `n` bytes from the file and returns them as a UTF-8 string.
-    function read(
-        self,
-        n: int,
-    ) -> string throws root.errors.Io | root.errors.InvalidArgument | root.errors.ParseError {
-        $rust_io_function
+    function bytes(self) -> uint8array throws root.errors.Io {
+        self.as<root.io.Read>.bytes()
     }
 
-    /// Reads up to `n` bytes from the file and returns them as raw bytes.
-    function read_bytes(
-        self,
-        n: int,
-    ) -> uint8array throws root.errors.Io | root.errors.InvalidArgument {
-        $rust_io_function
+    implements root.io.Read {
+        function read(self, limit: int) -> uint8array? throws root.errors.Io {
+            $rust_io_function
+        }
     }
 
     /// Closes the file handle, flushing any pending writes.
@@ -51,17 +39,20 @@ class File {
         $rust_io_function
     }
 
-    /// Writes `data` to the file at the current cursor position. Returns the number of bytes written.
-    function write(self, data: string) -> int throws root.errors.Io | root.errors.InvalidArgument {
-        $rust_io_function
+    implements root.io.Write {
+        function write_some(self, data: uint8array) -> int throws root.errors.Io {
+            $rust_io_function
+        }
+
+        function flush(self) -> void throws root.errors.Io {
+            $rust_io_function
+        }
     }
 
-    /// Writes raw bytes `data` to the file at the current cursor position. Returns the number of bytes written.
-    function write_bytes(
-        self,
-        data: uint8array,
-    ) -> int throws root.errors.Io | root.errors.InvalidArgument {
-        $rust_io_function
+    /// Writes all of `data` to the file at the current cursor position.
+    /// Returns the encoded byte count.
+    function write(self, data: string | uint8array) -> int throws root.errors.Io {
+        self.as<root.io.Write>.write(data)
     }
 }
 

baml_language/crates/baml_builtins2/baml_std/baml/ns_future/future.baml

@@ -238,3 +238,34 @@ function any<T, Err>(
         winners[0]
     }
 }
+
+/// Run `body` under a deadline.
+function with_timeout<T, E>(
+    limit: baml.time.Duration,
+    body: () -> T throws E,
+) -> T throws E | baml.errors.Timeout | baml.panics.Cancelled {
+    let token = baml.spawn.CancelToken.new();
+    let work = spawn with baml.spawn.options(cancel = token) { body() };
+    let deadline = spawn {
+        baml.sys.sleep(limit);
+        token.cancel()
+    };
+    defer {
+        deadline.cancel()
+    }
+    (await work) catch (e) {
+        baml.panics.Cancelled => {
+            if (!token.is_cancelled()) {
+                throw e;
+            }
+            let limit_ms = limit.to_milliseconds();
+            let reported = limit_ms.to_int() catch (conversion) {
+                baml.errors.InvalidArgument => null,
+            };
+            throw baml.errors.Timeout {
+                message: `operation timed out after ${limit_ms}ms`,
+                duration_ms: reported,
+            }
+        },
+    }
+}

baml_language/crates/baml_builtins2/baml_std/baml/ns_io/read.baml

@@ -0,0 +1,32 @@
+interface Read {
+    /// Read up to `limit` bytes from the source.
+    /// Returns `null` after EOF. A non-null result is never empty.
+    function read(self, limit: int) -> uint8array? throws root.errors.Io
+
+    /// Consume all remaining bytes through EOF.
+    function bytes(self) -> uint8array throws root.errors.Io {
+        let result = b"";
+        while let chunk: uint8array = self.read(65536) {
+            result = result.concat(chunk);
+        }
+        result
+    }
+
+    /// Consume all remaining bytes through EOF and decode strict UTF-8.
+    function text(
+        self,
+    ) -> string throws root.errors.Io | root.errors.ParseError {
+        let data = self.bytes();
+        _decode_utf8(data)
+    }
+}
+
+function _decode_utf8(data: uint8array) -> string throws root.errors.ParseError {
+    string.from_utf8(data) catch (e) {
+        root.errors.InvalidArgument => {
+            throw root.errors.ParseError {
+            message: "invalid utf-8",
+        }
+        },
+    }
+}