# 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).

```sh
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
  `&lt;` 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

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

use crate::config::Config;
```

become

```rust
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:

```rust
/// 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.

```rust
//! 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:

```rust
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:

```rust
    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:

```rust
    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.**

  ```sh
  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 `&lt;` and not the think tag; class private, untrusted true. A
second test with a different question still posts only that question.