boxmaker/docs/plans/M6b/06-brokerd-consult-http.md

M6b task 06: allowlisted POST; audit body equals the bytes

Branch: m6b. Check it: git branch --show-current prints m6b and git status --short is empty; otherwise stop. Never switch branch. Commit subject: brokerd: consult posts the allowlisted body Log row: the Task cell is M6b/06-brokerd-consult-http.

First action. Run this before any Read of crate source. Do not ls crates/brokerd/tests/. Do not Read podman.rs, docs/implementer-log.md (except the first 12 lines at the log step). The given test injects an in-process fake; do not open a socket in the tests and do not reach the network (T27).

cp docs/plans/M6b/files/crates/brokerd/tests/consult_allow.rs crates/brokerd/tests/

Goal

An allowed consult POSTs the allowlisted JSON (wire_body) through ConsultTransport. Tests inject a fake; production uses loopback HTTP/1.1 (HttpTransport). The audit body equals the posted bytes. messages has one user item and no tools. The tool result is choices[0].message.content, tag-neutralised, then capped. ConsultStack sends Consult to that runtime and everything else to inner, so existing Routed tests stay unedited. main loads the secret after Config::load and wraps Routed when [consult] and the secret are both present.

Every place that changes

Place	What happens there
crates/brokerd/src/consult.rs	ConsultTransport, ConsultRuntime, ConsultStack, parse_message_content
crates/brokerd/src/consult_http.rs	new: HttpTransport
crates/brokerd/src/lib.rs	pub mod consult_http;
crates/brokerd/src/main.rs	load_secret; wrap Routed in ConsultStack
crates/brokerd/tests/consult_allow.rs	new, given
everything else	no change: not ledger, not handle's decide path

Files

• Copy, from docs/plans/M6b/files/ (the same path under it): the new test crates/brokerd/tests/consult_allow.rs.
• Create: crates/brokerd/src/consult_http.rs (the module below).
• Modify: crates/brokerd/src/consult.rs, crates/brokerd/src/lib.rs, crates/brokerd/src/main.rs, docs/implementer-log.md.
• Do not touch: any other file.

The given test holds mod runtime even though it does not name Recording: support/client.rs imports crate::runtime. Leave that module in the test.

Facts you need

• Runtime::run_labelled default calls run. Implement both: run forwards to run_labelled and drops the label. Implementing only run_labelled recurses.
• No choices[0] index. as_array().and_then(|a| a.first()). A missing content is "the response has no content".
• Neutralise then cap. bxcore::sanitize::neutralise then cap_bytes(..., QUESTION_CAP). The given test builds the think tag with concat!("<", "think", ">") and asserts < in the result. Do not write that tag whole anywhere (T33).
• Label. run_labelled returns (RunOutput { content, truncated }, None). Result class comes from the grant, as today.
• HTTP/1.1. TcpStream to the loopback host:port from parse_loopback_http. Headers: POST {path} HTTP/1.1, Host, Authorization: Bearer {bearer}, Content-Type: application/json, Content-Length, Connection: close. Response cap 262_144 bytes. Status must contain 200 or end with 200. HTTPS is later.
• load_secret in main after Config::load, before the instance lock. Print {e} and exit 1 on Err. Print a warning if Loaded has one. Role is already "brokerd" inside load_secret.
• Wrap after Routed. ConsultStack { consult: ConsultRuntime::new(consult, loaded.secret, HttpTransport), inner: runtime } when both cfg.consult and the loaded secret are Some. Otherwise keep runtime.
• The items the edits use. This table is complete.

  Item	Signature
  Runtime::run	run(&self, spec: &RunSpec) -> Result<RunOutput, RunError>
  Runtime::run_labelled	run_labelled(&self, spec: &RunSpec) -> Result<(RunOutput, Option<Label>), RunError>
  RunSpec::tool	tool(&self) -> ToolName
  RunSpec::arguments	arguments(&self) -> &ToolArgs
  RunError::Unavailable	Unavailable(String)
  RunOutput	{ content: String, truncated: bool }
  Secret::new	new(text: String) -> Secret
  Secret::expose	expose(&self) -> &str
  Loaded	{ secret: Secret, warning: Option<String> }
  bxcore::sanitize::neutralise	neutralise(s: &str) -> String
  Consult::temperature	temperature(&self) -> f64
  Consult::timeout	timeout(&self) -> Duration
  parse_loopback_http	parse_loopback_http(url: &str) -> Result<(String, u16, String), String>
  TcpStream::connect	connect<A: ToSocketAddrs>(addr: A) -> io::Result<TcpStream>
  TcpStream::set_read_timeout	set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()>
  TcpStream::set_write_timeout	set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()>

