gatewayd: serve, posting approvals, marking them, and the listener

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

crates/gatewayd/tests/serve_approvals.rs

@@ -0,0 +1,199 @@
+//! The serve loop posts what `gateway.sock` brings, where it belongs, and marks it when it ends
+//! or when a restart left it behind (M4b spec, sections 6 and 8). Do not edit.
+
+#[path = "support/brokerd_client.rs"]
+mod brokerd_client;
+#[path = "support/fake_loop.rs"]
+mod fake_loop;
+#[path = "support/fake_mm.rs"]
+mod fake_mm;
+#[path = "support/gateway.rs"]
+mod gateway;
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use std::path::PathBuf;
+use std::sync::{Mutex, mpsc};
+use std::time::Duration;
+
+use brokerd_client::{ask, pending};
+use fake_loop::{done, serve_loop};
+use fake_mm::{DM, FakeMm, KYLE, id, post};
+use gateway::{WAIT, config, loop_dir, read_state, start};
+use tmp::TempDir;
+
+fn socket(home: &TempDir) -> PathBuf {
+    home.path().join("run/gateway/gateway.sock")
+}
+
+/// A turn that waits until the returned sender says so.
+fn held_loop(home: &TempDir) -> mpsc::Sender<()> {
+    loop_dir(home);
+    let (release_tx, release_rx) = mpsc::channel::<()>();
+    let release = Mutex::new(release_rx);
+    let _turns = serve_loop(&home.path().join("run/loop/loop.sock"), move |_, _| {
+        let _ = release
+            .lock()
+            .unwrap()
+            .recv_timeout(Duration::from_secs(20));
+        vec![done("done")]
+    });
+    release_tx
+}
+
+#[test]
+fn an_approval_goes_into_the_asking_thread_and_is_marked_when_it_ends() {
+    let home = TempDir::new("sa-thread");
+    let release = held_loop(&home);
+    let mm = FakeMm::start();
+    let _running = start(config(&home, &mm.url(), ""));
+    let mut ws = mm.next_ws(WAIT);
+    let p1 = id('p', 1);
+    ws.posted(&post(&p1, KYLE, DM, "", "read my notes", 5), "D");
+    let until = std::time::Instant::now() + WAIT;
+    let path = home.path().join("gateway/state.json");
+    let in_flight = || -> bool {
+        let Ok(text) = std::fs::read_to_string(&path) else {
+            return false;
+        };
+        let v: serde_json::Value = serde_json::from_str(&text).unwrap();
+        !v["in_flight"].as_array().unwrap().is_empty()
+    };
+    while !in_flight() {
+        assert!(std::time::Instant::now() < until, "the turn never started");
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    let asking = ask(
+        &socket(&home),
+        pending(42, &format!("mm-{p1}"), r#"{"path":"/n/a"}"#),
+    )
+    .unwrap();
+    let posts = mm.posts();
+    assert_eq!(posts.len(), 1, "{posts:?}");
+    let (channel, root, text) = &posts[0];
+    assert_eq!(
+        (channel.as_str(), root.as_str()),
+        (DM, p1.as_str()),
+        "in the asking thread"
+    );
+    assert!(text.starts_with("**Approval 42** · tool `read_file` · grant `notes` · taint private · expires 2026-09-24 08:20 UTC\n```\n{\"path\":\"/n/a\"}\n```\n"), "{text}");
+    assert_eq!(asking.posts, [FakeMm::post_id(0)]);
+    assert_eq!(
+        read_state(&home)["approvals"],
+        serde_json::json!([{"approval": 42, "channel": DM, "posts": [FakeMm::post_id(0)]}])
+    );
+    asking.close();
+    let edits = mm.wait_edits(1, WAIT);
+    assert_eq!(edits.len(), 1, "{edits:?}");
+    assert_eq!(edits[0].0, FakeMm::post_id(0));
+    assert_eq!(
+        edits[0].1,
+        format!("{text}\n\nNo longer pending (answered elsewhere, expired, or its turn ended).")
+    );
+    let until = std::time::Instant::now() + WAIT;
+    while !read_state(&home)["approvals"]
+        .as_array()
+        .unwrap()
+        .is_empty()
+    {
+        assert!(std::time::Instant::now() < until, "still recorded");
+        std::thread::sleep(Duration::from_millis(20));
+    }
+    release.send(()).unwrap();
+}
+
+#[test]
+fn without_a_turn_in_flight_it_goes_to_the_first_approvers_direct_messages_in_parts() {
+    let home = TempDir::new("sa-dm");
+    loop_dir(&home);
+    let mm = FakeMm::start();
+    let _running = start(config(&home, &mm.url(), ""));
+    let _ws = mm.next_ws(WAIT);
+    let long = format!("{{\"content\":\"{}\"}}", "x".repeat(20_000));
+    let asking = ask(&socket(&home), pending(7, "chat-1789700000-1", &long)).unwrap();
+    let posts = mm.posts();
+    assert_eq!(posts.len(), 2, "two parts");
+    assert_eq!(
+        (posts[0].0.as_str(), posts[0].1.as_str()),
+        (DM, ""),
+        "a new thread in the direct channel"
+    );
+    assert_eq!(
+        (posts[1].0.as_str(), posts[1].1.clone()),
+        (DM, FakeMm::post_id(0)),
+        "the next part in its thread"
+    );
+    assert!(
+        posts[1]
+            .2
+            .contains("Reply in this thread with `!approve 7`")
+    );
+    assert_eq!(asking.posts, [FakeMm::post_id(0), FakeMm::post_id(1)]);
+    asking.close();
+    let edits = mm.wait_edits(1, WAIT);
+    assert_eq!(
+        edits[0].0,
+        FakeMm::post_id(1),
+        "only the last part is edited"
+    );
+}
+
+#[test]
+fn an_approval_that_cannot_be_posted_is_refused_to_brokerd() {
+    let home = TempDir::new("sa-refused");
+    loop_dir(&home);
+    let mm = FakeMm::start();
+    let _running = start(config(&home, &mm.url(), ""));
+    let _ws = mm.next_ws(WAIT);
+    mm.refuse_posts(true);
+    let why = ask(&socket(&home), pending(8, "chat-1789700000-1", "{}"))
+        .err()
+        .expect("refused");
+    assert!(
+        why.starts_with("cannot post approval 8: status 400"),
+        "{why}"
+    );
+    assert!(
+        read_state(&home)["approvals"]
+            .as_array()
+            .unwrap()
+            .is_empty()
+    );
+}
+
+#[test]
+fn a_restart_marks_the_posts_it_left_behind() {
+    let home = TempDir::new("sa-restart");
+    let (a, b) = (id('q', 1), id('q', 2));
+    let state = serde_json::json!({
+        "channels": {}, "recent": [], "threads": [], "in_flight": [],
+        "approvals": [{"approval": 5, "channel": DM, "posts": [a, b]}]
+    });
+    home.write("gateway/state.json", &state.to_string());
+    loop_dir(&home);
+    let mm = FakeMm::start();
+    let _running = start(config(&home, &mm.url(), ""));
+    let _ws = mm.next_ws(WAIT);
+    let edits = mm.wait_edits(1, WAIT);
+    assert_eq!(edits, [(b, "**Approval 5**\n\nNo longer answerable here: gatewayd restarted. If it is still pending it is posted again.".to_string())]);
+    assert!(
+        read_state(&home)["approvals"]
+            .as_array()
+            .unwrap()
+            .is_empty()
+    );
+}
+
+#[test]
+fn the_socket_is_there_while_gatewayd_runs() {
+    let home = TempDir::new("sa-socket");
+    loop_dir(&home);
+    let mm = FakeMm::start();
+    let running = start(config(&home, &mm.url(), ""));
+    let _ws = mm.next_ws(WAIT);
+    use std::os::unix::fs::{FileTypeExt, PermissionsExt};
+    let meta = std::fs::metadata(socket(&home)).unwrap();
+    assert!(meta.file_type().is_socket());
+    assert_eq!(meta.permissions().mode() & 0o777, 0o600);
+    drop(running);
+}

crates/gatewayd/tests/support/brokerd_client.rs

@@ -0,0 +1,97 @@
+//! `brokerd`'s side of one `gateway.sock` connection, for tests: ask, read what comes back,
+//! answer with an outcome, or close. Do not edit.
+
+#![allow(dead_code)] // each test file uses a different part of this module
+
+use std::os::unix::net::UnixStream;
+use std::path::Path;
+use std::time::Duration;
+
+use proto::{
+    ApprovalOutcome, ApprovalReply, ApprovalRequest, CallId, DataClass, Envelope, GatewayOutcome,
+    Message, PROTOCOL_VERSION, PendingApproval, SessionId, Timestamp, read_frame, write_frame,
+};
+
+pub struct Asking {
+    stream: UnixStream,
+    pub posts: Vec<String>,
+}
+
+/// An approval of `session`, with these arguments.
+pub fn pending(approval: u64, session: &str, arguments: &str) -> PendingApproval {
+    PendingApproval {
+        approval,
+        session: SessionId::new(session).unwrap(),
+        call: CallId(3),
+        tool: "read_file".to_string(),
+        arguments: arguments.to_string(),
+        grant: "notes".to_string(),
+        taint: DataClass::Private,
+        created: Timestamp::parse("2026-09-24T08:05:00.000Z").unwrap(),
+        expires: Timestamp::parse("2026-09-24T08:20:00.000Z").unwrap(),
+    }
+}
+
+fn frame(id: u64, last: bool, msg: Message) -> Envelope {
+    Envelope {
+        v: PROTOCOL_VERSION,
+        id,
+        r#final: last,
+        msg,
+    }
+}
+
+/// Connect and ask; `Ok` with the posts, or `Err` with gatewayd's error detail.
+pub fn ask(socket: &Path, info: PendingApproval) -> Result<Asking, String> {
+    let mut stream = UnixStream::connect(socket).map_err(|e| e.to_string())?;
+    stream
+        .set_read_timeout(Some(Duration::from_secs(10)))
+        .unwrap();
+    write_frame(
+        &mut stream,
+        &frame(
+            1,
+            true,
+            Message::ApprovalRequest(ApprovalRequest { approval: info }),
+        ),
+    )
+    .unwrap();
+    match read_frame(&mut stream).map_err(|e| e.to_string())?.msg {
+        Message::ApprovalPosted(p) => Ok(Asking {
+            stream,
+            posts: p.posts,
+        }),
+        Message::Error(e) => Err(e.detail),
+        other => panic!("{other:?}"),
+    }
+}
+
+impl Asking {
+    /// The answer, within `wait`, or `None`.
+    pub fn answer(&mut self, wait: Duration) -> Option<ApprovalReply> {
+        self.stream.set_read_timeout(Some(wait)).unwrap();
+        match read_frame(&mut self.stream) {
+            Ok(Envelope {
+                msg: Message::ApprovalReply(r),
+                ..
+            }) => Some(r),
+            _ => None,
+        }
+    }
+
+    pub fn outcome(&mut self, outcome: GatewayOutcome) {
+        write_frame(
+            &mut self.stream,
+            &frame(
+                2,
+                true,
+                Message::ApprovalOutcome(ApprovalOutcome { outcome }),
+            ),
+        )
+        .unwrap();
+    }
+
+    pub fn close(self) {
+        let _ = self.stream.shutdown(std::net::Shutdown::Both);
+    }
+}

crates/gatewayd/tests/support/fake_mm.rs

@@ -30,6 +30,10 @@ struct Inner {
     posts: Vec<(String, String, String)>,
     /// Every REST call: method and path.
     calls: Vec<(String, String)>,
+    /// Every edit: the post and its new text.
+    edits: Vec<(String, String)>,
+    /// When set, `POST /posts` is refused with 400.
+    refuse_posts: bool,
 }
 
 pub struct FakeMm {
@@ -223,6 +227,28 @@ impl FakeMm {
     pub fn calls(&self) -> Vec<(String, String)> {
         self.inner.lock().unwrap().calls.clone()
     }
+
+    /// The id of the n-th post made (from 0).
+    pub fn post_id(n: usize) -> String {
+        id('x', u32::try_from(n).unwrap())
+    }
+
+    pub fn edits(&self) -> Vec<(String, String)> {
+        self.inner.lock().unwrap().edits.clone()
+    }
+
+    /// Wait until at least `n` edits were made, for at most `wait`.
+    pub fn wait_edits(&self, n: usize, wait: Duration) -> Vec<(String, String)> {
+        let until = Instant::now() + wait;
+        while self.edits().len() < n && Instant::now() < until {
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        self.edits()
+    }
+
+    pub fn refuse_posts(&self, refuse: bool) {
+        self.inner.lock().unwrap().refuse_posts = refuse;
+    }
 }
 
 fn connection(mut stream: TcpStream, inner: &Mutex<Inner>, ws_tx: &mpsc::Sender<WsPeer>) {
@@ -298,6 +324,9 @@ fn rest(inner: &Mutex<Inner>, method: &str, path: &str, body: &[u8]) -> (u16, Va
             };
             (201, json!({"id": channel, "type": "D"}))
         }
+        ("POST", "/api/v4/posts") if inner.refuse_posts => {
+            (400, json!({"id": "api.post.create_post.refused"}))
+        }
         ("POST", "/api/v4/posts") => {
             let p: Value = serde_json::from_slice(body).unwrap();
             let n = u32::try_from(inner.posts.len()).unwrap();
@@ -326,6 +355,27 @@ fn rest(inner: &Mutex<Inner>, method: &str, path: &str, body: &[u8]) -> (u16, Va
                     .unwrap_or(json!({"order": [], "posts": {}})),
             )
         }
+        ("PUT", p) if p.starts_with("/api/v4/posts/") && p.ends_with("/patch") => {
+            let post_id = p
+                .trim_start_matches("/api/v4/posts/")
+                .trim_end_matches("/patch")
+                .to_string();
+            let message = serde_json::from_slice::<Value>(body).unwrap()["message"]
+                .as_str()
+                .unwrap()
+                .to_string();
+            inner.edits.push((post_id.clone(), message.clone()));
+            (200, post(&post_id, BOT, DM, "", &message, 1))
+        }
+        ("GET", p) if p.starts_with("/api/v4/users/") => {
+            let user = p.trim_start_matches("/api/v4/users/");
+            let name = if user == KYLE {
+                "kyle".to_string()
+            } else {
+                format!("user-{}", &user[..4])
+            };
+            (200, json!({"id": user, "username": name}))
+        }
         _ => (404, json!({"message": "not found"})),
     }
 }