From e167fc871b97514d92a9e3c82a7ffecd1ed7e6df Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Thu, 24 Sep 2026 18:12:32 -0700 Subject: [PATCH] brokerd: memory calls to memoryd with the session's state; labels only stricter Implemented-By: OpenCode session (model recorded in docs/implementer-log.md) --- crates/brokerd/src/config.rs | 18 +++ crates/brokerd/src/container.rs | 7 + crates/brokerd/src/lib.rs | 1 + crates/brokerd/src/main.rs | 7 + crates/brokerd/src/memory_port.rs | 119 +++++++++++++++++ crates/brokerd/src/policy/mod.rs | 7 + crates/brokerd/src/runner.rs | 39 ++++-- crates/brokerd/tests/memory_port.rs | 200 ++++++++++++++++++++++++++++ crates/brokerd/tests/support/rig.rs | 1 + docs/implementer-log.md | 1 + docs/runbook.md | 18 +++ 11 files changed, 410 insertions(+), 8 deletions(-) create mode 100644 crates/brokerd/src/memory_port.rs create mode 100644 crates/brokerd/tests/memory_port.rs diff --git a/crates/brokerd/src/config.rs b/crates/brokerd/src/config.rs index 04c6bc4..e3d15c9 100644 --- a/crates/brokerd/src/config.rs +++ b/crates/brokerd/src/config.rs @@ -53,6 +53,14 @@ pub struct Gateway { pub socket: PathBuf, } +/// `[memory]`: where `memoryd` takes the decided `recall` and `memory` calls. The memory tools are +/// always on, so the section is optional; an empty path means the default under `home`. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Default)] +#[serde(deny_unknown_fields, default)] +pub struct Memory { + pub socket: PathBuf, +} + fn default_podman() -> PathBuf { PathBuf::from("podman") } @@ -141,6 +149,8 @@ pub struct Config { pub runner: Option, #[serde(default)] pub gateway: Option, + #[serde(default)] + pub memory: Memory, } /// The longest an approval may wait: a day, the longest `loopd` waits after a pending frame. @@ -283,6 +293,14 @@ impl Config { Some(gateway.socket.clone()) } } + /// `memoryd`'s socket for `brokerd`. + pub fn memory_socket(&self) -> PathBuf { + if self.memory.socket.as_os_str().is_empty() { + self.paths.home.join("run/broker-memory/memory.sock") + } else { + self.memory.socket.clone() + } + } pub fn audit_dir(&self) -> PathBuf { self.paths.home.join("audit") } diff --git a/crates/brokerd/src/container.rs b/crates/brokerd/src/container.rs index d4f8f3b..1c7de6d 100644 --- a/crates/brokerd/src/container.rs +++ b/crates/brokerd/src/container.rs @@ -34,6 +34,9 @@ pub const EGRESS_POLL: Duration = Duration::from_millis(20); /// A log sink. Call it as `(self.log)("a line")`. pub type Log = Arc; +/// A memory tool that reached the container runtime: fixed, it reaches the model. +pub const IN_MEMORYD: &str = "the memory tools run in memoryd, not in a container"; + pub struct Podman { runner: Runner, egress_dir: PathBuf, @@ -240,6 +243,10 @@ impl Runtime for Podman { /// an egress starts the proxy first and gives the tool the directory. `run_container` is step 3 /// to 7 for either. Already written. fn run(&self, spec: &RunSpec) -> Result { + // `Routed` sends the memory tools to `memoryd`; one here is refused, never run. + if spec.tool().is_memory() { + return Err(RunError::Unavailable(IN_MEMORYD.to_string())); + } let n = self.next.fetch_add(1, Ordering::SeqCst); let name = podman::container_name(spec.session(), spec.call(), n); let input = spec.arguments().canonical_json(); diff --git a/crates/brokerd/src/lib.rs b/crates/brokerd/src/lib.rs index f2fd837..c708315 100644 --- a/crates/brokerd/src/lib.rs +++ b/crates/brokerd/src/lib.rs @@ -11,6 +11,7 @@ pub mod container; pub mod courier; pub mod grants; pub mod ledger; +pub mod memory_port; pub mod pipes; pub mod podman; pub mod policy; diff --git a/crates/brokerd/src/main.rs b/crates/brokerd/src/main.rs index 98df926..0dad1b8 100644 --- a/crates/brokerd/src/main.rs +++ b/crates/brokerd/src/main.rs @@ -8,6 +8,7 @@ use std::sync::Arc; use brokerd::audit::{AuditError, RECOVERED_NOTICE}; use brokerd::config::Config; use brokerd::container::Podman; +use brokerd::memory_port::{MEMORY_WAIT, MemoryPort, Routed}; use brokerd::runner::{Refusing, Runtime}; use brokerd::serve::{self, ServeError}; @@ -66,6 +67,12 @@ fn main() -> ExitCode { } } + // The memory tools go to `memoryd`, whatever runs the others (M5a spec, section 8). + let runtime: Box = Box::new(Routed { + memory: MemoryPort::new(cfg.memory_socket(), MEMORY_WAIT), + other: runtime, + }); + let started = match serve::start(cfg, accept_break, runtime, log) { Ok(started) => started, Err(ServeError::Audit(e @ AuditError::NothingToAccept)) => { diff --git a/crates/brokerd/src/memory_port.rs b/crates/brokerd/src/memory_port.rs new file mode 100644 index 0000000..efd1277 --- /dev/null +++ b/crates/brokerd/src/memory_port.rs @@ -0,0 +1,119 @@ +//! Where decided `recall` and `memory` calls go: to `memoryd` over its `brokerd` socket, with the +//! session's state, instead of a container (M5a spec, section 8). `Routed` sends every other call +//! to the configured runtime. + +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::time::Duration; + +use proto::{ + Envelope, MemoryCall, Message, PROTOCOL_VERSION, ToolRequest, ToolResponse, read_frame, + write_frame, +}; + +use crate::policy::Label; +use crate::runner::{RunError, RunOutput, RunSpec, Runtime}; + +/// The sentence when `memoryd` cannot be reached or answers wrongly: fixed, it reaches the model. +pub const UNREACHABLE: &str = + "memory is unavailable: memoryd did not answer (see docs/runbook.md#memoryd-unreachable)"; + +/// How long `brokerd` waits for `memoryd` to answer one call. +pub const MEMORY_WAIT: Duration = Duration::from_secs(30); + +/// A client of `memoryd`'s `brokerd` socket. One connection per call. +pub struct MemoryPort { + socket: PathBuf, + wait: Duration, +} + +impl MemoryPort { + pub fn new(socket: PathBuf, wait: Duration) -> MemoryPort { + MemoryPort { socket, wait } + } + + fn ask(&self, spec: &RunSpec) -> Option { + // `None` at the first thing that fails. 1. A MemoryCall: request with the spec's session, + // call, `spec.tool().as_str()`, and `spec.arguments().canonical_json()`; taint and + // untrusted from `spec.state()`. 2. `UnixStream::connect(&self.socket)`, then read and + // write timeouts of `self.wait`. 3. `write_frame` an Envelope { v: PROTOCOL_VERSION, id: + // spec.call().0, final: true, msg: Message::MemoryCall(call) }. 4. `read_frame`; a + // Message::ToolResponse is Some(it), any other message None. + let call = MemoryCall { + request: ToolRequest { + session: spec.session().clone(), + call: spec.call(), + tool: spec.tool().as_str().to_string(), + arguments: spec.arguments().canonical_json(), + }, + taint: spec.state().taint, + untrusted: spec.state().untrusted, + }; + let mut stream = UnixStream::connect(&self.socket).ok()?; + stream.set_read_timeout(Some(self.wait)).ok()?; + stream.set_write_timeout(Some(self.wait)).ok()?; + write_frame( + &mut stream, + &Envelope { + v: PROTOCOL_VERSION, + id: spec.call().0, + r#final: true, + msg: Message::MemoryCall(call), + }, + ) + .ok()?; + let env = read_frame(&mut stream).ok()?; + match env.msg { + Message::ToolResponse(response) => Some(response), + _ => None, + } + } +} + +impl Runtime for MemoryPort { + fn run(&self, spec: &RunSpec) -> Result { + self.run_labelled(spec).map(|(output, _)| output) + } + + fn run_labelled(&self, spec: &RunSpec) -> Result<(RunOutput, Option