Add BrokerPort: loopd asks brokerd for every tool call

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/loopd/src/baseline.rs

@@ -25,6 +25,7 @@ pub enum BaselineError {
     Read(PathBuf, std::io::Error),
     Parse(PathBuf, serde_json::Error),
     Hash,
+    Core(PathBuf, std::io::Error),
 }
 
 impl std::fmt::Display for BaselineError {
@@ -33,6 +34,11 @@ impl std::fmt::Display for BaselineError {
             BaselineError::Read(path, err) => write!(f, "{}: {err}", path.display()),
             BaselineError::Parse(path, err) => write!(f, "{}: {err}", path.display()),
             BaselineError::Hash => write!(f, "hashing the baseline failed"),
+            BaselineError::Core(path, err) => write!(
+                f,
+                "{}: {err}; see docs/runbook.md#core-memory-unreadable",
+                path.display()
+            ),
         }
     }
 }
@@ -58,7 +64,7 @@ impl Baseline {
             // A missing file is fine; an unreadable one is reported so the owner is not given a
             // session without the memory they curated and no sign of it.
             Err(e) if e.kind() != std::io::ErrorKind::NotFound => {
-                return Err(BaselineError::Read(core.clone(), e));
+                return Err(BaselineError::Core(core.clone(), e));
             }
             Err(_) => {}
         }

crates/loopd/src/broker_port.rs