The edits, exactly

Each quoted "before" block occurs exactly once after task 05.

crates/brokerd/src/consult.rs

I1. The imports

use bxcore::secrets::{Loaded, SecretError, SecretSpec};
use serde::{Deserialize, Deserializer, Serialize};

use crate::config::Config;

become

use bxcore::secrets::{Loaded, Secret, SecretError, SecretSpec};
use serde::{Deserialize, Deserializer, Serialize};

use crate::args::ToolArgs;
use crate::config::Config;
use crate::policy::Label;
use crate::runner::{RunError, RunOutput, RunSpec, Runtime};

I2. After load_secret (after its closing }) append:

/// One POST of the allowlisted body. Tests inject a fake; production uses loopback HTTP/1.1.
pub trait ConsultTransport: Send + Sync {
    fn post(
        &self,
        url: &str,
        bearer: &str,
        body: &[u8],
        timeout: Duration,
    ) -> Result<Vec<u8>, String>;
}

/// Runs an allowed `consult` through `transport`.
pub struct ConsultRuntime<T> {
    consult: Consult,
    secret: Secret,
    transport: T,
}

impl<T: ConsultTransport> ConsultRuntime<T> {
    pub fn new(consult: Consult, secret: Secret, transport: T) -> ConsultRuntime<T> {
        ConsultRuntime {
            consult,
            secret,
            transport,
        }
    }
}

impl<T: ConsultTransport> Runtime for ConsultRuntime<T> {
    fn run(&self, spec: &RunSpec) -> Result<RunOutput, RunError> {
        self.run_labelled(spec).map(|(output, _)| output)
    }

    fn run_labelled(&self, spec: &RunSpec) -> Result<(RunOutput, Option<Label>), RunError> {
        let question = match spec.arguments() {
            ToolArgs::Consult { question } => question.as_str(),
            _ => {
                return Err(RunError::Unavailable(
                    "consult was routed a different tool".to_string(),
                ));
            }
        };
        let body = wire_body_parts(&self.consult.model, self.consult.temperature(), question);
        let bytes = self
            .transport
            .post(
                &self.consult.url,
                self.secret.expose(),
                body.as_bytes(),
                self.consult.timeout(),
            )
            .map_err(RunError::Unavailable)?;
        let content = parse_message_content(&bytes).map_err(RunError::Unavailable)?;
        let neutralised = bxcore::sanitize::neutralise(&content);
        let (content, truncated) = cap_bytes(&neutralised, QUESTION_CAP);
        Ok((RunOutput { content, truncated }, None))
    }
}

/// `choices[0].message.content`, ignoring unknown fields.
pub fn parse_message_content(bytes: &[u8]) -> Result<String, String> {
    let value: serde_json::Value =
        serde_json::from_slice(bytes).map_err(|e| format!("consult response is not JSON: {e}"))?;
    match value
        .get("choices")
        .and_then(|c| c.as_array())
        .and_then(|a| a.first())
        .and_then(|c| c.get("message"))
        .and_then(|m| m.get("content"))
        .and_then(|c| c.as_str())
    {
        Some(content) => Ok(content.to_string()),
        None => Err("the response has no content".to_string()),
    }
}

/// `consult` first, everything else to `inner`.
pub struct ConsultStack {
    pub consult: Box<dyn Runtime>,
    pub inner: Box<dyn Runtime>,
}

