roundabout

public

Star0

Fork 0

Watch0

8 commits 1 branch 0 tags 0 releases158.8 KiB

Commit `ecc1cf8df0f5`

tarball

atagen <boss@atagen.co>2026-09-27 13:43 UTC

9 files changed+317-18

daemon: hand idle cores to builds, minus what others expect to need

Grants matched the ask exactly, so a lone m build sat on 4 of 12 cores. Spare cores are cheap to give since cpu oversubscription only slows things down, but they only go spare after subtracting cores other sessions have claimed with rabt expect, and each bonus core is charged ram at the build's own per-core rate because more jobs mean more memory. Intents live in the daemon and keep it running until they lapse, and SessionEnd withdraws them early.

Modified file: README.md+6 additions−1 deletions@@ -13,6 +13,7 @@ rabt run --size m -- cargo build
 rabt run --ram 6G --cores 8 --time 10m -- nix build .#thing
 rabt status
 rabt suggest -- cargo build --release
+rabt expect --size l   # building soon, save me a seat
 rabt watch 7        # exits when lease 7 is done or overdue
 rabt kill 7
 rabt extend 7 10m
@@ -21,7 +22,9 @@ rabt artifacts check | list | gc [--cap] | pin <path>
 
 `run` exits with the build's code, 124 if `--deadline` killed it, 75 if `--max-wait` ran out.
 
-a granted build gets `CARGO_BUILD_JOBS`, `MAKEFLAGS`, `CMAKE_BUILD_PARALLEL_LEVEL` and a `NIX_CONFIG` with `cores`/`max-jobs` to match its grant, so it actually uses the cores it asked for and no more.
+a granted build gets `CARGO_BUILD_JOBS`, `MAKEFLAGS`, `CMAKE_BUILD_PARALLEL_LEVEL` and a `NIX_CONFIG` with `cores`/`max-jobs` to match its grant, so it actually uses the cores it got and no more.
+
+the grant can be bigger than the ask. idle cores go to whoever's building, minus what other sessions have said they'll want via `rabt expect`. spare cores never hold up admission, oversubscribing cpu only slows things down, but each one is charged ram at the build's own ram per core since more jobs eat more memory. a session's build counts as an `expect` too, so agents that forget still leave a mark. `--exact` opts out.
 
 ## claude code
 
@@ -71,6 +74,8 @@ time = "5m"
 linger = "5s"
 backfill_max_time = "2m"
 # max_wait = "30m"
+elastic = true         # hand idle cores to whoever's building
+intent_ttl = "20m"     # how long a `rabt expect` or a build keeps its claim
 
 [overrun]
 ram = "warn"           # or "throttle", systemd-run MemoryHigh on linux

Modified file: nix/hooks.nix+1

@@ -17,4 +17,5 @@ in

1717 UserPromptSubmit = [ { hooks = [ (hook "prompt-submit") ]; } ];

1818 Stop = [ { hooks = [ (hook "stop") ]; } ];

1919 SubagentStop = [ { hooks = [ (hook "subagent-stop") ]; } ];

20+ SessionEnd = [ { hooks = [ (hook "session-end") ]; } ];

2021 }