@@ -0,0 +1,220 @@
+//! The tool port to `brokerd`: one connection per call, a deadline per frame, and plain failures.
+//!
+//! Whatever goes wrong reaching `brokerd` or reading its frames, the model gets a plain failure and
+//! the turn goes on. Every failure prints one line ending at `#broker-unavailable`.
+
+use std::os::unix::net::UnixStream;
+use std::path::PathBuf;
+use std::time::{Duration, Instant};
+
+use crate::tools::{Pending, ToolPort};
+use proto::{
+    Envelope, FrameError, Message, PROTOCOL_VERSION, Timestamp, ToolRequest, ToolResponse,
+    read_frame, write_frame,
+};
+
+/// The failure every outage reaches the model as.
+pub const UNAVAILABLE: &str = "the tool broker is unavailable";
+/// What the model is told when no broker is configured.
+pub const NOT_CONFIGURED: &str = "no tool broker is configured";
+/// The failure a request too big for one frame reaches.
+pub const TOO_LARGE: &str = "the request is too large for the tool broker";
+/// The runbook entry every outage line ends with.
+pub const POINTER: &str = "see docs/runbook.md#broker-unavailable";
+
+/// "loopd: {UNAVAILABLE}: {why}; {POINTER}"
+pub fn unavailable_line(why: &str) -> String {
+    format!("loopd: {UNAVAILABLE}: {why}; {POINTER}")
+}
+
+/// "loopd: {NOT_CONFIGURED}: every tool call except the core ones will fail; {POINTER}"
+pub fn not_configured_line() -> String {
+    format!("loopd: {NOT_CONFIGURED}: every tool call except the core ones will fail; {POINTER}")
+}
+
+/// Reads through `stream`, but only what is left before `until`: a read past the deadline fails
+/// with a timeout, so a peer that sends one byte at a time cannot hold the turn.
+struct Deadline<'a> {
+    stream: &'a UnixStream,
+    until: Instant,
+}
+
+impl std::io::Read for Deadline<'_> {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        let left = self.until.saturating_duration_since(Instant::now());
+        if left.is_zero() {
+            return Err(std::io::Error::from(std::io::ErrorKind::TimedOut));
+        }
+        self.stream.set_read_timeout(Some(left))?;
+        let mut stream = self.stream; // `Read` is implemented for `&UnixStream`
+        stream.read(buf)
+    }
+}
+
+/// The tool port to `brokerd`. One connection per call, with a deadline per frame.
+pub struct BrokerPort {
+    socket: PathBuf,
+    timeout: Duration,
+    log: Box<dyn Fn(&str) + Send + Sync>,
+}
+
+impl BrokerPort {
+    pub fn new(socket: PathBuf, timeout: Duration) -> BrokerPort {
+        BrokerPort {
+            socket,
+            timeout,
+            log: Box::new(|l| eprintln!("{l}")),
+        }
+    }
+
+    pub fn with_log(
+        socket: PathBuf,
+        timeout: Duration,
+        log: Box<dyn Fn(&str) + Send + Sync>,
+    ) -> BrokerPort {
+        BrokerPort {
+            socket,
+            timeout,
+            log,
+        }
+    }
+}
+
+impl ToolPort for BrokerPort {
+    fn call(&self, request: &ToolRequest, on_pending: &mut dyn FnMut(&Pending)) -> ToolResponse {
+        let want = request.call.0;
+
+        // The first deadline, taken before connecting.
+        let mut until = match Instant::now().checked_add(self.timeout) {
+            Some(until) => until,
+            None => return self.unavailable("the timeout is too large"),
+        };
+
+        // Exit 2.
+        let mut stream = match UnixStream::connect(&self.socket) {
+            Ok(stream) => stream,
+            Err(e) => {
+                return self
+                    .unavailable(&format!("cannot connect to {}: {e}", self.socket.display()));
+            }
+        };
+
+        // Exit 3. `set_write_timeout` fails for a zero timeout.
+        if let Err(e) = stream.set_write_timeout(Some(self.timeout)) {
+            return self.unavailable(&format!("cannot set a timeout: {e}"));
+        }
+
+        // Exit 4. One request frame; the broker is fine if it is too large to send.
+        let request_frame = Envelope {
+            v: PROTOCOL_VERSION,
+            id: want,
+            r#final: true,
+            msg: Message::ToolRequest(request.clone()),
+        };
+        if let Err(e) = write_frame(&mut stream, &request_frame) {
+            return match e {
+                FrameError::TooLarge(_) => ToolResponse::Failed {
+                    message: TOO_LARGE.to_string(),
+                },
+                _ => self.unavailable(&format!("cannot send the request: {e}")),
+            };
+        }
+
+        // Exit 5-11, once per frame. A pending frame is reported, then we read again.
+        let mut seen_pending = false;
+        loop {
+            let mut deadline = Deadline {
+                stream: &stream,
+                until,
+            };
+            let env = match read_frame(&mut deadline) {
+                Ok(env) => env,
+                // Exit 5.
+                Err(FrameError::Closed) => {
+                    return self.unavailable("the connection closed before the final answer");
+                }
+                Err(FrameError::Io(e))
+                    if e.kind() == std::io::ErrorKind::WouldBlock
+                        || e.kind() == std::io::ErrorKind::TimedOut =>
+                {
+                    return self.unavailable("no answer in time");
+                }
+                Err(e) => return self.unavailable(&format!("bad frame: {e}")),
+            };
+
+            // Exit 6.
+            if env.id != want {
+                return self
+                    .unavailable(&format!("an answer for request {}, not {}", env.id, want));
+            }
+
+            let r#final = env.r#final;
+            // Exit 7.
+            let resp = match env.msg {
+                Message::ToolResponse(resp) => resp,
+                Message::Error(e) => {
+                    return self
+                        .unavailable(&format!("the broker reported an error: {}", e.detail));
+                }
+                _ => return self.unavailable("an unexpected message"),
+            };
+
+            match resp {
+                // Exit 8.
+                ToolResponse::PendingApproval { approval, expires } => {
+                    if r#final {
+                        return self.unavailable("a pending frame marked final");
+                    }
+                    // Exit 10.
+                    if seen_pending {
+                        return self.unavailable("a second pending frame");
+                    }
+                    let wait = expires
+                        .unix_millis()
+                        .saturating_sub(Timestamp::now().unix_millis());
+                    until = match Instant::now().checked_add(Duration::from_millis(wait)) {
+                        Some(until) => until,
+                        None => return self.unavailable("an expiry too far away"),
+                    };
+                    until = match until.checked_add(self.timeout) {
+                        Some(until) => until,
+                        None => return self.unavailable("an expiry too far away"),
+                    };
+                    seen_pending = true;
+                    on_pending(&Pending { approval, expires });
+                }
+                // Exit 9 and 11.
+                ToolResponse::Result { .. }
+                | ToolResponse::Failed { .. }
+                | ToolResponse::Denied { .. } => {
+                    if r#final {
+                        return resp;
+                    }
+                    return self.unavailable("an answer not marked final");
+                }
+            }
+        }
+    }
+}
+
+/// The one failure no outage reaches the model as, and the one line printed with it.
+impl BrokerPort {
+    fn unavailable(&self, why: &str) -> ToolResponse {
+        (self.log)(&unavailable_line(why));
+        ToolResponse::Failed {
+            message: UNAVAILABLE.to_string(),
+        }
+    }
+}
+
+/// The port used when no broker is configured: every call fails with `NOT_CONFIGURED` and prints
+/// nothing.
+pub struct NoBroker;
+
+impl ToolPort for NoBroker {
+    fn call(&self, _request: &ToolRequest, _on_pending: &mut dyn FnMut(&Pending)) -> ToolResponse {
+        ToolResponse::Failed {
+            message: NOT_CONFIGURED.to_string(),
+        }
+    }
+}

