Add the table of pending approvals

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

crates/brokerd/src/approvals.rs

@@ -0,0 +1,76 @@
+//! The pending-approval table: whoever takes an entry out answers it, and everyone else finds it
+//! gone. In memory only.
+
+use crate::policy::{Ask, Decision};
+use proto::{DenyReason, PendingApproval, Timestamp};
+use std::collections::BTreeMap;
+use std::sync::mpsc::{Receiver, Sender, channel};
+use std::sync::{Mutex, MutexGuard};
+
+/// What the waiting thread is told. Boxed: clippy's `large_enum_variant` rejects it unboxed.
+#[derive(Debug)]
+pub enum Verdict {
+    Run(Box<Decision>),
+    Denied(DenyReason),
+}
+
+#[derive(Debug)]
+pub struct Entry {
+    pub info: PendingApproval,
+    pub ask: Ask,
+    pub reply: Sender<Verdict>,
+}
+
+#[derive(Debug, Default)]
+pub struct Table {
+    entries: Mutex<BTreeMap<u64, Entry>>,
+}
+
+impl Table {
+    pub fn new() -> Table {
+        Table {
+            entries: Mutex::new(BTreeMap::new()),
+        }
+    }
+
+    fn lock(&self) -> MutexGuard<'_, BTreeMap<u64, Entry>> {
+        self.entries
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    pub fn insert(&self, info: PendingApproval, ask: Ask) -> Receiver<Verdict> {
+        let (sender, receiver) = channel();
+        self.lock().insert(
+            info.approval,
+            Entry {
+                info,
+                ask,
+                reply: sender,
+            },
+        );
+        receiver
+    }
+
+    pub fn take(&self, id: u64) -> Option<Entry> {
+        self.lock().remove(&id)
+    }
+
+    pub fn take_expired(&self, now: Timestamp) -> Vec<Entry> {
+        // One lock: taking it once per id would let an approve slip in between.
+        let mut entries = self.lock();
+        let ids: Vec<u64> = entries
+            .values()
+            .filter(|entry| now >= entry.info.expires)
+            .map(|entry| entry.info.approval)
+            .collect();
+        ids.iter().filter_map(|id| entries.remove(id)).collect()
+    }
+
+    pub fn list(&self) -> Vec<PendingApproval> {
+        self.lock()
+            .values()
+            .map(|entry| entry.info.clone())
+            .collect()
+    }
+}

crates/brokerd/src/lib.rs

@@ -1,5 +1,6 @@
 //! The broker: the only role that holds authority.
 
+pub mod approvals;
 pub mod args;
 pub mod audit;
 pub mod config;

crates/brokerd/tests/approvals.rs