impl Runtime for ConsultStack {
    fn run(&self, spec: &RunSpec) -> Result<RunOutput, RunError> {
        self.run_labelled(spec).map(|(output, _)| output)
    }

    fn run_labelled(&self, spec: &RunSpec) -> Result<(RunOutput, Option<Label>), RunError> {
        if spec.tool() == crate::args::ToolName::Consult {
            self.consult.run_labelled(spec)
        } else {
            self.inner.run_labelled(spec)
        }
    }
}

crates/brokerd/src/consult_http.rs

Write this module.

//! Loopback HTTP/1.1 POST for `consult`. HTTPS is a later milestone.

use std::io::{Read, Write};
use std::net::TcpStream;
use std::time::{Duration, Instant};

use crate::consult::{ConsultTransport, parse_loopback_http};

/// Largest response body accepted, in bytes.
pub const MAX_RESPONSE: usize = 262_144;

/// Production transport: one TCP connection to the configured loopback URL.
pub struct HttpTransport;

impl ConsultTransport for HttpTransport {
    fn post(
        &self,
        url: &str,
        bearer: &str,
        body: &[u8],
        timeout: Duration,
    ) -> Result<Vec<u8>, String> {
        post_http(url, bearer, body, timeout)
    }
}

fn post_http(url: &str, bearer: &str, body: &[u8], timeout: Duration) -> Result<Vec<u8>, String> {
    let (host, port, path) = parse_loopback_http(url)?;
    let addr = format!("{host}:{port}");
    let mut stream =
        TcpStream::connect(&addr).map_err(|e| format!("cannot connect to {addr}: {e}"))?;
    stream
        .set_read_timeout(Some(timeout))
        .map_err(|e| format!("cannot set read timeout: {e}"))?;
    stream
        .set_write_timeout(Some(timeout))
        .map_err(|e| format!("cannot set write timeout: {e}"))?;
    let host_header = if port == 80 {
        host
    } else {
        format!("{host}:{port}")
    };
    let length = body.len();
    let header = format!(
        "POST {path} HTTP/1.1\r\nHost: {host_header}\r\nAuthorization: Bearer {bearer}\r\nContent-Type: application/json\r\nContent-Length: {length}\r\nConnection: close\r\n\r\n"
    );
    stream
        .write_all(header.as_bytes())
        .map_err(|e| format!("cannot write consult request: {e}"))?;
    stream
        .write_all(body)
        .map_err(|e| format!("cannot write consult request: {e}"))?;
    stream
        .flush()
        .map_err(|e| format!("cannot write consult request: {e}"))?;
    read_response(&mut stream, timeout)
}

fn read_response(stream: &mut TcpStream, timeout: Duration) -> Result<Vec<u8>, String> {
    let until = Instant::now()
        .checked_add(timeout)
        .ok_or_else(|| "the consult timeout is too large".to_string())?;
    let mut buf = Vec::new();
    let mut chunk = [0u8; 4096];
    loop {
        if Instant::now() >= until {
            return Err("consult timed out".to_string());
        }
        let left = until.saturating_duration_since(Instant::now());
        stream
            .set_read_timeout(Some(left))
            .map_err(|e| format!("cannot set read timeout: {e}"))?;
        match stream.read(&mut chunk) {
            Ok(0) => break,
            Ok(n) => {
                buf.extend_from_slice(&chunk[..n]);
                if buf.len() > MAX_RESPONSE {
                    return Err("consult response is too large".to_string());
                }
            }
            Err(e) if e.kind() == std::io::ErrorKind::TimedOut => {
                return Err("consult timed out".to_string());
            }
            Err(e) => return Err(format!("cannot read consult response: {e}")),
        }
    }
    split_http_body(&buf)
}

