Plan M4b: approvals over Mattermost in 15 tasks

Skeletons for new files only; existing files change by exact edits, so no task overwrites an earlier one's work. Each task's tests were run against a reference built in order from master, with the gate at each step and mutations checked; each skeleton compiles and fails its tests, and every todo!() comment was read beside its signature. The reference is kept off this machine. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-24 03:34:21 -07:00
parent b1f732a42a
commit 2d415fc1be
64 changed files with 7701 additions and 2 deletions
@@ -0,0 +1,447 @@
+//! The courier carries a pending approval to `gatewayd` and its answer back, against a fake
+//! `gatewayd` on a real socket (M4b spec, section 5). Do not edit.
+
+#[path = "support/client.rs"]
+mod client;
+#[path = "support/fake_gateway.rs"]
+mod fake_gateway;
+#[path = "support/rig.rs"]
+mod rig;
+#[path = "support/runtime.rs"]
+mod runtime;
+#[path = "support/sink.rs"]
+mod sink;
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use std::os::unix::net::UnixStream;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::time::{Duration, Instant};
+
+use brokerd::broker::{self, Broker};
+use brokerd::config::Gateway;
+use brokerd::courier::{self, Timing};
+use brokerd::ledger::Answer;
+use client::{Serve, next, open};
+use fake_gateway::{FakeGateway, KYLE, POST};
+use proto::{
+    ApprovalAnswer, ApprovalReply, AuditEvent, DenyReason, GatewayOutcome, Message, Timestamp,
+    ToolResponse,
+};
+use rig::{Rig, grant_text, request};
+use runtime::Recording;
+
+const WAIT: Duration = Duration::from_secs(5);
+
+/// A rig whose `[gateway]` is a fake gatewayd in its home, and an `ask` grant for /n.
+fn rig_with_gateway(tag: &str) -> (Rig, FakeGateway) {
+    let mut rig = Rig::new(tag);
+    let socket = rig.dir.path().join("gw.sock");
+    let fake = FakeGateway::start(&socket);
+    rig.cfg.gateway = Some(Gateway { socket });
+    rig.grant(
+        "notes",
+        &grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
+    );
+    (rig, fake)
+}
+
+/// A call that waits for approval; its connection and the approval id.
+fn ask(broker: &Arc<Broker>) -> (UnixStream, u64) {
+    let mut stream = open(
+        broker,
+        broker::handle,
+        9,
+        Message::ToolRequest(request("s1", 9, "read_file", r#"{"path":"/n/a"}"#)),
+    );
+    match next(&mut stream).msg {
+        Message::ToolResponse(ToolResponse::PendingApproval { approval, .. }) => (stream, approval),
+        other => panic!("{other:?}"),
+    }
+}
+
+fn final_answer(stream: &mut UnixStream) -> ToolResponse {
+    let env = next(stream);
+    assert!(env.r#final, "{env:?}");
+    match env.msg {
+        Message::ToolResponse(r) => r,
+        other => panic!("{other:?}"),
+    }
+}
+
+/// The `Approval` record's (answer, by, post, reason).
+fn approval_record(
+    rig: &Rig,
+) -> (
+    ApprovalAnswer,
+    Option<String>,
+    Option<String>,
+    Option<String>,
+) {
+    let events = rig.events();
+    let found = events.iter().find_map(|e| match e {
+        AuditEvent::Approval {
+            answer,
+            by,
+            post,
+            reason,
+            ..
+        } => Some((*answer, by.clone(), post.clone(), reason.clone())),
+        _ => None,
+    });
+    found.expect("an Approval record")
+}
+
+fn some(s: &str) -> Option<String> {
+    Some(s.to_string())
+}
+
+/// Sets the courier's stop flag when dropped, so a failing assertion ends the courier too.
+struct StopOnDrop<'a>(&'a AtomicBool);
+
+impl Drop for StopOnDrop<'_> {
+    fn drop(&mut self) {
+        self.0.store(true, std::sync::atomic::Ordering::SeqCst);
+    }
+}
+
+/// The courier must end by itself within five seconds.
+fn ends_by_itself<T>(courier: &std::thread::ScopedJoinHandle<'_, T>) {
+    let until = Instant::now() + WAIT;
+    while !courier.is_finished() {
+        assert!(Instant::now() < until, "the courier did not stop");
+        std::thread::sleep(Duration::from_millis(20));
+    }
+}
+
+#[test]
+fn approved_in_mattermost_it_runs_and_the_record_names_who_and_where() {
+    let (rig, fake) = rig_with_gateway("courier-approve");
+    let rt = Recording::answering("the notes");
+    let broker = rig.broker(&rt);
+    let (mut stream, id) = ask(&broker);
+    let mut conn = fake.next(WAIT).expect("the approval reaches gatewayd");
+    assert_eq!(
+        (conn.request.approval, conn.request.tool.as_str()),
+        (id, "read_file")
+    );
+    assert_eq!(conn.request.arguments, r#"{"path":"/n/a"}"#);
+    conn.posted();
+    conn.approve();
+    assert_eq!(conn.outcome(), Some(GatewayOutcome::Runs {}));
+    assert!(matches!(
+        final_answer(&mut stream),
+        ToolResponse::Result { .. }
+    ));
+    assert_eq!(rt.count(), 1);
+    assert_eq!(
+        approval_record(&rig),
+        (ApprovalAnswer::Approved, some(KYLE), some(POST), None)
+    );
+}
+
+#[test]
+fn refused_in_mattermost_it_is_denied_with_the_reason() {
+    let (rig, fake) = rig_with_gateway("courier-refuse");
+    let rt = Recording::answering("x");
+    let broker = rig.broker(&rt);
+    let (mut stream, _) = ask(&broker);
+    let mut conn = fake.next(WAIT).unwrap();
+    conn.posted();
+    conn.refuse("not on a Friday");
+    assert_eq!(conn.outcome(), Some(GatewayOutcome::Refused {}));
+    assert_eq!(
+        final_answer(&mut stream),
+        ToolResponse::Denied {
+            reason: DenyReason::ApprovalRefused
+        }
+    );
+    assert_eq!(rt.count(), 0);
+    assert_eq!(
+        approval_record(&rig),
+        (
+            ApprovalAnswer::Refused,
+            some(KYLE),
+            some(POST),
+            some("not on a Friday")
+        )
+    );
+}
+
+#[test]
+fn approved_after_the_grant_is_gone_it_is_denied_on_the_recheck() {
+    let (rig, fake) = rig_with_gateway("courier-recheck");
+    let rt = Recording::answering("x");
+    let broker = rig.broker(&rt);
+    let (mut stream, _) = ask(&broker);
+    let mut conn = fake.next(WAIT).unwrap();
+    conn.posted();
+    rig.remove_grant("notes");
+    conn.approve();
+    assert_eq!(
+        conn.outcome(),
+        Some(GatewayOutcome::Denied {
+            reason: DenyReason::NoGrant
+        })
+    );
+    assert_eq!(
+        final_answer(&mut stream),
+        ToolResponse::Denied {
+            reason: DenyReason::NoGrant
+        }
+    );
+    assert_eq!(rt.count(), 0);
+}
+
+#[test]
+fn answered_in_bxctl_the_courier_closes_without_an_outcome() {
+    let (rig, fake) = rig_with_gateway("courier-bxctl");
+    let rt = Recording::answering("x");
+    let broker = rig.broker(&rt);
+    let (mut stream, id) = ask(&broker);
+    let mut conn = fake.next(WAIT).unwrap();
+    conn.posted();
+    let entry = broker.table().take(id).unwrap();
+    brokerd::admin::answer(
+        &broker,
+        entry,
+        Answer::Approved {
+            by: some("bxctl"),
+            post: None,
+        },
+        Timestamp::now(),
+    );
+    assert!(
+        conn.closed_within(WAIT),
+        "the courier closes, and sends nothing"
+    );
+    assert!(matches!(
+        final_answer(&mut stream),
+        ToolResponse::Result { .. }
+    ));
+    assert_eq!(approval_record(&rig).1, some("bxctl"));
+}
+
+#[test]
+fn an_answer_after_it_ended_is_gone() {
+    let rig = Rig::new("courier-gone");
+    rig.grant(
+        "notes",
+        &grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
+    );
+    let rt = Recording::answering("x");
+    let broker = rig.broker(&rt);
+    let (_stream, id) = ask(&broker);
+    // The courier is run by hand here, so that the answer can arrive after the entry is taken.
+    let socket = rig.dir.path().join("gw.sock");
+    let fake = FakeGateway::start(&socket);
+    let info = broker.table().list().remove(0);
+    let stop = AtomicBool::new(false);
+    std::thread::scope(|scope| {
+        let courier = scope.spawn(|| {
+            courier::carry(
+                &broker,
+                &socket,
+                &info,
+                &stop,
+                Timing {
+                    retry: Duration::from_millis(100),
+                    tick: Duration::from_secs(5),
+                },
+            )
+        });
+        let _guard = StopOnDrop(&stop);
+        let mut conn = fake.next(WAIT).unwrap();
+        conn.posted();
+        let entry = broker.table().take(id).unwrap();
+        brokerd::admin::answer(&broker, entry, Answer::Expired, Timestamp::now());
+        conn.approve();
+        assert_eq!(conn.outcome(), Some(GatewayOutcome::Gone {}));
+        ends_by_itself(&courier);
+    });
+}
+
+#[test]
+fn if_gatewayd_goes_away_the_approval_is_posted_again() {
+    let (rig, fake) = rig_with_gateway("courier-again");
+    let rt = Recording::answering("x");
+    let broker = rig.broker(&rt);
+    let (mut stream, id) = ask(&broker);
+    let mut first = fake.next(WAIT).unwrap();
+    first.posted();
+    first.close();
+    let mut second = fake.next(Duration::from_secs(10)).expect("posted again");
+    assert_eq!(second.request.approval, id);
+    second.posted();
+    second.approve();
+    assert_eq!(second.outcome(), Some(GatewayOutcome::Runs {}));
+    assert!(matches!(
+        final_answer(&mut stream),
+        ToolResponse::Result { .. }
+    ));
+    assert!(
+        rig.lines.all().iter().any(|l| l.contains(&format!(
+            "approval {id}: gatewayd went away; posting it again"
+        ))),
+        "{:?}",
+        rig.lines.all()
+    );
+}
+
+#[test]
+fn a_bad_answer_answers_nothing() {
+    let (rig, fake) = rig_with_gateway("courier-bad");
+    let rt = Recording::answering("x");
+    let broker = rig.broker(&rt);
+    let (_stream, id) = ask(&broker);
+    for bad in [
+        ApprovalReply {
+            approval: id + 1,
+            approved: true,
+            by: KYLE.to_string(),
+            post: POST.to_string(),
+            reason: None,
+        },
+        ApprovalReply {
+            approval: id,
+            approved: true,
+            by: "kyle".to_string(),
+            post: POST.to_string(),
+            reason: None,
+        },
+        ApprovalReply {
+            approval: id,
+            approved: false,
+            by: KYLE.to_string(),
+            post: POST.to_string(),
+            reason: Some("x".repeat(501)),
+        },
+    ] {
+        let mut conn = fake.next(Duration::from_secs(10)).unwrap();
+        conn.posted();
+        conn.reply(bad);
+        assert_eq!(conn.outcome(), None, "no outcome for a bad answer");
+        assert!(broker.table().contains(id), "still pending");
+    }
+    assert_eq!(rt.count(), 0);
+}
+
+#[test]
+fn unreachable_at_first_then_posted_and_said_once() {
+    let rig = Rig::new("courier-late");
+    rig.grant(
+        "notes",
+        &grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
+    );
+    let rt = Recording::answering("x");
+    let broker = rig.broker(&rt);
+    let (_stream, id) = ask(&broker);
+    let socket = rig.dir.path().join("gw.sock");
+    let info = broker.table().list().remove(0);
+    let stop = AtomicBool::new(false);
+    let timing = Timing {
+        retry: Duration::from_millis(100),
+        tick: Duration::from_millis(20),
+    };
+    std::thread::scope(|scope| {
+        let courier = scope.spawn(|| courier::carry(&broker, &socket, &info, &stop, timing));
+        let _guard = StopOnDrop(&stop);
+        std::thread::sleep(Duration::from_millis(500));
+        let fake = FakeGateway::start(&socket);
+        let mut conn = fake.next(WAIT).unwrap();
+        conn.posted();
+        let until = Instant::now() + WAIT;
+        while !rig
+            .lines
+            .all()
+            .iter()
+            .any(|l| l.contains("posted in Mattermost"))
+        {
+            assert!(Instant::now() < until, "{:?}", rig.lines.all());
+            std::thread::sleep(Duration::from_millis(20));
+        }
+        broker.table().take(id);
+        ends_by_itself(&courier);
+    });
+    let lines = rig.lines.all();
+    let unreachable: Vec<&String> = lines
+        .iter()
+        .filter(|l| l.contains("gatewayd is not reachable"))
+        .collect();
+    assert_eq!(
+        unreachable.len(),
+        1,
+        "said once, not every retry: {lines:?}"
+    );
+    assert!(
+        unreachable[0].starts_with(&format!(
+            "brokerd: approval {id}: gatewayd is not reachable at "
+        )),
+        "{lines:?}"
+    );
+    assert!(
+        unreachable[0]
+            .ends_with("; bxctl can still answer it\nsee docs/runbook.md#gateway-unreachable"),
+        "{lines:?}"
+    );
+    assert!(
+        lines.contains(&format!("brokerd: approval {id} posted in Mattermost")),
+        "{lines:?}"
+    );
+}
+
+#[test]
+fn if_gatewayd_cannot_post_it_is_tried_again() {
+    let (rig, fake) = rig_with_gateway("courier-cannot");
+    let rt = Recording::answering("x");
+    let broker = rig.broker(&rt);
+    let (mut stream, _) = ask(&broker);
+    fake.next(WAIT).unwrap().cannot_post();
+    let mut conn = fake.next(Duration::from_secs(10)).expect("tried again");
+    conn.posted();
+    conn.approve();
+    assert_eq!(conn.outcome(), Some(GatewayOutcome::Runs {}));
+    assert!(matches!(
+        final_answer(&mut stream),
+        ToolResponse::Result { .. }
+    ));
+    assert!(
+        rig.lines
+            .all()
+            .iter()
+            .any(|l| l.contains("gatewayd could not post it: Mattermost is down"))
+    );
+}
+
+#[test]
+fn without_a_gateway_nothing_is_carried() {
+    let rig = Rig::new("courier-none");
+    rig.grant(
+        "notes",
+        &grant_text("read_file", "ask", "", "paths = [\"/n\"]"),
+    );
+    let socket = rig.dir.path().join("gw.sock");
+    let fake = FakeGateway::start(&socket);
+    let rt = Recording::answering("x");
+    let broker = rig.broker(&rt);
+    let (mut stream, id) = ask(&broker);
+    assert!(
+        fake.next(Duration::from_millis(500)).is_none(),
+        "no [gateway]: no connection"
+    );
+    let entry = broker.table().take(id).unwrap();
+    brokerd::admin::answer(
+        &broker,
+        entry,
+        Answer::Approved {
+            by: some("bxctl"),
+            post: None,
+        },
+        Timestamp::now(),
+    );
+    assert!(matches!(
+        final_answer(&mut stream),
+        ToolResponse::Result { .. }
+    ));
+}