Modified file: plugin/skills/roundabout/SKILL.md+12 additions@@ -7,6 +7,16 @@ description: use before running any build, test suite or other heavy command (ca
 
 this box is shared with other agents and a human who'd like their video to keep playing. builds queue through roundabout, whose cli is `rabt`, and only run when there's room.
 
+## first, say you're coming
+
+when you pick up a task that'll involve building, before you start editing
+
+```
+rabt expect --size m
+```
+
+sized like the biggest build you expect to run. idle cores get handed out to whoever's building, and this is how the others know to leave you some. it lapses 20 minutes after your last build or `expect`, so one call covers an edit-compile loop. `rabt expect --done` if the plan changes.
+
 ## running something
 
 ```
@@ -29,6 +39,8 @@ sizes out of the box
 
 guess honestly. under-asking is how the box ooms, and an honest `--time` is what lets small jobs slip past big ones. it tells you afterwards if you were off.
 
+you may get more cores than you asked for when the box is quiet, `CARGO_BUILD_JOBS` and friends are set to match. `--exact` turns that off, for benchmarks and anything else that wants a fixed number.
+
 ## waiting
 
 a queued build prints its place and why to stderr. bash moves slow commands to the background on its own, but for anything likely over a couple of minutes

Modified file: src/cli.rs+82 additions−2 deletions@@ -38,9 +38,26 @@ pub enum Rabt {
         max_wait: Option<Dur>,
         #[pound(long, hidden, env = "CLAUDE_CODE_SESSION_ID")]
         session: Option<String>,
+        /// exactly the cores asked for, no spares
+        #[pound(long)]
+        exact: bool,
         #[pound(trailing)]
         cmd: Vec<String>,
     },
+    /// say you'll be building soon, so others leave you room
+    Expect {
+        #[pound(short, long)]
+        size: Option<String>,
+        #[pound(long)]
+        ram: Option<Bytes>,
+        #[pound(long)]
+        cores: Option<Amount<u32>>,
+        /// changed your mind
+        #[pound(long)]
+        done: bool,
+        #[pound(long, hidden, env = "CLAUDE_CODE_SESSION_ID")]
+        session: Option<String>,
+    },
     /// who's running, who's waiting
     Status {
         #[pound(long)]
@@ -64,7 +81,7 @@ pub enum Rabt {
     },
     /// claude code hook entrypoint, json on stdin
     Hook {
-        /// pre-tool-use post-tool-use prompt-submit stop subagent-stop
+        /// pre-tool-use post-tool-use prompt-submit stop subagent-stop session-end
         event: String,
     },
     #[pound(hidden)]
@@ -198,6 +215,36 @@ fn reply(ev: io::Result<Option<Event>>) -> i32 {
     }
 }
 
+fn expect(cfg: &Config, session: Option<String>, done: bool, a: &run::Args) -> i32 {
+    let Some(session) = session.filter(|s| !s.is_empty()) else {
+        eprintln!("rabt: no CLAUDE_CODE_SESSION_ID, so nobody to expect on behalf of");
+        return 2;
+    };
+    let need = if done {
+        None
+    } else {
+        match run::need(cfg, a) {
+            Ok(n) => Some(n),
+            Err(e) => {
+                eprintln!("rabt: {e}");
+                return 2;
+            }
+        }
+    };
+    let mut s = match client::connect_or_spawn() {
+        Ok(s) => s,
+        Err(e) => {
+            eprintln!("rabt: can't reach the daemon: {e}");
+            return 1;
+        }
+    };
+    if let Err(e) = proto::send(&mut s, &Request::Expect { session, need }) {
+        eprintln!("rabt: {e}");
+        return 1;
+    }
+    reply(Reader::new(&s).and_then(|mut r| r.recv()))
+}
+
 fn status(json: bool) -> i32 {
     let Ok(Some(Event::Status(st))) = ask(&Request::Status) else {
         if json {
@@ -229,6 +276,11 @@ fn status(json: bool) -> i32 {
                 l.waiting.map(|r| r.to_string()).unwrap_or_default()
             )
         };
+        let cores = if l.granted > l.need.cores {
+            format!("{}+{}", l.need.cores, l.granted - l.need.cores)
+        } else {
+            l.need.cores.to_string()
+        };
         println!(
             "{:>4} {}{} {} ram {} cores {} rss {}  {}",
             l.id,
@@ -236,11 +288,20 @@ fn status(json: bool) -> i32 {
             state,
             Dur(l.need.time),
             l.need.ram,
-            l.need.cores,
+            cores,
             l.rss,
             l.cmd.join(" ")
         );
     }
+    for i in &st.intents {
+        println!(
+            "     expect {} ram {} cores {}, {} left",
+            &i.session[..i.session.len().min(8)],
+            i.need.ram,
+            i.need.cores,
+            Dur::secs(i.expires_in)
+        );
+    }
     0
 }
 
@@ -299,6 +360,7 @@ pub fn dispatch(cfg: &Config, cmd: Rabt) -> i32 {
             deadline,
             max_wait,
             session,
+            exact,
             cmd,
         } => run::run(
             cfg,
@@ -311,9 +373,27 @@ pub fn dispatch(cfg: &Config, cmd: Rabt) -> i32 {
                 deadline,
                 max_wait,
                 session,
+                exact,
                 cmd,
             },
         ),
+        Rabt::Expect {
+            size,
+            ram,
+            cores,
+            done,
+            session,
+        } => expect(
+            cfg,
+            session,
+            done,
+            &run::Args {
+                size,
+                ram,
+                cores,
+                ..run::Args::default()
+            },
+        ),
         Rabt::Status { json } => status(json),
         Rabt::Suggest { cmd } => suggest(cfg, &cmd),
         Rabt::Kill { lease } => reply(ask(&Request::Kill { lease })),

Modified file: src/config.rs+4 additions@@ -58,6 +58,8 @@ pub struct Queue {
     pub linger: Dur,
     pub backfill_max_time: Dur,
     pub max_wait: Option<Dur>,
+    pub elastic: bool,
+    pub intent_ttl: Dur,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
@@ -212,6 +214,8 @@ impl Default for Queue {
             linger: Dur::secs(5),
             backfill_max_time: Dur::secs(120),
             max_wait: None,
+            elastic: true,
+            intent_ttl: Dur::secs(1200),
         }
     }
 }

Modified file: src/daemon.rs+138 additions−6 deletions@@ -19,7 +19,7 @@ use crate::{
     client,
     config::Config,
     paths,
-    proto::{self, Acquire, Event, LeaseId, LeaseInfo, Reader, Request, Status},
+    proto::{self, Acquire, Event, IntentInfo, LeaseId, LeaseInfo, Reader, Request, Status},
     sched::{self, Capacity, Disk, FsId, Need, Queued, Reason, Running, Verdict},
     sys,
     units::Bytes,
@@ -29,6 +29,14 @@ struct Run {
     started: Instant,
     rss: Bytes,
     reminded: Option<Instant>,
+    /// asked-for plus bonus. only the asked-for part counts for admission,
+    /// oversubscribed cores just make everyone slower
+    cores: u32,
+}
+
+struct Intent {
+    need: Need,
+    until: Instant,
 }
 
 struct Lease {
@@ -60,6 +68,7 @@ struct State {
     cfg: Config,
     next: LeaseId,
     leases: BTreeMap<LeaseId, Lease>,
+    intents: BTreeMap<String, Intent>,
     cap: Capacity,
     baselines: BTreeMap<FsId, Bytes>,
     conns: usize,
@@ -80,6 +89,7 @@ impl State {
             cfg,
             next: 1,
             leases: BTreeMap::new(),
+            intents: BTreeMap::new(),
             cap: Capacity {
                 ram_budget: Bytes::ZERO,
                 ram_live: Bytes::ZERO,
@@ -242,26 +252,34 @@ impl State {
         let cores = self.cap.cores;
         let mut position = 0;
         for (id, verdict) in verdicts {
-            let Some(lease) = self.leases.get_mut(&id) else {
-                continue;
-            };
             match verdict {
                 Verdict::Admit => {
+                    let (extra, extra_ram) = self.bonus(id);
+                    let Some(lease) = self.leases.get_mut(&id) else {
+                        continue;
+                    };
+                    lease.req.need.ram += extra_ram;
+                    let granted = lease.req.need.cores.clamp(1, cores.max(1)) + extra;
                     lease.run = Some(Run {
                         started: Instant::now(),
                         rss: Bytes::ZERO,
                         reminded: None,
+                        cores: granted,
                     });
                     let _ = proto::send(
                         &mut lease.out,
                         &Event::Granted {
                             lease: id,
-                            cores: lease.req.need.cores.clamp(1, cores.max(1)),
+                            cores: granted,
+                            ram: lease.req.need.ram,
                             heavy: lease.heavy,
                         },
                     );
                 }
                 Verdict::Wait(reason) => {
+                    let Some(lease) = self.leases.get_mut(&id) else {
+                        continue;
+                    };
                     position += 1;
                     lease.waiting = Some(reason);
                     if lease.last_pos != position {
@@ -280,8 +298,91 @@ impl State {
         }
     }
 
+    fn held_by(&self, session: &str) -> (u32, Bytes) {
+        self.leases
+            .values()
+            .filter(|l| l.req.session.as_deref() == Some(session))
+            .fold((0, Bytes::ZERO), |(c, r), l| {
+                (
+                    c + l.run.as_ref().map_or(l.req.need.cores, |run| run.cores),
+                    r + l.req.need.ram,
+                )
+            })
+    }
+
+    /// idle cores the lease can have on top of its ask, minus whatever other
+    /// sessions said they'll want and don't hold yet. capped by ram at the
+    /// lease's own ram per core, since every extra job eats its share
+    fn bonus(&self, id: LeaseId) -> (u32, Bytes) {
+        let Some(lease) = self.leases.get(&id) else {
+            return (0, Bytes::ZERO);
+        };
+        if lease.req.exact || !self.cfg.queue.elastic {
+            return (0, Bytes::ZERO);
+        }
+        let need = lease.req.need;
+        let (taken, queued_ram) = self.leases.iter().filter(|(k, _)| **k != id).fold(
+            (need.cores, need.ram),
+            |(c, r), (_, l)| match &l.run {
+                Some(run) => (c + run.cores, r),
+                None => (c + l.req.need.cores, r + l.req.need.ram),
+            },
+        );
+        let mut spare = self.cap.cores.saturating_sub(taken);
+        let mut ram_spare = sched::headroom(&self.running(), &self.cap)
+            .0
+            .saturating_sub(queued_ram);
+        for (session, intent) in &self.intents {
+            if lease.req.session.as_deref() == Some(session) {
+                continue;
+            }
+            let (cores, ram) = self.held_by(session);
+            spare = spare.saturating_sub(intent.need.cores.saturating_sub(cores));
+            ram_spare = ram_spare.saturating_sub(intent.need.ram.saturating_sub(ram));
+        }
+        let per_core = need.ram.0 / u64::from(need.cores.max(1));
+        let by_ram = ram_spare
+            .0
+            .checked_div(per_core)
+            .map_or(spare, |n| n as u32);
+        let extra = spare.min(by_ram);
+        (extra, Bytes(per_core * u64::from(extra)))
+    }
+
+    /// explicit intents set the size, builds only keep one alive or start a
+    /// default one sized like themselves
+    fn intend(&mut self, session: &str, need: Need, explicit: bool) {
+        let until = Instant::now() + self.cfg.queue.intent_ttl.0;
+        match self.intents.get_mut(session) {
+            Some(i) => {
+                i.until = until;
+                if explicit {
+                    i.need = need;
+                }
+            }
+            None => {
+                self.intents
+                    .insert(session.to_owned(), Intent { need, until });
+            }
+        }
+    }
+
+    fn expire_intents(&mut self) {
+        let had = !self.intents.is_empty();
+        let now = Instant::now();
+        self.intents.retain(|_, i| i.until > now);
+        if had && self.intents.is_empty() && self.leases.is_empty() {
+            self.idle_since = now;
+        }
+    }
+
     fn remove(&mut self, id: LeaseId) {
         if let Some(lease) = self.leases.remove(&id) {
+            if let Some(s) = &lease.req.session
+                && let Some(i) = self.intents.get_mut(s)
+            {
+                i.until = Instant::now() + self.cfg.queue.intent_ttl.0;
+            }
             for mut w in lease.watchers {
                 let _ = proto::send(&mut w, &Event::Fired { overdue: false });
             }
@@ -321,6 +422,7 @@ impl State {
                     need: l.req.need,
                     heavy: l.heavy,
                     running: l.run.is_some(),
+                    granted: l.run.as_ref().map_or(0, |r| r.cores),
                     elapsed: l.elapsed().as_secs(),
                     rss: l.run.as_ref().map_or(Bytes::ZERO, |r| r.rss),
                     overdue: l.overdue(grace),
@@ -333,8 +435,18 @@ impl State {
 
     fn status(&self) -> Status {
         let (ram_free, cores_free) = sched::headroom(&self.running(), &self.cap);
+        let now = Instant::now();
         Status {
             leases: self.info(),
+            intents: self
+                .intents
+                .iter()
+                .map(|(session, i)| IntentInfo {
+                    session: session.clone(),
+                    need: i.need,
+                    expires_in: i.until.saturating_duration_since(now).as_secs(),
+                })
+                .collect(),
             ram_free,
             ram_budget: self.cap.ram_budget,
             cores_free,
@@ -356,6 +468,9 @@ fn serve(shared: &Shared, mut stream: UnixStream) -> io::Result<()> {
                 st.next += 1;
                 let fs = st.fs_for(&req.cwd);
                 let heavy = st.heavy(&req.need);
+                if let Some(s) = req.session.clone() {
+                    st.intend(&s, req.need, false);
+                }
                 st.leases.insert(
                     id,
                     Lease {
@@ -377,6 +492,19 @@ fn serve(shared: &Shared, mut stream: UnixStream) -> io::Result<()> {
             lock(shared).remove(id);
             Ok(())
         }
+        Request::Expect { session, need } => {
+            let mut st = lock(shared);
+            if let Some(need) = need {
+                st.intend(&session, need, true);
+            } else {
+                st.intents.remove(&session);
+                if st.intents.is_empty() && st.leases.is_empty() {
+                    st.idle_since = Instant::now();
+                }
+            }
+            drop(st);
+            proto::send(&mut stream, &Event::Ok)
+        }
         Request::Status => {
             let status = lock(shared).status();
             proto::send(&mut stream, &Event::Status(status))
@@ -493,7 +621,11 @@ fn tick(shared: &Shared) {
         st.refresh();
         st.reschedule();
         st.fire_overdue();
-        if st.conns == 0 && st.leases.is_empty() && st.idle_since.elapsed() >= st.cfg.queue.linger.0
+        st.expire_intents();
+        if st.conns == 0
+            && st.leases.is_empty()
+            && st.intents.is_empty()
+            && st.idle_since.elapsed() >= st.cfg.queue.linger.0
         {
             // systemd owns the socket and queues whoever knocks while we're
             // leaving, so there's nothing to tidy

Modified file: src/hook.rs+20 additions@@ -249,6 +249,25 @@ fn session(input: &Input) -> String {
         .unwrap_or_else(|| input.session_id.clone())
 }
 
+/// both ids, since around /clear the env var and the json can disagree about
+/// which session just ended. never spawns, a missing daemon holds no intents
+fn session_end(input: &Input) {
+    let mut ids = vec![session(input), input.session_id.clone()];
+    ids.dedup();
+    for session in ids.into_iter().filter(|s| !s.is_empty()) {
+        if let Some(mut s) = client::connect() {
+            let _ = proto::send(
+                &mut s,
+                &Request::Expect {
+                    session,
+                    need: None,
+                },
+            );
+            let _ = Reader::new(&s).and_then(|mut r| r.recv::<Event>());
+        }
+    }
+}
+
 fn remind(event: &str, input: &Input) {
     let late = overdue(&session(input));
     if late.is_empty() {
@@ -313,6 +332,7 @@ pub fn main(cfg: &Config, event: &str) -> i32 {
         "prompt-submit" => remind("UserPromptSubmit", &input),
         "stop" => stop(cfg, "Stop", &input),
         "subagent-stop" => stop(cfg, "SubagentStop", &input),
+        "session-end" => session_end(&input),
         other => eprintln!("rabt: unknown hook {other}"),
     }
     0

Modified file: src/proto.rs+38 additions−6 deletions@@ -20,20 +20,40 @@ pub struct Acquire {
     pub cmd: Vec<String>,
     pub session: Option<String>,
     pub deadline: Option<Dur>,
+    #[serde(default)]
+    pub exact: bool,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(tag = "op", rename_all = "snake_case")]
 pub enum Request {
+    /// `None` withdraws it
+    Expect {
+        session: String,
+        need: Option<Need>,
+    },
     Acquire(Acquire),
-    Started { pid: i32 },
-    Usage { rss: Bytes },
+    Started {
+        pid: i32,
+    },
+    Usage {
+        rss: Bytes,
+    },
     Done,
     Status,
-    Kill { lease: LeaseId },
-    Extend { lease: LeaseId, by: Dur },
-    Watch { lease: LeaseId },
-    Overdue { session: String },
+    Kill {
+        lease: LeaseId,
+    },
+    Extend {
+        lease: LeaseId,
+        by: Dur,
+    },
+    Watch {
+        lease: LeaseId,
+    },
+    Overdue {
+        session: String,
+    },
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -45,6 +65,8 @@ pub struct LeaseInfo {
     pub need: Need,
     pub heavy: bool,
     pub running: bool,
+    #[serde(default)]
+    pub granted: u32,
     /// seconds since queued or started, whichever it's doing
     pub elapsed: u64,
     pub rss: Bytes,
@@ -53,9 +75,17 @@ pub struct LeaseInfo {
     pub position: usize,
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct IntentInfo {
+    pub session: String,
+    pub need: Need,
+    pub expires_in: u64,
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Status {
     pub leases: Vec<LeaseInfo>,
+    pub intents: Vec<IntentInfo>,
     pub ram_free: Bytes,
     pub ram_budget: Bytes,
     pub cores_free: u32,
@@ -73,6 +103,8 @@ pub enum Event {
     Granted {
         lease: LeaseId,
         cores: u32,
+        /// grows with any bonus cores, since more jobs means more ram
+        ram: Bytes,
         heavy: bool,
     },
     Kill,

Modified file: src/run.rs+16 additions−3 deletions@@ -24,6 +24,7 @@ use crate::{
 pub const EXIT_TIMEOUT: i32 = 124;
 pub const EXIT_GAVE_UP: i32 = 75;
 
+#[derive(Default)]
 pub struct Args {
     pub size: Option<String>,
     pub ram: Option<Bytes>,
@@ -33,6 +34,7 @@ pub struct Args {
     pub deadline: Option<Dur>,
     pub max_wait: Option<Dur>,
     pub session: Option<String>,
+    pub exact: bool,
     pub cmd: Vec<String>,
 }
 
@@ -49,7 +51,7 @@ fn say(msg: &str) {
     eprintln!("rabt: {msg}");
 }
 
-fn need(cfg: &Config, a: &Args) -> Result<Need, String> {
+pub fn need(cfg: &Config, a: &Args) -> Result<Need, String> {
     let base = cfg.preset(a.size.as_deref().unwrap_or(&cfg.enforce.default_size))?;
     Ok(Need {
         ram: a.ram.unwrap_or(base.ram),
@@ -73,6 +75,7 @@ struct Grant {
     stream: UnixStream,
     lease: u64,
     cores: u32,
+    ram: Bytes,
 }
 
 fn acquire(req: &Acquire, max_wait: Option<Dur>) -> Result<Grant, i32> {
@@ -93,10 +96,17 @@ fn acquire(req: &Acquire, max_wait: Option<Dur>) -> Result<Grant, i32> {
                 Ok(Some(Event::Granted {
                     lease,
                     cores,
+                    ram,
                     heavy,
                 })) => {
+                    let bonus = cores.saturating_sub(req.need.cores);
                     say(&format!(
-                        "lease {lease} go, cores {cores}{}",
+                        "lease {lease} go, cores {cores}{}{}",
+                        if bonus > 0 {
+                            format!(" ({bonus} spare)")
+                        } else {
+                            String::new()
+                        },
                         if heavy { ", heavy" } else { "" }
                     ));
                     let _ = stream.set_read_timeout(None);
@@ -104,6 +114,7 @@ fn acquire(req: &Acquire, max_wait: Option<Dur>) -> Result<Grant, i32> {
                         stream,
                         lease,
                         cores,
+                        ram,
                     });
                 }
                 Ok(Some(Event::Queued {
@@ -230,7 +241,7 @@ pub fn run(cfg: &Config, a: &Args) -> i32 {
         ));
         return 2;
     }
-    let need = match need(cfg, a) {
+    let mut need = match need(cfg, a) {
         Ok(n) => n,
         Err(e) => {
             say(&e);
@@ -258,12 +269,14 @@ pub fn run(cfg: &Config, a: &Args) -> i32 {
         cmd: a.cmd.clone(),
         session: a.session.clone(),
         deadline: a.deadline,
+        exact: a.exact,
     };
 
     let grant = match acquire(&req, a.max_wait.or(cfg.queue.max_wait)) {
         Ok(g) => g,
         Err(code) => return code,
     };
+    need.ram = need.ram.max(grant.ram);
 
     let mut child = match command(cfg, a, &need, &grant).spawn() {
         Ok(c) => c,

roundabout

Commit ecc1cf8df0f5

Commit `ecc1cf8df0f5`