From c38e450f44257835b06c1cfe0745ab5ab6dce2ca Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Thu, 24 Sep 2026 05:07:50 -0700 Subject: [PATCH] gatewayd: serve, posting approvals, marking them, and the listener Implemented-By: OpenCode session (model recorded in docs/implementer-log.md) --- crates/gatewayd/src/serve/approvals.rs | 197 +++++++++++++++++ crates/gatewayd/src/serve/mod.rs | 38 +++- crates/gatewayd/tests/serve_approvals.rs | 199 ++++++++++++++++++ .../gatewayd/tests/support/brokerd_client.rs | 97 +++++++++ crates/gatewayd/tests/support/fake_mm.rs | 50 +++++ docs/implementer-log.md | 1 + 6 files changed, 580 insertions(+), 2 deletions(-) create mode 100644 crates/gatewayd/src/serve/approvals.rs create mode 100644 crates/gatewayd/tests/serve_approvals.rs create mode 100644 crates/gatewayd/tests/support/brokerd_client.rs diff --git a/crates/gatewayd/src/serve/approvals.rs b/crates/gatewayd/src/serve/approvals.rs new file mode 100644 index 0000000..1c756d4 --- /dev/null +++ b/crates/gatewayd/src/serve/approvals.rs @@ -0,0 +1,197 @@ +//! The serve loop's part in approvals (M4b spec, sections 6 and 8): posting what a `gateway.sock` +//! connection brings, marking it when it ends, and marking what a restart left behind. The loop +//! alone touches Mattermost and the state file; connection threads only ask. + +use std::sync::mpsc::Sender; + +use proto::{ApprovalReply, GatewayOutcome, PendingApproval}; + +use crate::approvals::Posted; +use crate::approvals::text::{self, Ended}; +use crate::gateway::{Ending, ToLoop}; +use crate::serve::{Gateway, Stop}; +use crate::state::ApprovalPosts; + +/// An approval's last post, the one that is edited when it ends. +#[derive(Debug)] +pub(super) struct Open { + pub(super) last_post: String, + pub(super) last_text: String, +} + +impl Gateway { + /// Everything the connection threads asked since the last look, never blocking. + pub(super) fn gateway_requests(&mut self) -> Result<(), Stop> { + while let Ok(asked) = self.gateway_rx.try_recv() { + match asked { + ToLoop::Show { + info, + answer, + shown, + } => { + let result = self.show(&info, answer)?; + let _ = shown.send(result); + } + ToLoop::Ended { approval, ending } => self.ended(approval, ending)?, + } + } + Ok(()) + } + + /// Where an approval goes: its session's turn in flight, or the first approver's direct + /// messages as a new thread (an empty root). + fn place(&self, info: &PendingApproval) -> Result<(String, String), String> { + // 1. `self.state.turn_of(info.session.as_str())`: if Some(turn), `Ok((turn.channel, + // turn.root))`. + // 2. Otherwise the first approver's direct messages, as a new thread: + // `self.config.first_approver()` (None is Err("no approver is configured")), then + // `self.client.direct_channel(&self.me.id, first)` (an Err(e) is Err(format!("no direct + // channel with the first approver: {e}"))). `Ok((channel, String::new()))`. + if let Some(turn) = self.state.turn_of(info.session.as_str()) { + return Ok((turn.channel, turn.root)); + } + let first = match self.config.first_approver() { + Some(first) => first, + None => return Err("no approver is configured".to_string()), + }; + match self.client.direct_channel(&self.me.id, first) { + Ok(channel) => Ok((channel, String::new())), + Err(e) => Err(format!("no direct channel with the first approver: {e}")), + } + } + + /// Post an approval's parts in order, record them, and register it for answers. An `Err` + /// inside is why it could not be posted; the outer one stops `gatewayd` (the state file). + fn show( + &mut self, + info: &PendingApproval, + answer: Sender, + ) -> Result, String>, Stop> { + let id = info.approval; + let (channel, mut root) = match self.place(info) { + Ok(place) => place, + Err(why) => return Ok(Err(why)), + }; + let shown = text::render(info); + let mut posts = Vec::new(); + let mut shown_at = 0; + for body in &shown.posts { + match self.client.create_post(&channel, &root, body) { + Ok(post) => { + if root.is_empty() { + root = post.id.clone(); + } + shown_at = post.create_at; + posts.push(post.id); + } + Err(e) => { + // Parts already posted must not look answerable. + if let (Some(last), Some(text)) = + (posts.last(), shown.posts.get(posts.len().wrapping_sub(1))) + { + self.mark(id, last, &text::with_ending(text, &Ended::NoLonger)); + } + return Ok(Err(format!("cannot post approval {id}: {e}"))); + } + } + } + self.state.add_approval(ApprovalPosts { + approval: id, + channel: channel.clone(), + posts: posts.clone(), + })?; + self.registry.add(Posted { + approval: id, + session: info.session.as_str().to_string(), + channel, + root, + shown_at, + answerable: shown.answerable, + answer, + }); + if let (Some(last_post), Some(last_text)) = (posts.last(), shown.posts.last()) { + self.open.insert( + id, + Open { + last_post: last_post.clone(), + last_text: last_text.clone(), + }, + ); + } + Ok(Ok(posts)) + } + + /// An approval ended: forget it, and add how it ended to its last post. + fn ended(&mut self, id: u64, ending: Ending) -> Result<(), Stop> { + // 1. `self.registry.remove(id);` and `self.state.remove_approval(id)?;`. + // 2. `let Some(open) = self.open.remove(&id) else { return Ok(()); };` + // 3. The `Ended`: Ending::NoLonger -> Ended::NoLonger. Ending::Answered { by, reason, + // outcome } -> `by` becomes "@" from `self.client.username(&by)`, or stays the + // id if that fails; then outcome Runs -> Ended::Runs { by }, Denied { reason } -> + // Ended::Denied { by, reason }, Refused -> Ended::Refused { by, reason }, Gone -> + // Ended::NoLonger. + // 4. `self.mark(id, &open.last_post, &text::with_ending(&open.last_text, &ended));` and + // Ok(()). + self.registry.remove(id); + self.state.remove_approval(id)?; + let Some(open) = self.open.remove(&id) else { + return Ok(()); + }; + let ended = match ending { + Ending::NoLonger => Ended::NoLonger, + Ending::Answered { + by, + reason: _, + outcome, + } => { + let by = match self.client.username(&by) { + Ok(username) => format!("@{username}"), + Err(_) => by, + }; + match outcome { + GatewayOutcome::Runs {} => Ended::Runs { by }, + GatewayOutcome::Denied { reason } => Ended::Denied { by, reason }, + GatewayOutcome::Refused {} => Ended::Refused { by, reason: None }, + GatewayOutcome::Gone {} => Ended::NoLonger, + } + } + }; + self.mark( + id, + &open.last_post, + &text::with_ending(&open.last_text, &ended), + ); + Ok(()) + } + + /// At the first connection after a start: the approval posts the last run left are no longer + /// answerable. Their last post is replaced by its title and the restart line. + pub(super) fn mark_left_behind(&mut self) -> Result<(), Stop> { + // For each `left` in `self.state.take_approvals()?`: if it has a last post, + // `self.mark(left.approval, last, &body)` with `body = format!("**Approval {}**\n\n{}", + // left.approval, text::ended_line(&Ended::Restarted))`. Then Ok(()). + for left in self.state.take_approvals()? { + let Some(last) = left.posts.last() else { + continue; + }; + let body = format!( + "**Approval {}**\n\n{}", + left.approval, + text::ended_line(&Ended::Restarted) + ); + self.mark(left.approval, last, &body); + } + Ok(()) + } + + /// Edit a post; a failure is logged, not retried. + fn mark(&self, id: u64, post: &str, body: &str) { + // `if let Err(e) = self.client.patch_post(post, body)`: log "gatewayd: cannot mark approval + // {id} as ended: {e}" with `(self.log)(&line)`. + if let Err(e) = self.client.patch_post(post, body) { + (self.log)(&format!( + "gatewayd: cannot mark approval {id} as ended: {e}" + )); + } + } +} diff --git a/crates/gatewayd/src/serve/mod.rs b/crates/gatewayd/src/serve/mod.rs index 32f0b9b..51eb69b 100644 --- a/crates/gatewayd/src/serve/mod.rs +++ b/crates/gatewayd/src/serve/mod.rs @@ -1,8 +1,10 @@ //! Startup, the event loop and reconnecting (M4a spec, section 9). `run` returns only when //! `gatewayd` must stop: a refused token, a state file it cannot keep, or a stop asked by a test. +mod approvals; mod handle; +use std::collections::HashMap; use std::fs::File; use std::path::PathBuf; use std::sync::Arc; @@ -12,7 +14,9 @@ use std::time::{Duration, Instant}; use proto::SessionId; +use crate::approvals::Registry; use crate::config::Config; +use crate::gateway::{self, ToLoop}; use crate::mm::rest::Client; use crate::mm::{Event, Me, MmError, parse_event}; use crate::net::Connector; @@ -102,6 +106,12 @@ pub(crate) struct Gateway { seq: u64, /// The turns a restart cut off have been answered. restarted: bool, + /// The approvals posted here, shared with the turns (for the notice) and judged by commands. + registry: Arc, + /// What `gateway.sock` connections ask of the loop. + gateway_rx: Receiver, + /// The last post of each approval posted here and not yet ended. + open: HashMap, } /// The wait before attempt `n` (from 0) after a loss. @@ -143,8 +153,21 @@ pub fn run(config: Config, token: Secret, tuning: Tuning, log: Log, stop: &Atomi Ok(s) => s, Err(e) => return Stop::State(e), }; + let listener = match gateway::listen(&config.gateway_socket()) { + Ok(listener) => listener, + Err(why) => return Stop::Start(why), + }; + let (gateway_tx, gateway_rx) = mpsc::channel(); + if let Err(e) = gateway::serve( + listener, + gateway_tx, + gateway::Timing::default(), + Arc::clone(&log), + ) { + return Stop::Start(format!("cannot start the gateway.sock thread: {e}")); + } let client = Arc::new(Client::new(connector, token, tuning.rest_timeout)); - let mut g = Gateway::new(config, client, state, log); + let mut g = Gateway::new(config, client, state, log, gateway_rx); let mut failures = 0; loop { if stop.load(Ordering::SeqCst) { @@ -206,7 +229,13 @@ fn connect(config: &Config, client: &Client, tuning: &Tuning) -> Result<(Me, Ws) } impl Gateway { - fn new(config: Config, client: Arc, state: State, log: Log) -> Gateway { + fn new( + config: Config, + client: Arc, + state: State, + log: Log, + gateway_rx: Receiver, + ) -> Gateway { let (done_tx, done_rx) = mpsc::channel(); let limit = usize::try_from(config.limits.queue).unwrap_or(usize::MAX); Gateway { @@ -226,6 +255,9 @@ impl Gateway { log, seq: 0, restarted: false, + registry: Arc::new(Registry::new()), + gateway_rx, + open: HashMap::new(), } } @@ -253,6 +285,7 @@ impl Gateway { for turn in self.state.take_in_flight()? { self.post(&turn.channel, &turn.root, INTERRUPTED); } + self.mark_left_behind()?; } Ok(()) } @@ -266,6 +299,7 @@ impl Gateway { return Err(Stop::Asked); } self.finished()?; + self.gateway_requests()?; if last_typing.elapsed() >= typing_every { last_typing = Instant::now(); if let Err(e) = self.typing(ws) { diff --git a/crates/gatewayd/tests/serve_approvals.rs b/crates/gatewayd/tests/serve_approvals.rs new file mode 100644 index 0000000..277d9a5 --- /dev/null +++ b/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); +} diff --git a/crates/gatewayd/tests/support/brokerd_client.rs b/crates/gatewayd/tests/support/brokerd_client.rs new file mode 100644 index 0000000..0d33661 --- /dev/null +++ b/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, +} + +/// 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 { + 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 { + 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); + } +} diff --git a/crates/gatewayd/tests/support/fake_mm.rs b/crates/gatewayd/tests/support/fake_mm.rs index 48cba6b..879ea90 100644 --- a/crates/gatewayd/tests/support/fake_mm.rs +++ b/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, ws_tx: &mpsc::Sender) { @@ -298,6 +324,9 @@ fn rest(inner: &Mutex, 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, 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::(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"})), } } diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 6b01446..d571baf 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone. | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model | |---|---|---|---|---|---|---|---| +| M4b/14-gatewayd-serve-approvals | 2026-09-24 | done | 1 | pass | none | Copied the skeleton `approvals.rs`, `serve_approvals.rs`, the new `support/brokerd_client.rs`, and a new `support/fake_mm.rs` (which gains `edits`/`wait_edits`, `username` via `GET /api/v4/users/{user}`, and `refuse_posts` returning 400 on `POST /api/v4/posts`). In `serve/mod.rs`: added `mod approvals;` before `mod handle;` and the `HashMap`, `Registry` and `gateway::{self, ToLoop}` imports; the three `registry`/`gateway_rx`/`open` fields on `Gateway`; `Gateway::new` gained a `gateway_rx: Receiver` last parameter and set `registry`, `gateway_rx`, `open`; `run` now `listen`s `gateway.sock` and `serve`s it on its own thread before `Client::new`, returning `Stop::Start` on either failure and passing `gateway_rx` to `Gateway::new`; `connected` calls `self.mark_left_behind()?` after the `take_in_flight` loop; `event_loop` calls `self.gateway_requests()?` after `self.finished()?`. Filled the four `todo!()`s in `approvals.rs`: `place` (`turn_of` the session's in-flight turn, else the first approver's `direct_channel` as a new empty thread, each error string as given); `show` was already written; `ended` removes from the registry and state, returns when there is no `open`, maps `Ending::NoLonger`/`Answered` to the `Ended` (the `by` run through `username` as `@` or left the id, and `outcome` → `Runs`/`Denied`/`Refused{reason:None}`/`Gone`); `mark_left_behind` edits each left-behind approval's last post with the title and the `Restarted` line; `mark` patches the post and logs a failure. The one clippy warning was the unused `reason` of `Ending::Answered` (the real reason comes from `outcome`), ignored with `reason: _`. All five `serve_approvals` tests pass three times; the whole gatewayd suite is green; `cargo fmt --all` then `make gate` prints `gate: ok` on the first run. | ? | | M4b/13-gatewayd-gateway-sock | 2026-09-24 | done | 1 | pass | none | Copied `docs/plans/M4b/files/crates/gatewayd/src/gateway.rs` to `crates/gatewayd/src/` and `docs/plans/M4b/files/crates/gatewayd/tests/gateway_sock.rs` to `crates/gatewayd/tests/`, then added `pub mod gateway;` to `lib.rs` after `pub mod deliver;`. Filled the four `todo!()`s the skeleton left in `gateway.rs`: `end` (send `ToLoop::Ended { approval, ending }` to the loop, the only thing that touches the loop); `frames` (`try_clone` the stream, spawn a reader thread that hands whole frames over an mpsc channel so the connection waits with a timeout without losing half a frame, `Ok(rx)` — the same as brokerd's courier); `listen` (the parent dir made 0700 via a recursive `DirBuilder` then `set_permissions`, a stale socket removed, anything else left in place as "cannot listen on : something else is there", bind, then the socket made 0600, with the spec's error strings and paths via `.display()`); and `serve` (spawn an accept loop that skips `incoming()` errors and starts a `connection` thread per stream, logging a spawn failure as "gatewayd: cannot start a thread for an approval: ", keeping a separate `log` clone for that message). The skeleton's `ToLoop` was already complete with `Ended`, so it needed no change. All six `gateway_sock` tests pass three times running in ~1 s; `make gate` prints `gate: ok` on the first run. | ? | | M4b/12-gatewayd-state-approvals | 2026-09-24 | done | 1 | pass | none | Copied `docs/plans/M4b/files/crates/gatewayd/tests/state_approvals.rs` into `crates/gatewayd/tests/`. In `state.rs`: added the `#[serde(default)] approvals: Vec` field after `in_flight` (optional so an M4a state file loads unchanged) and the `ApprovalPosts { approval, channel, posts }` struct (`#[serde(deny_unknown_fields)`) after `struct StateFile`; in `problem` the `in_flight` loop now has a loop over `approvals` requiring a `valid_id` channel and at least one `valid_id` post, else "approval {}'s posts are not valid"; added `turn_of` (find the `in_flight` entry of a session, cloned), `add_approval` (replace any entry of the same approval, push, save), `remove_approval` (retain by approval, save only when one was removed) and `take_approvals` (`std::mem::take`, save only when there were some, return them) at the end of `impl State`. `cargo fmt --all` then `cargo test -p gatewayd --test state_approvals --test state --test serve_restart` green (4, 4 and 7 passed); `make gate` prints `gate: ok` on the first run. | ? | | M4b/11-gatewayd-approvals-registry | 2026-09-24 | done | 1 | pass | none | Copied `docs/plans/M4b/files/crates/gatewayd/src/approvals/mod.rs` (the structs `Posted`/`Judged`/`Registry`, the `pub mod text;` still there, and the five `todo!()`s) to `crates/gatewayd/src/approvals/`, and `docs/plans/M4b/files/crates/gatewayd/tests/approvals_judge.rs` to `crates/gatewayd/tests/`. Filled `add` (insert by `approval`), `remove` (remove by id), `holds` (contains_key), `waiting_sessions` (the `session` of every entry), and `judge` (the six spec-section-7 checks in order: `user` in `approvers` else `NOT_APPROVER`, the command to `(id, approved, reason)` — `Approve` true/None, `Deny` false/`reason.clone()`, `Malformed` `USAGE`, then `get(&id)` present else "not pending here", `answerable` else "bxctl approvals", `root` matches else "its own thread", `at <= shown_at` else "answered before shown in full", then `remove` and `send` the `ApprovalReply`, a gone connection → "not pending here"). Brought `Approve`/`Deny`/`Malformed` into scope with the `ApprovalCommand` type, and cloned the `answer` `Sender` out of the entry before the early-return refusals so the held lock's `remove` does not conflict. All six `approvals_judge` and six `approvals_text` tests pass; `make gate` prints `gate: ok` on the first run. | ? |