fn split_http_body(buf: &[u8]) -> Result<Vec<u8>, String> {
    const SEP: &[u8] = b"\r\n\r\n";
    let sep = buf
        .windows(SEP.len())
        .position(|w| w == SEP)
        .ok_or_else(|| "consult response has no header terminator".to_string())?;
    let (head, rest) = match buf.split_at_checked(sep) {
        Some(pair) => pair,
        None => return Err("consult response is truncated".to_string()),
    };
    let body = match rest.get(SEP.len()..) {
        Some(body) => body,
        None => return Err("consult response is truncated".to_string()),
    };
    let head = std::str::from_utf8(head)
        .map_err(|_| "consult response headers are not UTF-8".to_string())?;
    let status_line = head.lines().next().unwrap_or("");
    if !status_line.contains(" 200 ") && !status_line.ends_with(" 200") {
        return Err(format!("consult HTTP status is not 200: {status_line}"));
    }
    Ok(body.to_vec())
}

crates/brokerd/src/lib.rs

L1. After pub mod consult; add pub mod consult_http;. rustfmt decides the final order.

crates/brokerd/src/main.rs

S1. After use brokerd::config::Config; add:

use brokerd::consult::{self, ConsultRuntime, ConsultStack};
use brokerd::consult_http::HttpTransport;

S2. Immediately after the Config::load match (after its closing };) and before the instance-lock comment, add:

    let consult_secret = match consult::load_secret(&cfg, &|k| std::env::var_os(k)) {
        Ok(loaded) => loaded,
        Err(e) => {
            eprintln!("brokerd: {e}");
            return ExitCode::from(1);
        }
    };
    if let Some(loaded) = &consult_secret
        && let Some(warning) = &loaded.warning
    {
        eprintln!("{warning}");
    }

S3. After the Routed { ... }); block, before let started = match serve::start, add:

    let runtime: Box<dyn Runtime> = match (cfg.consult.clone(), consult_secret) {
        (Some(consult), Some(loaded)) => Box::new(ConsultStack {
            consult: Box::new(ConsultRuntime::new(consult, loaded.secret, HttpTransport)),
            inner: runtime,
        }),
        _ => runtime,
    };

Steps

• 1. Copy the given test.

  cp docs/plans/M6b/files/crates/brokerd/tests/consult_allow.rs crates/brokerd/tests/
  

• 2. See it fail. cargo test -p brokerd --test consult_allow, timeout 600000 ms: the target does not compile (E0432 unresolved ConsultRuntime / ConsultTransport).

• 3. The edits. I1, I2, write consult_http.rs, L1, S1–S3. cargo check -p brokerd prints no error and no warning. wc -l crates/brokerd/src/consult.rs, consult_http.rs, main.rs, broker.rs each ≤ 500.

• 4. The tests. cargo test -p brokerd --test consult_allow --test consult_deny --test memory_port --test memory_tools: every test passes, 2 in consult_allow. The memory-port tests still pass (ConsultStack is not in those tests; Routed is unedited).

• 5. Run the gate. cargo fmt --all, then make gate, timeout 600000 ms. Last line gate: ok. Test count 1611 (1609 before, plus 2). After gate: ok, do not add up test result: lines.

• 6. Log and commit. Task cell M6b/06-brokerd-consult-http. Read only the first 12 lines of docs/implementer-log.md; insert the row under the header. Stage with git add crates/brokerd docs/implementer-log.md Cargo.lock, then commit as AGENTS.md says.

Done when

• make gate prints gate: ok with 1611 tests.
• git diff HEAD~1 --stat lists only crates/brokerd/src/consult.rs, crates/brokerd/src/consult_http.rs, crates/brokerd/src/lib.rs, crates/brokerd/src/main.rs, crates/brokerd/tests/consult_allow.rs and the log.

Stop and report if

• The branch check fails; a "before" block is missing or duplicated; step 2 compiles; a test opens a TCP port or sleeps; you write a chat-template tag whole; cargo check warns; consult.rs exceeds 500 lines; a memory-port test fails; a test outside the named files fails in the gate (git stash -u, rerun, git stash pop, report both).

What the given tests pin

An allowed consult POSTs bytes equal to wire_body; the audit body is those bytes; messages has length 1, role user, the question, no tools; the tool result contains blue and sky and < and not the think tag; class private, untrusted true. A second test with a different question still posts only that question.