Read & Write interfaces

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

@@ -24,6 +24,10 @@ function rpc_line(id: int?, method: string, params: map<string, unknown>?) -> st
 class McpConnection {
     server_name: string,
     p: baml.sys.Process,
+    // Framed once at connect: the stream buffers whatever it read past the
+    // last newline, so re-framing per `_request` would drop the start of the
+    // next message.
+    out: baml.sys.ReadPipeLines,
     next_id: int,
 
     // Spawn the server and perform the initialize handshake. The connection
@@ -42,7 +46,6 @@ class McpConnection {
                 env: null,
                 timeout_ms: timeout_ms,
                 stdin: null,
-                keep_stdin_open: true,
             },
         ) catch_all (e) {
             _ => {
@@ -53,7 +56,10 @@ class McpConnection {
                 }
             },
         };
-        let conn = McpConnection { server_name: name, p: p, next_id: 1 };
+        // `lines()` only wraps the pipe — reading it is what can fail, and that
+        // surfaces at `_request`.
+        let out = p.stdout.lines();
+        let conn = McpConnection { server_name: name, p: p, out: out, next_id: 1 };
         let init: map<string, unknown> = {
             "protocolVersion": "2025-06-18",
             "capabilities": {},
@@ -69,7 +75,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) {
+        self.p.stdin.write_all(framed.to_utf8()) catch_all (e) {
             _ => {
                 throw root.errors.NetworkFailure {
                     provider: `mcp/${self.server_name}`,
@@ -87,7 +93,7 @@ class McpConnection {
         self.next_id += 1;
         self._send(rpc_line(id, method, params));
         while (true) {
-            let next = self.p.stdout.next() catch_all (e) {
+            let next = self.out.next() catch_all (e) {
                 _ => {
                     throw root.errors.NetworkFailure {
                         provider: `mcp/${self.server_name}`,

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 one of those IO errors —
+    /// `baml.io.Read` has no separate channel for it.
+    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.to_utf8());
                 }
                 null
             },

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

@@ -9,29 +9,19 @@ class File {
     /// 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
+    ) -> string throws root.errors.Io | root.errors.ParseError {
+        self.as<root.io.Read>.read_to_string()
     }
 
     /// 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>.read_to_end()
     }
 
-    /// 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, n: int) -> uint8array? throws root.errors.Io {
+            $rust_io_function
+        }
     }
 
     /// Closes the file handle, flushing any pending writes.
@@ -51,17 +41,23 @@ 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(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 `data` to the file at the current cursor position. Returns the number of bytes written.
+    function write(self, data: string | uint8array) -> int throws root.errors.Io {
+        let bytes = match (data) {
+            string => data.to_utf8(),
+            uint8array => data,
+        };
+        self.as<root.io.Write>.write(bytes)
     }
 }
 

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

@@ -238,3 +238,44 @@ function any<T, Err>(
         winners[0]
     }
 }
+
+/// Runs `body` under a deadline, throwing `baml.errors.Timeout` if `limit`
+/// elapses first.
+///
+/// The work runs under a cancel token that a sleeping task fires, so a body
+/// parked in a sys-op (a socket read, a child's stdout) is unparked rather
+/// than left running — this is the general form of a deadline in BAML, since
+/// `read` and friends take no timeout of their own.
+function with_timeout<T, E>(
+    limit: baml.time.Duration,
+    body: () -> T throws E,
+) -> T throws E | baml.errors.Timeout {
+    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()
+    };
+    let out = (await work) catch (e) {
+        // The token fires only from the timer above, so a cancelled body is
+        // an expired deadline — anything the body itself threw is `E` and
+        // propagates untouched.
+        baml.panics.Cancelled => {
+            let limit_ms = limit.to_milliseconds();
+            // `duration_ms` is an `int`, so a limit too large to fit is
+            // reported by message alone rather than failing the throw itself.
+            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,
+            }
+        }
+    };
+    // The body won; stop the timer rather than leaving it to fire later.
+    deadline.cancel();
+    out
+}

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

@@ -0,0 +1,31 @@
+interface Read {
+    /// Read some bytes from the source.
+    /// When the end of the source has already been reached, returns `null`.
+    ///
+    /// The `limit` parameter specifies the maximum number of bytes to read.
+    ///
+    /// This is the only method that implementors are required to provide.
+    function read(self, limit: int) -> uint8array? throws root.errors.Io
+
+    /// Reads all remaining bytes from the source until EOF is reached.
+    ///
+    /// Drains in fixed-size chunks rather than asking for everything at once:
+    /// an implementor is free to size a buffer from `limit`, so a limit of
+    /// `int.max_value()` is a request to allocate more memory than exists.
+    function read_to_end(self) -> uint8array throws root.errors.Io {
+        let buf = b"";
+        while let chunk: uint8array = self.read(65536) {
+            buf = buf.concat(chunk);
+        }
+        buf
+    }
+
+    /// Reads all remaining bytes as a UTF-8 string until EOF is reached.
+    /// If the source is not valid UTF-8, an error is thrown.
+    function read_to_string(self) -> string throws root.errors.Io | root.errors.ParseError {
+        let buf = self.read_to_end();
+        string.from_utf8(buf) catch (_) {
+            _ => throw root.errors.ParseError { message: "invalid utf-8" },
+        }
+    }
+}

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

@@ -0,0 +1,20 @@
+interface Write {
+    /// Write some bytes to the destination.
+    ///
+    /// Returns the number of bytes successfully written.
+    function write(self, data: uint8array) -> int throws root.errors.Io
+
+    /// Flush any buffered data to the destination.
+    function flush(self) -> void throws root.errors.Io
+
+    /// Keep writing data until all bytes have been written.
+    function write_all(self, data: uint8array) -> void throws root.errors.Io {
+        while (data.length() > 0) {
+            let written = self.write(data);
+            if (written == 0) {
+                throw root.errors.Io { message: "stream closed without writing all data" };
+            }
+            data = data.slice(written, data.length());
+        }
+    }
+}

baml_language/crates/baml_builtins2/baml_std/baml/ns_net/net.baml

@@ -35,45 +35,26 @@ class TcpStream {
         $rust_io_function
     }
 
-    /// Reads the next available bytes from the stream. Blocks until data arrives.
-    /// Returns an empty `uint8array` once the peer has closed the connection (EOF).
-    ///
-    /// `timeout` bounds the wait for data; `null` (the default) blocks
-    /// indefinitely. On expiry this throws `root.errors.Timeout` and the stream
-    /// remains usable for a subsequent read.
-    function read(
-        self,
-        timeout: root.time.Duration? = null,
-    ) -> uint8array throws root.errors.Io | root.errors.Timeout {
-        self._read(_timeout_nanos(timeout))
-    }
-
-    function _read(
-        self,
-        timeout_nanos: bigint,
-    ) -> uint8array throws root.errors.Io | root.errors.Timeout {
-        $rust_io_function
-    }
-
-    /// Writes `data` to the stream. Blocks until every byte has been flushed.
-    ///
-    /// `timeout` bounds the flush; `null` (the default) blocks indefinitely. On
-    /// expiry this throws `root.errors.Timeout`; an unknown amount of `data` may
-    /// already have been written.
-    function write(
-        self,
-        data: uint8array,
-        timeout: root.time.Duration? = null,
-    ) -> null throws root.errors.Io | root.errors.Timeout {
-        self._write(data, _timeout_nanos(timeout))
-    }
-
-    function _write(
-        self,
-        data: uint8array,
-        timeout_nanos: bigint,
-    ) -> null throws root.errors.Io | root.errors.Timeout {
-        $rust_io_function
+    implements root.io.Read {
+        /// Reads the next available bytes from the stream. Blocks until data arrives.
+        /// Returns `null` once the stream has been closed and no more data is available (EOF).
+        function read(self, n: int) -> uint8array? throws root.errors.Io {
+            $rust_io_function
+        }
+    }
+
+    implements root.io.Write {
+        /// Writes `data` to the stream.
+        ///
+        /// If cancelled (e.g. by a timeout), an unknown amount of `data` may
+        /// already have been written.
+        function write(self, data: uint8array) -> int throws root.errors.Io {
+            $rust_io_function
+        }
+
+        /// No-op, TCP streams don't need to be flushed.
+        function flush(self) -> void throws never {
+        }
     }
 
     /// Closes the connection.