crates/loopd/src/config.rs

@@ -136,6 +136,23 @@ impl Default for Baseline {
     }
 }
 
+/// The tool broker (`brokerd`). No socket means no broker: every non-core tool call fails.
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct Broker {
+    pub socket: Option<std::path::PathBuf>,
+    pub timeout_ms: u64,
+}
+
+impl Default for Broker {
+    fn default() -> Self {
+        Self {
+            socket: None,
+            timeout_ms: 120_000,
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct Config {
@@ -154,6 +171,8 @@ pub struct Config {
     pub r#loop: Loop,
     #[serde(default)]
     pub baseline: Baseline,
+    #[serde(default)]
+    pub broker: Broker,
 }
 
 #[derive(Debug)]

crates/loopd/src/lib.rs

@@ -1,6 +1,7 @@
 //! The agent loop: sessions, prompt assembly and memory. It holds no authority.
 
 pub mod baseline;
+pub mod broker_port;
 pub mod channel;
 pub mod config;
 pub mod http;

crates/loopd/src/main.rs

@@ -6,12 +6,14 @@ use std::os::unix::net::UnixListener;
 use std::path::Path;
 use std::process::ExitCode;
 use std::sync::Arc;
+use std::time::Duration;
 
+use loopd::broker_port::{BrokerPort, NoBroker, not_configured_line};
 use loopd::channel::{self, Context};
 use loopd::config::Config;
 use loopd::llama::Client;
 use loopd::selftest::{SelfTestError, run};
-use loopd::tools::{FakeTools, Registry};
+use loopd::tools::{Registry, ToolPort};
 
 fn main() -> ExitCode {
     let args: Vec<String> = std::env::args().skip(1).collect();
@@ -38,7 +40,7 @@ fn run_selftest_check(client: &Client) -> Result<(), SelfTestError> {
             Ok(())
         }
         Err(e) => {
-            eprintln!("selftest: FAILED: {e}");
+            eprintln!("selftest: FAILED: {e}; see docs/runbook.md#loopd-selftest-failed");
             Err(e)
         }
     }
@@ -104,12 +106,17 @@ fn run_serve(path: &str) -> ExitCode {
     }
     eprintln!("loopd: serving on {}", socket.display());
 
-    let ctx = Arc::new(Context::new(
-        cfg,
-        client,
-        Box::new(FakeTools::new()),
-        Registry::m3a(),
-    ));
+    let port: Box<dyn ToolPort> = match &cfg.broker.socket {
+        Some(socket) => Box::new(BrokerPort::new(
+            socket.clone(),
+            Duration::from_millis(cfg.broker.timeout_ms),
+        )),
+        None => {
+            eprintln!("{}", not_configured_line());
+            Box::new(NoBroker)
+        }
+    };
+    let ctx = Arc::new(Context::new(cfg, client, port, Registry::m3a()));
     if let Err(e) = channel::serve(listener, ctx) {
         eprintln!("loopd: the channel server stopped: {e}");
         return ExitCode::from(1);

crates/loopd/src/session.rs

@@ -29,7 +29,11 @@ impl std::fmt::Display for SessionError {
             SessionError::Exists(id) => write!(f, "session already exists: {}", id.as_str()),
             SessionError::NotFound(id) => write!(f, "session not found: {}", id.as_str()),
             SessionError::Io(path, err) => write!(f, "{}: {err}", path.display()),
-            SessionError::Torn { path, line, why } => write!(f, "{}:{line}: {why}", path.display()),
+            SessionError::Torn { path, line, why } => write!(
+                f,
+                "{}:{line}: {why}; see docs/runbook.md#session-log-damaged",
+                path.display()
+            ),
             SessionError::Baseline(err) => write!(f, "{err}"),
             SessionError::Encode(err) => write!(f, "{err}"),
         }