loopd: scheduler yields to a held subagent
Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
This commit is contained in:
@@ -59,6 +59,12 @@ impl Context {
|
||||
drop(busy);
|
||||
Ok(Held { ctx: self, id })
|
||||
}
|
||||
|
||||
/// True when any held session id starts with `prefix` (`sub-` for a live child).
|
||||
pub fn holds_prefix(&self, prefix: &str) -> bool {
|
||||
let busy = self.busy.lock().unwrap_or_else(|p| p.into_inner());
|
||||
busy.iter().any(|id| id.as_str().starts_with(prefix))
|
||||
}
|
||||
}
|
||||
|
||||
/// While a turn runs, the session is in the busy set. The guard holds only an immutable borrow of
|
||||
|
||||
@@ -43,6 +43,9 @@ impl Scheduler {
|
||||
self.leftovers_closed = true;
|
||||
}
|
||||
self.heartbeat(ctx, now, &mut work);
|
||||
if ctx.holds_prefix("sub-") {
|
||||
return work;
|
||||
}
|
||||
|
||||
let jobs_dir = ctx.cfg.paths.home.join("jobs");
|
||||
let runs_path = jobs_dir.join("runs.jsonl");
|
||||
@@ -184,6 +187,9 @@ impl Scheduler {
|
||||
if !due {
|
||||
return;
|
||||
}
|
||||
if ctx.holds_prefix("sub-") {
|
||||
return;
|
||||
}
|
||||
|
||||
let id = match SessionId::new("heartbeat") {
|
||||
Ok(id) => id,
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
//! Heartbeat and cron yield to a held subagent. Do not edit.
|
||||
|
||||
mod support;
|
||||
|
||||
use bxcore::{SessionId, Timestamp};
|
||||
use loopd::channel::Context;
|
||||
use loopd::llama::Client;
|
||||
use loopd::memory::NoMemory;
|
||||
use loopd::scheduler::{Scheduler, TickWork};
|
||||
use loopd::tools::Registry;
|
||||
use std::sync::Arc;
|
||||
use support::{FakeServer, Home, Reply, ScriptedPort};
|
||||
|
||||
const CHAT: &str = "/v1/chat/completions";
|
||||
|
||||
fn sse() -> Reply {
|
||||
let data1 = r#"data: {"choices":[{"finish_reason":null,"index":0,"delta":{"content":"noted"}}],"id":"h","timings":{"cache_n":0,"prompt_n":10,"predicted_n":1}}"#;
|
||||
let data2 = r#"data: {"choices":[{"finish_reason":"stop","index":0,"delta":{}}],"id":"h","timings":{"cache_n":0,"prompt_n":10,"predicted_n":1}}"#;
|
||||
let body = format!("{data1}\n\n{data2}\n\ndata: [DONE]\n\n");
|
||||
Reply::raw(format!(
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
|
||||
body.len()
|
||||
))
|
||||
}
|
||||
|
||||
fn ts(s: &str) -> Timestamp {
|
||||
Timestamp::parse(s).unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_held_subagent_skips_a_due_heartbeat() {
|
||||
let home = Home::new();
|
||||
home.write("heartbeat.md", "Look around.\n");
|
||||
home.write("compact.md", "Summarise.\n");
|
||||
let server = FakeServer::start();
|
||||
let mut cfg = home.config(&server.socket);
|
||||
cfg.heartbeat.every_s = 60;
|
||||
cfg.heartbeat.prompt = home.dir.join("heartbeat.md");
|
||||
server.route(CHAT, vec![sse()]);
|
||||
let ctx = Arc::new(Context::new(
|
||||
cfg.clone(),
|
||||
Client::new(cfg),
|
||||
Box::new(ScriptedPort::new(vec![])),
|
||||
Registry::m2b(),
|
||||
Box::new(NoMemory),
|
||||
));
|
||||
let held = ctx
|
||||
.hold(SessionId::new("sub-mm-thread-1").unwrap())
|
||||
.expect("hold");
|
||||
let mut sched = Scheduler::new(ts("2026-10-03T07:00:00.000Z"));
|
||||
let work = sched.tick(&ctx, ts("2026-10-03T07:01:00.000Z"));
|
||||
assert!(work.is_empty(), "{work:?}");
|
||||
assert!(server.requests_to(CHAT).is_empty());
|
||||
drop(held);
|
||||
let work = sched.tick(&ctx, ts("2026-10-03T07:02:00.000Z"));
|
||||
assert_eq!(work, vec![TickWork::Heartbeat]);
|
||||
assert_eq!(server.requests_to(CHAT).len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_held_subagent_skips_a_due_cron_job() {
|
||||
let home = Home::new();
|
||||
home.write("compact.md", "Summarise.\n");
|
||||
home.write(
|
||||
"jobs/mail.toml",
|
||||
"schedule = \"0 7 * * *\"\nprompt = \"Check mail.\"\n",
|
||||
);
|
||||
let server = FakeServer::start();
|
||||
let cfg = home.config(&server.socket);
|
||||
server.route(CHAT, vec![sse()]);
|
||||
let ctx = Arc::new(Context::new(
|
||||
cfg.clone(),
|
||||
Client::new(cfg),
|
||||
Box::new(ScriptedPort::new(vec![])),
|
||||
Registry::m2b(),
|
||||
Box::new(NoMemory),
|
||||
));
|
||||
let held = ctx
|
||||
.hold(SessionId::new("sub-mm-thread-1").unwrap())
|
||||
.expect("hold");
|
||||
let mut sched = Scheduler::new(ts("2026-10-03T06:00:00.000Z"));
|
||||
let work = sched.tick(&ctx, ts("2026-10-03T07:00:00.000Z"));
|
||||
assert!(work.is_empty(), "{work:?}");
|
||||
assert!(server.requests_to(CHAT).is_empty());
|
||||
drop(held);
|
||||
let work = sched.tick(&ctx, ts("2026-10-03T07:00:00.000Z"));
|
||||
assert_eq!(work.len(), 1);
|
||||
assert!(matches!(work[0], TickWork::Job { .. }), "{work:?}");
|
||||
}
|
||||
Reference in New Issue
Block a user