From 79735b21e490172d5e0478670bf766ed3480fef9 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 3 Oct 2026 20:03:13 -0700 Subject: [PATCH] loopd: scheduler yields to a held subagent Co-Authored-By: ornith-1.5-35b-a3b --- crates/loopd/src/channel.rs | 6 ++ crates/loopd/src/scheduler.rs | 6 ++ crates/loopd/tests/scheduler_subagent.rs | 89 ++++++++++++++++++++++++ docs/implementer-log.md | 1 + 4 files changed, 102 insertions(+) create mode 100644 crates/loopd/tests/scheduler_subagent.rs diff --git a/crates/loopd/src/channel.rs b/crates/loopd/src/channel.rs index ca5b4bf..1ccab5e 100644 --- a/crates/loopd/src/channel.rs +++ b/crates/loopd/src/channel.rs @@ -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 diff --git a/crates/loopd/src/scheduler.rs b/crates/loopd/src/scheduler.rs index 3a01663..39c81f6 100644 --- a/crates/loopd/src/scheduler.rs +++ b/crates/loopd/src/scheduler.rs @@ -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, diff --git a/crates/loopd/tests/scheduler_subagent.rs b/crates/loopd/tests/scheduler_subagent.rs new file mode 100644 index 0000000..562a8cb --- /dev/null +++ b/crates/loopd/tests/scheduler_subagent.rs @@ -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:?}"); +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 9d0de44..da4a18d 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 | |---|---|---|---|---|---|---|---| +| M6a/04-loopd-scheduler-skip | 2026-10-03 | done | 1 | pass | none | Copied the given test `crates/loopd/tests/scheduler_subagent.rs` from `docs/plans/M6a/files/`. Step 2 (`cargo test -p loopd --test scheduler_subagent`) failed as expected: both tests failed, `work` not empty (a `Heartbeat` and a `Job { name: "mail", session: "job-mail-20261003t0700" }`) and `requests_to(CHAT)` not empty while the child was held. Applied H1 in channel.rs: added `holds_prefix` after `hold`'s closing `}` inside `impl Context`, locking `busy` like `hold` and returning whether any id starts with the prefix. Applied K2 in scheduler.rs `heartbeat`: after the `if !due { return; }` block and before `SessionId::new("heartbeat")`, `if ctx.holds_prefix("sub-") { return; }`. Applied K1 in scheduler.rs `tick`: immediately after `self.heartbeat(...)` and before `jobs_dir`, `if ctx.holds_prefix("sub-") { return work; }` so leftover-close on the first tick is already done. `cargo check -p loopd` clean, no warning; `wc -l` prints 307 channel.rs, 310 scheduler.rs, both under 500. All suites pass: `scheduler_subagent` 2, `scheduler_heartbeat` 6, `scheduler_cron` 6, `scheduler_heartbeat_failed` 1. `make gate` printed `gate: ok` with 1587 tests (1585 before, plus 2). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b | | M6a/03-loopd-subagent-spawn | 2026-10-03 | done | 1 | pass | none | Copied the given test `crates/loopd/tests/subagent_spawn.rs` from `docs/plans/M6a/files/`. Step 2 (`cargo build --tests --keep-going -p loopd`) failed as expected: E0560 `Runtime` has no field `ctx`. Added `pub ctx: Option<&'a crate::channel::Context>` to `Runtime` in turn.rs R1, filled it `Some(&*ctx)` in channel.rs C1 (live turn), `Some(ctx)` in the cron and heartbeat literals in scheduler.rs S1/S2 (both already held an immutable borrow of `ctx`), and `None` in the seven test literals: support/turn.rs, compact_turn.rs (the helper and the inline one), session_slot.rs, end_to_end.rs, turn_broker.rs, tool_summary_call.rs. Replaced the stub `run` in subagent.rs with the full spawn: build `sub--` from the already-consumed `call_id.0`, overflow returns `denial_message(InvalidArguments, Some(...))` naming subagent-id without holding, then `rt.ctx` `None` and `hold Busy` and block/baseline/create errors each return `failed`, a `Create` `Exists` falls back to `open`, the child runs one `run_turn` plus `finish_turn` on `slots.background`, memory is updated, `untrusted` is read before the hold drops, and the result is `{id} completed\n{summary}` or `{id} failed`, all `Public`. Added `RUNBOOK` and the use lines the body needs. Appended `## subagent-id` to docs/runbook.md via the task's `{ echo; cat ...; }` (grep -c prints 1). `cargo check -p loopd` clean, no warning; `wc -l` prints 185 subagent.rs, 335 turn.rs, 301 channel.rs, 304 scheduler.rs, all under 500. All suites pass: `subagent_spawn` 5, `subagent_dispatch` 3, `subagent_registry` 2, `session_slot` 4, `tool_summary_call` 6, `compact_turn`, `channel`. `make gate` printed `gate: ok` with 1585 tests (1580 before, plus 5). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b | | M6a/02-loopd-subagent-dispatch | 2026-10-03 | done | 1 | pass | none | Copied the given test `crates/loopd/tests/subagent_dispatch.rs` from `docs/plans/M6a/files/`. Step 2 (`cargo build --tests --keep-going -p loopd`) failed as expected: E0599 no `Subagent` variant and E0425 no `child_id`. Added `Dispatch::Subagent { task }` in tools.rs D1, routed `subagent` in `dispatch` and `dispatch_call_tool` (D2/D3) via `crate::subagent::dispatch_args`, added the `Subagent` arm in `run_call` (turn.rs T1) calling the stub `run`, and appended `child_id`, `parse_task`, `dispatch_args` and the stub `run` to subagent.rs. `parse_task` walks every key (unknown field), requires `task`, demands a string task and a JSON object arguments, and maps errors to `denial_message(InvalidArguments, Some(&detail))`. `run` stays a compiling stub, no spawn, no `Runtime.ctx`. `cargo check -p loopd` clean, no warning; `wc -l` prints 76 for subagent.rs, 477 for tools.rs, 332 for turn.rs, all under 500. All suites pass: `subagent_dispatch` 3, `subagent_registry` 2, `tools` 11. `make gate` printed `gate: ok` with 1580 tests (1577 before, plus 3). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b | | M6a/01-loopd-subagent-registry | 2026-10-03 | done | 1 | pass | none | Copied the given test `crates/loopd/tests/subagent_registry.rs` from `docs/plans/M6a/files/`. Created `crates/loopd/src/subagent.rs` with `SUBAGENT` and `schema()` (a non-core `ToolSchema`, no `parse_task`/`dispatch_args`/`run`). Added `pub mod subagent;` after `pub mod sse;` in lib.rs (rustfmt order). Edited `m5c()` in tools.rs to push the non-core `subagent` entry after the `ask_memory` insert at index 3, so `ask_memory` stays at 3 and `subagent` is last among the non-core names; `core_schemas()` still returns only the six core names and `find("subagent")`/`find("isolated")` both hit it. `cargo check -p loopd` clean, no warning; `wc -l` prints 22 for subagent.rs and 470 for tools.rs, both under 500. All suites pass: `subagent_registry` 2, `ask_memory_tool`, `tools` 11, `baseline` 6. `make gate` printed `gate: ok` with 1577 tests (1575 before, plus 2). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b |