M4b review, and follow-up tasks 16 to 18

The review of tasks 01 to 15 goes in the implementer log, with the device check against the owner's Mattermost. Three follow-ups: - 16: the refusal without a runner says why. - 17: the texts that say who asked. - 18: approvals from non-approvers in direct or group messages go to an approver. Task 14's given serve_approvals.rs becomes serve_approvals.rs-task14, since task 18 hands over a new version. Tip T31 is added. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-24 11:35:24 -07:00
parent d2e814161f
commit f6dd7f1741
12 changed files with 1202 additions and 19 deletions
@@ -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);
+}