@@ -0,0 +1,159 @@
+//! The pending-approval table: whoever takes an entry answers it, and everyone else finds it
+//! gone. Do not edit.
+
+#[path = "support/build.rs"]
+mod build;
+
+use std::sync::{Arc, Barrier};
+
+use brokerd::approvals::{Table, Verdict};
+use brokerd::policy::{Ask, Outcome, SessionState, decide};
+use build::{grant, now, read, set, ts};
+use proto::{CallId, DataClass, DenyReason, Mode, PendingApproval, SessionId, Timestamp};
+
+fn ask() -> Ask {
+    let grants = set(vec![grant("n", "read_file", Mode::Ask).paths(&["/n"])]);
+    match decide(read("/n/a"), &grants, SessionState::default(), now()) {
+        Outcome::Ask(ask) => ask,
+        other => panic!("the test's call does not ask: {other:?}"),
+    }
+}
+
+fn info(approval: u64, expires: &str) -> PendingApproval {
+    PendingApproval {
+        approval,
+        session: SessionId::new("s1").unwrap(),
+        call: CallId(approval + 100),
+        tool: "read_file".to_string(),
+        arguments: r#"{"path":"/n/a"}"#.to_string(),
+        grant: "n".to_string(),
+        taint: DataClass::Private,
+        created: now(),
+        expires: ts(expires),
+    }
+}
+
+const LATER: &str = "2026-09-18T12:15:00.000Z";
+
+#[test]
+fn a_new_table_is_empty_and_lists_in_id_order() {
+    let table = Table::new();
+    assert_eq!(table.list(), []);
+    let _a = table.insert(info(7, LATER), ask());
+    let _b = table.insert(info(3, LATER), ask());
+    assert_eq!(table.list(), [info(3, LATER), info(7, LATER)]);
+}
+
+#[test]
+fn an_entry_can_be_taken_once() {
+    let table = Table::new();
+    let _rx = table.insert(info(5, LATER), ask());
+    let entry = table.take(5).expect("the entry is there");
+    assert_eq!(entry.info, info(5, LATER));
+    assert_eq!(entry.ask.grant(), "n");
+    assert!(table.take(5).is_none(), "taken twice");
+    assert_eq!(table.list(), []);
+}
+
+#[test]
+fn an_id_never_added_is_not_there() {
+    let table = Table::new();
+    let _rx = table.insert(info(5, LATER), ask());
+    assert!(table.take(6).is_none());
+    assert_eq!(table.list().len(), 1);
+}
+
+#[test]
+fn the_verdict_goes_to_the_waiting_side() {
+    let table = Table::new();
+    let rx = table.insert(info(5, LATER), ask());
+    let entry = table.take(5).unwrap();
+    entry
+        .reply
+        .send(Verdict::Denied(DenyReason::ApprovalRefused))
+        .unwrap();
+    match rx.recv().unwrap() {
+        Verdict::Denied(reason) => assert_eq!(reason, DenyReason::ApprovalRefused),
+        Verdict::Run(_) => panic!("the verdict changed on the way"),
+    }
+}
+
+#[test]
+fn an_entry_expires_at_its_expiry_and_not_before() {
+    let table = Table::new();
+    let _a = table.insert(info(9, "2026-09-18T12:00:01.000Z"), ask());
+    let _b = table.insert(info(2, "2026-09-18T12:00:00.500Z"), ask());
+    let _c = table.insert(info(4, LATER), ask());
+
+    let before = ts("2026-09-18T12:00:00.499Z");
+    assert!(table.take_expired(before).is_empty());
+
+    // Exactly at `expires` is expired.
+    let at = ts("2026-09-18T12:00:00.500Z");
+    let due: Vec<u64> = table
+        .take_expired(at)
+        .iter()
+        .map(|e| e.info.approval)
+        .collect();
+    assert_eq!(due, [2]);
+
+    let after = ts("2026-09-18T13:00:00.000Z");
+    let due: Vec<u64> = table
+        .take_expired(after)
+        .iter()
+        .map(|e| e.info.approval)
+        .collect();
+    assert_eq!(due, [4, 9], "in id order");
+    assert_eq!(table.list(), []);
+    assert!(table.take_expired(Timestamp::MAX).is_empty());
+}
+
+#[test]
+fn two_takers_at_once_one_gets_it() {
+    for round in 0..100 {
+        let table = Arc::new(Table::new());
+        let _rx = table.insert(info(1, LATER), ask());
+        let start = Arc::new(Barrier::new(2));
+        let takers: Vec<_> = (0..2)
+            .map(|_| {
+                let table = Arc::clone(&table);
+                let start = Arc::clone(&start);
+                std::thread::spawn(move || {
+                    start.wait();
+                    table.take(1).is_some()
+                })
+            })
+            .collect();
+        let got: Vec<bool> = takers.into_iter().map(|t| t.join().unwrap()).collect();
+        assert_eq!(
+            got.iter().filter(|g| **g).count(),
+            1,
+            "round {round}: {got:?}"
+        );
+    }
+}
+
+#[test]
+fn a_taker_and_the_expiry_at_once_one_gets_it() {
+    for round in 0..100 {
+        let table = Arc::new(Table::new());
+        let _rx = table.insert(info(1, "2026-09-18T12:00:00.000Z"), ask());
+        let start = Arc::new(Barrier::new(2));
+        let t = {
+            let (table, start) = (Arc::clone(&table), Arc::clone(&start));
+            std::thread::spawn(move || {
+                start.wait();
+                usize::from(table.take(1).is_some())
+            })
+        };
+        let e = {
+            let (table, start) = (Arc::clone(&table), Arc::clone(&start));
+            std::thread::spawn(move || {
+                start.wait();
+                table.take_expired(now()).len()
+            })
+        };
+        let total = t.join().unwrap() + e.join().unwrap();
+        assert_eq!(total, 1, "round {round}");
+    }
+}