kyle/boxmaker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gatewayd: approvals asked by non-approvers in direct messages go to an approver

Browse Source 
Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)
This commit is contained in:
kyle
2026-09-24 12:07:06 -07:00
parent 989548419d
commit e10d93c3e8
6 changed files with 338 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/gatewayd/src/serve/approvals.rs
+89 -26
View File
@@ -7,11 +7,30 @@ use std::sync::mpsc::Sender;
use proto::{ApprovalReply, GatewayOutcome, PendingApproval};
use crate::approvals::Posted;
use crate::approvals::text::{self, Ended};
use crate::approvals::text::{self, Asked, Ended};
use crate::gateway::{Ending, ToLoop};
use crate::serve::{Gateway, Stop};
use crate::state::ApprovalPosts;
/// Who asked for the turn now running in a session: the author of the last message queued, the
/// channel's type, and the turn's text. Kept in memory only, and forgotten when the turn ends.
#[derive(Debug, Clone)]
pub(super) struct Asker {
pub(super) user: String,
pub(super) channel_type: String,
pub(super) prompt: String,
}
/// Where an approval goes, and what is posted besides it.
struct Place {
channel: String,
root: String,
/// A post before the approval's own: who asked and what, or whose session.
first: Option<String>,
/// The asking thread, told where the approval went.
note: Option<(String, String)>,
}
/// An approval's last post, the one that is edited when it ends.
#[derive(Debug)]
pub(super) struct Open {
@@ -38,44 +57,85 @@ impl Gateway {
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()),
/// Where an approval goes (M4b spec, section 6): its turn's thread; or, when a non-approver
/// asked in a direct or group message, the first approver's direct messages with who asked
/// and what, and a note in the asking thread; or, for a session not followed here, the first
/// approver's direct messages with a line saying so.
fn place(&self, info: &PendingApproval) -> Result<Place, String> {
let id = info.approval;
let session = info.session.as_str();
let Some(turn) = self.state.turn_of(session) else {
return Ok(Place {
channel: self.approver_dm()?,
root: String::new(),
first: Some(text::unfollowed_post(id, session)),
note: None,
});
};
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}")),
}
let redirect = self.askers.get(session).filter(|a| {
matches!(a.channel_type.as_str(), "D" | "G")
&& !self.config.approvers().contains(&a.user)
});
let Some(asker) = redirect else {
return Ok(Place {
channel: turn.channel,
root: turn.root,
first: None,
note: None,
});
};
let by = match self.client.username(&asker.user) {
Ok(name) => format!("@{name}"),
Err(_) => asker.user.clone(),
};
let asked = Asked {
by,
channel_type: asker.channel_type.clone(),
prompt: asker.prompt.clone(),
};
Ok(Place {
channel: self.approver_dm()?,
root: String::new(),
first: Some(text::asked_post(id, &asked)),
note: Some((turn.channel, turn.root)),
})
}
/// 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).
/// The first approver's direct channel with the bot.
fn approver_dm(&self) -> Result<String, String> {
let first = self
.config
.first_approver()
.ok_or("no approver is configured")?;
self.client
.direct_channel(&self.me.id, first)
.map_err(|e| format!("no direct channel with the first approver: {e}"))
}
/// Post an approval's first post, if any, and its 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<ApprovalReply>,
) -> Result<Result<Vec<String>, String>, Stop> {
let id = info.approval;
let (channel, mut root) = match self.place(info) {
let place = match self.place(info) {
Ok(place) => place,
Err(why) => return Ok(Err(why)),
};
let (channel, mut root) = (place.channel, place.root);
let shown = text::render(info);
// A first post saying who asked, or whose session, then the approval's own posts.
let bodies: Vec<String> = place
.first
.into_iter()
.chain(shown.posts.iter().cloned())
.collect();
let mut posts = Vec::new();
let mut shown_at = 0;
for body in &shown.posts {
for body in &bodies {
match self.client.create_post(&channel, &root, body) {
Ok(post) => {
if root.is_empty() {
@@ -87,7 +147,7 @@ impl Gateway {
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)))
(posts.last(), bodies.get(posts.len().wrapping_sub(1)))
{
self.mark(id, last, &text::with_ending(text, &Ended::NoLonger));
}
@@ -109,7 +169,7 @@ impl Gateway {
answerable: shown.answerable,
answer,
});
if let (Some(last_post), Some(last_text)) = (posts.last(), shown.posts.last()) {
if let (Some(last_post), Some(last_text)) = (posts.last(), bodies.last()) {
self.open.insert(
id,
Open {
@@ -118,6 +178,9 @@ impl Gateway {
},
);
}
if let Some((note_channel, note_root)) = place.note {
self.post(&note_channel, &note_root, &text::sent_to_approver(id));
}
Ok(Ok(posts))
}
crates/gatewayd/src/serve/handle.rs
+22 -7
View File
@@ -7,6 +7,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
use crate::approvals::Judged;
use crate::deliver::{LOOP_DOWN, Notice, deliver};
use crate::mm::{Post, typing};
use crate::serve::approvals::Asker;
use crate::serve::{Gateway, Stop};
use crate::sessions::{BUSY, Batch, Ignored, Pushed, Route};
use crate::state::InFlight;
@@ -103,6 +104,10 @@ impl Gateway {
}
}
Route::Queue(message) => {
self.authors.insert(
message.session.clone(),
(post.user_id.clone(), channel_type.to_string()),
);
if message.joins_thread {
self.state.join_thread(&message.thread.root)?;
}
@@ -136,6 +141,15 @@ impl Gateway {
channel: batch.thread.channel.clone(),
root: batch.thread.root.clone(),
})?;
if let Some((user, channel_type)) = self.authors.get(&batch.session) {
let asker = Asker {
user: user.clone(),
channel_type: channel_type.clone(),
prompt: batch.text.clone(),
};
self.askers
.insert(batch.session.as_str().to_string(), asker);
}
let (registry, wait) = (Arc::clone(&self.registry), self.notice);
let (client, socket, log, done) = (
Arc::clone(&self.client),
@@ -164,16 +178,17 @@ impl Gateway {
Ok(())
}
/// Turns that ended: out of flight, and the next batch of each session started.
/// Turns that ended: out of flight, who asked forgotten, and the next batch of each session
/// started. Never blocks.
pub(super) fn finished(&mut self) -> Result<(), Stop> {
// `while let Ok(session) = self.done_rx.try_recv() {` (never blocks)
// `self.state.end_turn(session.as_str())?;`
// `if let Some(batch) = self.queues.finish(&session) { self.start(batch)?; }` `}`
// Then Ok(()).
while let Ok(session) = self.done_rx.try_recv() {
self.state.end_turn(session.as_str())?;
if let Some(batch) = self.queues.finish(&session) {
self.start(batch)?;
self.askers.remove(session.as_str());
match self.queues.finish(&session) {
Some(batch) => self.start(batch)?,
None => {
self.authors.remove(&session);
}
}
}
Ok(())
crates/gatewayd/src/serve/mod.rs
+6
View File
@@ -115,6 +115,10 @@ pub(crate) struct Gateway {
gateway_rx: Receiver<ToLoop>,
/// The last post of each approval posted here and not yet ended.
open: HashMap<u64, approvals::Open>,
/// The author and channel type of the last message queued in each session.
authors: HashMap<SessionId, (String, String)>,
/// Who asked for each turn running, by session id.
askers: HashMap<String, approvals::Asker>,
/// How long a turn waits for its approval to come through `gateway.sock` before the notice.
notice: Duration,
}
@@ -264,6 +268,8 @@ impl Gateway {
registry: Arc::new(Registry::new()),
gateway_rx,
open: HashMap::new(),
authors: HashMap::new(),
askers: HashMap::new(),
notice,
}
}
crates/gatewayd/tests/serve_approvals.rs
+22 -9
View File
@@ -112,28 +112,41 @@ fn without_a_turn_in_flight_it_goes_to_the_first_approvers_direct_messages_in_pa
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.len(), 3, "a first post, then 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]
posts[0]
.2
.starts_with("**Approval 7** was asked in session `chat-1789700000-1`"),
"{}",
posts[0].2
);
for (n, part) in posts[1..].iter().enumerate() {
assert_eq!(
(part.0.as_str(), part.1.clone()),
(DM, FakeMm::post_id(0)),
"part {n} in its thread"
);
}
assert!(posts[1].2.starts_with("**Approval 7**"), "{}", posts[1].2);
assert!(
posts[2]
.2
.contains("Reply in this thread with `!approve 7`")
);
assert_eq!(asking.posts, [FakeMm::post_id(0), FakeMm::post_id(1)]);
assert_eq!(
asking.posts,
[FakeMm::post_id(0), FakeMm::post_id(1), FakeMm::post_id(2)]
);
asking.close();
let edits = mm.wait_edits(1, WAIT);
assert_eq!(
edits[0].0,
FakeMm::post_id(1),
FakeMm::post_id(2),
"only the last part is edited"
);
}
crates/gatewayd/tests/serve_redirect.rs
+198
View File
@@ -0,0 +1,198 @@
//! An approval asked for by someone who may talk but not approve, in a direct or group message,
//! goes to the first approver's direct messages: first who asked and what, then the approval, and
//! the asking thread is told where it went. In a channel it stays in its thread (M4b spec,
//! section 6). 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::sync::{Mutex, mpsc};
use std::time::Duration;
use brokerd_client::{Asking, ask, pending};
use fake_loop::{done, serve_loop};
use fake_mm::{DM, EVE, EVE_DM, FakeMm, KYLE, WsPeer, id, post};
use gateway::{Running, WAIT, loop_dir, start};
use gatewayd::config::Config;
use proto::{GatewayOutcome, Turn};
use tmp::TempDir;
const GROUP: &str = "g0000000000000000000000000";
const CHANNEL: &str = "o0000000000000000000000000";
/// KYLE and EVE may talk, in direct messages, GROUP and CHANNEL; only KYLE approves.
fn config(home: &TempDir, url: &str) -> Config {
let text = format!(
"[mattermost]\nurl = \"{url}\"\n[secrets.mattermost_token]\nenv = \"NOT_READ_BY_RUN\"\n[allow]\nusers = [\"{KYLE}\", \"{EVE}\"]\nchannels = [\"{GROUP}\", \"{CHANNEL}\"]\n[approvals]\nusers = [\"{KYLE}\"]\n[paths]\nhome = \"{}\"\n[limits]\ntyping_every_ms = 100\n",
home.path().display()
);
Config::parse(&text).unwrap()
}
/// A loop whose turns wait for `go`, then answer "done".
fn held(home: &TempDir) -> (mpsc::Receiver<Turn>, mpsc::Sender<()>) {
loop_dir(home);
let (go_tx, go_rx) = mpsc::channel::<()>();
let go = Mutex::new(go_rx);
let turns = serve_loop(&home.path().join("run/loop/loop.sock"), move |_, _| {
let _ = go.lock().unwrap().recv_timeout(Duration::from_secs(20));
vec![done("done")]
});
(turns, go_tx)
}
struct Stage {
_home: TempDir,
mm: FakeMm,
_running: Running,
ws: WsPeer,
go: mpsc::Sender<()>,
asking: Asking,
root: String,
}
/// gatewayd up, EVE's message `text` in `channel` (of `channel_type`), its turn held, and
/// approval 42 asked for its session.
fn stage(tag: &str, channel: &str, channel_type: &str, text: &str) -> Stage {
let home = TempDir::new(tag);
let (turns, go) = held(&home);
let mm = FakeMm::start();
let running = start(config(&home, &mm.url()));
let mut ws = mm.next_ws(WAIT);
running.wait_log("gatewayd: connected to ");
let root = id('p', 1);
ws.posted(&post(&root, EVE, channel, "", text, 5), channel_type);
turns.recv_timeout(WAIT).expect("the turn started");
let socket = home.path().join("run/gateway/gateway.sock");
let session = format!("mm-{root}");
let asking = ask(&socket, pending(42, &session, r#"{"path":"/n/a"}"#)).unwrap();
Stage {
_home: home,
mm,
_running: running,
ws,
go,
asking,
root,
}
}
/// EVE's posts in `channel`, redirected: who asked, the approval, and the note, in that order.
fn redirected(s: &Stage, channel: &str, place: &str) {
let posts = s.mm.wait_posts(3, WAIT);
assert_eq!(posts.len(), 3, "{posts:?}");
assert_eq!(
(posts[0].0.as_str(), posts[0].1.as_str()),
(DM, ""),
"a new thread in the approver's direct messages"
);
assert!(
posts[0].2.starts_with(&format!(
"**Approval 42** was asked by @user-e000 in {place} with the bot. Their message:\n"
)),
"{}",
posts[0].2
);
assert_eq!(
(posts[1].0.as_str(), posts[1].1.clone()),
(DM, FakeMm::post_id(0)),
"the approval in that thread"
);
assert!(posts[1].2.starts_with("**Approval 42**"), "{}", posts[1].2);
assert!(
posts[1]
.2
.contains("Reply in this thread with `!approve 42`")
);
assert_eq!(
posts[2],
(
channel.to_string(),
s.root.clone(),
"waiting for approval 42: it was sent to an approver".to_string()
),
"the note, in the asking thread"
);
assert_eq!(s.asking.posts, [FakeMm::post_id(0), FakeMm::post_id(1)]);
}
#[test]
fn a_direct_message_from_a_non_approver_goes_to_the_approver_who_answers_it_there() {
let mut s = stage("rd-dm", EVE_DM, "D", "read my notes");
redirected(&s, EVE_DM, "a direct message");
assert!(
s.mm.posts()[0].2.contains("\n```\nread my notes\n```"),
"the prompt, fenced: {}",
s.mm.posts()[0].2
);
let thread = FakeMm::post_id(0);
s.ws.posted(&post(&id('c', 1), KYLE, DM, &thread, "!approve 42", 6), "D");
let reply = s.asking.answer(WAIT).expect("the answer reaches brokerd");
assert_eq!(
(reply.approval, reply.approved, reply.by.as_str()),
(42, true, KYLE)
);
s.asking.outcome(GatewayOutcome::Runs {});
let edits = s.mm.wait_edits(1, WAIT);
assert_eq!(
edits[0].0,
FakeMm::post_id(1),
"the approval's post is marked"
);
let approval_text = s.mm.posts()[1].2.clone();
assert!(
edits[0].1.starts_with(&approval_text) && edits[0].1.len() > approval_text.len(),
"its own text, with the ending: {}",
edits[0].1
);
s.go.send(()).unwrap();
}
#[test]
fn a_long_prompt_is_cut() {
let long = "y".repeat(2_500);
let s = stage("rd-long", EVE_DM, "D", &long);
redirected(&s, EVE_DM, "a direct message");
let first = &s.mm.posts()[0].2;
assert!(first.contains(&"y".repeat(2_000)), "{first}");
assert!(!first.contains(&"y".repeat(2_001)), "{first}");
assert!(first.ends_with("\n(… 500 more characters)"), "{first}");
s.go.send(()).unwrap();
}
#[test]
fn a_group_message_goes_to_the_approver_with_the_mention_stripped() {
let s = stage("rd-group", GROUP, "G", "@boxmaker-straylight read my notes");
redirected(&s, GROUP, "a group message");
let first = &s.mm.posts()[0].2;
assert!(first.contains("\n```\nread my notes\n```"), "{first}");
s.go.send(()).unwrap();
}
#[test]
fn in_a_channel_it_stays_in_the_thread() {
let s = stage(
"rd-channel",
CHANNEL,
"O",
"@boxmaker-straylight read my notes",
);
s.mm.wait_posts(1, WAIT);
std::thread::sleep(Duration::from_millis(300));
let posts = s.mm.posts();
assert_eq!(posts.len(), 1, "{posts:?}");
assert_eq!(
(posts[0].0.as_str(), posts[0].1.as_str()),
(CHANNEL, s.root.as_str())
);
assert!(posts[0].2.starts_with("**Approval 42**"), "{}", posts[0].2);
s.go.send(()).unwrap();
}