loopd: subagent spawn on the background slot

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
This commit is contained in:
2026-10-03 19:59:25 -07:00
co-authored by ornith-1.5-35b-a3b
parent fe145fb83a
commit 4491e1b80d
13 changed files with 471 additions and 12 deletions
+1
View File
@@ -146,6 +146,7 @@ pub fn handle(stream: UnixStream, ctx: Arc<Context>) {
client: &ctx.client,
port: &*ctx.port,
registry: &ctx.registry,
ctx: Some(&*ctx),
};
let mut session = if turn.resume {
+2
View File
@@ -136,6 +136,7 @@ impl Scheduler {
client: &ctx.client,
port: &*ctx.port,
registry: &ctx.registry,
ctx: Some(ctx),
};
let text = format!("It is {}.\n\n{}", now.to_rfc3339(), job.prompt);
let mut emit = |_event: &TurnEvent| {};
@@ -247,6 +248,7 @@ impl Scheduler {
client: &ctx.client,
port: &*ctx.port,
registry: &ctx.registry,
ctx: Some(ctx),
};
let text = format!("It is {}.\n\n{file}", now.to_rfc3339());
+118 -12
View File
@@ -1,8 +1,20 @@
//! The `subagent` tool: loopd answers it itself (M6 spec, section 3).
//! It is not core; the parent finds it with `find_tool` and calls it through `call_tool`.
use crate::baseline::Baseline;
use crate::channel::Busy;
use crate::compact;
use crate::session::{Session, SessionError};
use crate::tools::{cap_result, denial_message};
use crate::turn::{Runtime, run_turn};
use bxcore::sanitize::neutralise;
use bxcore::{CallId, DataClass, DenyReason, SessionId, Timestamp, TurnEvent};
pub const SUBAGENT: &str = "subagent";
/// What `subagent` returns when a child session id would not fit a `SessionId`.
pub const RUNBOOK: &str = "see docs/runbook.md#subagent-id";
pub fn schema() -> crate::llama::ToolSchema {
crate::llama::ToolSchema {
name: SUBAGENT.to_string(),
@@ -62,18 +74,112 @@ pub fn dispatch_args(arguments: &str) -> crate::tools::Dispatch {
}
}
/// Spawn is task 03. The match arm in `run_call` needs this name now.
/// Spawn the child, run one turn on the background slot, and return the parent's tool text.
pub fn run(
rt: &crate::turn::Runtime<'_>,
parent: &crate::session::Session,
rt: &Runtime<'_>,
parent: &Session,
task: &str,
call_id: bxcore::CallId,
on_event: &mut dyn FnMut(&bxcore::TurnEvent),
) -> (String, bxcore::DataClass, bool) {
let _ = (rt, parent, task, call_id, on_event);
(
"subagent stub".to_string(),
bxcore::DataClass::Public,
false,
)
call_id: CallId,
on_event: &mut dyn FnMut(&TurnEvent),
) -> (String, DataClass, bool) {
let raw = format!("sub-{}-{}", parent.id().as_str(), call_id.0);
let child = match SessionId::new(&raw) {
Ok(id) => id,
Err(_) => {
return (
denial_message(
DenyReason::InvalidArguments,
Some(&format!(
"child session id `{raw}` would not fit; {RUNBOOK}"
)),
),
DataClass::Public,
false,
);
}
};
let Some(ctx) = rt.ctx else {
return failed(&child);
};
let held = match ctx.hold(child.clone()) {
Ok(held) => held,
Err(Busy) => return failed(&child),
};
let block = match ctx.memory.block() {
Ok(block) => block,
Err(_) => {
drop(held);
return failed(&child);
}
};
let baseline = match Baseline::assemble_with(rt.cfg, rt.registry, &block) {
Ok(baseline) => baseline,
Err(_) => {
drop(held);
return failed(&child);
}
};
let (task, _) = cap_result(task, rt.cfg.r#loop.tool_result_cap);
let text = format!("It is {}.\n\n{task}", Timestamp::now().to_rfc3339());
let mut session = match Session::create(
&rt.cfg.paths.home,
child.clone(),
baseline,
rt.cfg.slots.background,
) {
Ok(session) => session,
Err(SessionError::Exists(_)) => match Session::open(&rt.cfg.paths.home, child.clone()) {
Ok(session) => session,
Err(_) => {
drop(held);
return failed(&child);
}
},
Err(_) => {
drop(held);
return failed(&child);
}
};
let outcome = run_turn(&mut session, rt, &text, None, on_event);
let outcome = compact::finish_turn(
&mut session,
rt,
&*ctx.memory,
&text,
None,
on_event,
outcome,
);
let _ = ctx.memory.updated(session.id(), session.turns());
let untrusted = session.untrusted();
drop(held);
match outcome {
Ok(done) => {
let summary = compact::cap_chars(
neutralise(&done.content).as_ref(),
rt.cfg.compact.summary_chars,
);
(
format!("{} completed\n{summary}", child.as_str()),
DataClass::Public,
untrusted,
)
}
Err(_) => (
format!("{} failed", child.as_str()),
DataClass::Public,
untrusted,
),
}
}
fn failed(id: &SessionId) -> (String, DataClass, bool) {
(format!("{} failed", id.as_str()), DataClass::Public, false)
}
+3
View File
@@ -57,6 +57,9 @@ pub struct Runtime<'a> {
pub client: &'a Client,
pub port: &'a dyn ToolPort,
pub registry: &'a Registry,
/// Present on a live `loopd` turn so `subagent` can hold the child. Tests that do not spawn
/// leave this `None`.
pub ctx: Option<&'a crate::channel::Context>,
}
/// True for the one 400 the server sends when the prompt does not fit.
+2
View File
@@ -75,6 +75,7 @@ impl World {
client: &self.client,
port: &self.port,
registry: &self.registry,
ctx: None,
}
}
@@ -244,6 +245,7 @@ fn a_successful_turn_stays_successful_when_compact_fails() {
client: &client,
port: &port,
registry: &registry,
ctx: None,
};
let mut events = Vec::new();
let outcome = run_turn(&mut s, &rt, "hi", None, &mut |e| events.push(e.clone()));
+1
View File
@@ -115,6 +115,7 @@ fn a_call_no_grant_allows_reaches_the_model_as_a_denial_and_is_audited() {
client: &client,
port: &port,
registry: &registry,
ctx: None,
};
let mut events = Vec::new();
let outcome = run_turn(
+1
View File
@@ -33,6 +33,7 @@ fn runtime<'a>(
client,
port,
registry,
ctx: None,
}
}
+313
View File
@@ -0,0 +1,313 @@
//! A `subagent` turn on the background slot. Do not edit.
mod support;
use bxcore::{LogRecord, SessionId};
use loopd::baseline::Baseline;
use loopd::channel::Context;
use loopd::llama::Client;
use loopd::memory::NoMemory;
use loopd::session::Session;
use loopd::tools::Registry;
use loopd::turn::{Runtime, run_turn};
use std::sync::Arc;
use support::{FakeServer, Home, Reply, ScriptedPort};
const CHAT: &str = "/v1/chat/completions";
fn sse_chunks(chunks: &[serde_json::Value]) -> Reply {
let mut body = String::new();
for chunk in chunks {
body.push_str("data: ");
body.push_str(&chunk.to_string());
body.push_str("\n\n");
}
body.push_str("data: [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 timings() -> serde_json::Value {
serde_json::json!({"cache_n": 0, "prompt_n": 10, "predicted_n": 1})
}
fn sse_text(text: &str) -> Reply {
sse_chunks(&[
serde_json::json!({
"choices": [{"finish_reason": null, "index": 0, "delta": {"content": text}}],
"id": "t",
"timings": timings()
}),
serde_json::json!({
"choices": [{"finish_reason": "stop", "index": 0, "delta": {}}],
"id": "t",
"timings": timings()
}),
])
}
fn sse_call_tool(tool: &str, arguments: serde_json::Value) -> Reply {
let args = serde_json::json!({"name": tool, "arguments": arguments}).to_string();
sse_named("call_tool", &args)
}
fn sse_named(name: &str, arguments: &str) -> Reply {
sse_chunks(&[
serde_json::json!({
"choices": [{
"finish_reason": null,
"index": 0,
"delta": {
"tool_calls": [{
"index": 0,
"id": "c1",
"type": "function",
"function": {"name": name, "arguments": arguments}
}]
}
}],
"id": "t",
"timings": timings()
}),
serde_json::json!({
"choices": [{"finish_reason": "tool_calls", "index": 0, "delta": {}}],
"id": "t",
"timings": timings()
}),
])
}
fn world(replies: Vec<Reply>) -> (Home, FakeServer, Arc<Context>) {
let home = Home::new();
home.write("compact.md", "Summarise.\n");
let server = FakeServer::start();
let cfg = home.config(&server.socket);
server.route(CHAT, replies);
let ctx = Arc::new(Context::new(
cfg.clone(),
Client::new(cfg),
Box::new(ScriptedPort::new(vec![])),
Registry::m5c(),
Box::new(NoMemory),
));
(home, server, ctx)
}
fn runtime(ctx: &Context) -> Runtime<'_> {
Runtime {
cfg: &ctx.cfg,
client: &ctx.client,
port: &*ctx.port,
registry: &ctx.registry,
ctx: Some(ctx),
}
}
fn parent(ctx: &Context, id: &str) -> Session {
let baseline = Baseline::assemble(&ctx.cfg, &ctx.registry).unwrap();
Session::create(
&ctx.cfg.paths.home,
SessionId::new(id).unwrap(),
baseline,
ctx.cfg.slots.main,
)
.unwrap()
}
fn last_tool_text(session: &Session) -> String {
for record in session.records().iter().rev() {
if let LogRecord::ToolResult { content, .. } = record {
return content.clone();
}
}
panic!("no tool result");
}
#[test]
fn a_child_runs_on_the_background_slot_and_the_parent_prefix_still_extends() {
let (home, server, ctx) = world(vec![
sse_call_tool("subagent", serde_json::json!({"task": "look around"})),
sse_text("the drawer is empty"),
sse_text("ok"),
]);
let mut session = parent(&ctx, "mm-thread");
let rt = runtime(&ctx);
let mut events = Vec::new();
run_turn(&mut session, &rt, "search", None, &mut |e| {
events.push(e.clone())
})
.unwrap();
let sent = server.requests_to(CHAT);
assert_eq!(sent.len(), 3);
assert_eq!(sent[0].json()["id_slot"], ctx.cfg.slots.main);
assert_eq!(sent[1].json()["id_slot"], ctx.cfg.slots.background);
assert_eq!(sent[2].json()["id_slot"], ctx.cfg.slots.main);
let first = sent[0].json()["messages"].as_array().unwrap().clone();
let third = sent[2].json()["messages"].as_array().unwrap().clone();
assert!(third.len() > first.len());
assert_eq!(&third[..first.len()], &first[..]);
let tool = third[first.len() + 1].clone();
assert_eq!(tool["role"], "tool");
assert!(
tool["content"]
.as_str()
.unwrap()
.starts_with("sub-mm-thread-1 completed\n"),
"{tool}"
);
assert!(
tool["content"]
.as_str()
.unwrap()
.contains("the drawer is empty"),
"{tool}"
);
let child_body = sent[1].json();
let child_msgs = child_body["messages"].as_array().unwrap();
let last = child_msgs.last().unwrap();
assert_eq!(last["role"], "user");
let content = last["content"].as_str().unwrap();
assert!(content.starts_with("It is "), "{content}");
assert!(content.contains("look around"), "{content}");
let child = Session::open(&home.dir, SessionId::new("sub-mm-thread-1").unwrap()).unwrap();
assert_eq!(child.slot(), ctx.cfg.slots.background);
}
#[test]
fn a_second_spawn_in_the_same_parent_uses_the_next_call_id() {
let (_home, _server, ctx) = world(vec![
sse_call_tool("subagent", serde_json::json!({"task": "one"})),
sse_text("first"),
sse_call_tool("subagent", serde_json::json!({"task": "two"})),
sse_text("second"),
sse_text("ok"),
]);
let mut session = parent(&ctx, "mm-thread");
let rt = runtime(&ctx);
let mut events = Vec::new();
run_turn(&mut session, &rt, "both", None, &mut |e| {
events.push(e.clone())
})
.unwrap();
let texts: Vec<String> = session
.records()
.iter()
.filter_map(|r| match r {
LogRecord::ToolResult { content, .. } => Some(content.clone()),
_ => None,
})
.collect();
assert!(
texts[0].starts_with("sub-mm-thread-1 completed\n"),
"{texts:?}"
);
assert!(
texts[1].starts_with("sub-mm-thread-2 completed\n"),
"{texts:?}"
);
}
#[test]
fn an_overflowing_parent_id_is_invalid_arguments() {
let parent_id = "a".repeat(59);
let (_home, server, ctx) = world(vec![
sse_call_tool("subagent", serde_json::json!({"task": "x"})),
sse_text("ok"),
]);
let mut session = parent(&ctx, &parent_id);
let rt = runtime(&ctx);
let mut events = Vec::new();
run_turn(&mut session, &rt, "go", None, &mut |e| {
events.push(e.clone())
})
.unwrap();
let text = last_tool_text(&session);
assert!(
text.contains("would not fit") && text.contains("subagent-id"),
"{text}"
);
assert_eq!(server.requests_to(CHAT).len(), 2);
}
#[test]
fn a_nested_child_returns_through_the_middle_tool_result() {
let (home, server, ctx) = world(vec![
sse_call_tool("subagent", serde_json::json!({"task": "outer"})),
sse_call_tool("subagent", serde_json::json!({"task": "inner"})),
sse_text("leaf"),
sse_text("middle"),
sse_text("ok"),
]);
let mut session = parent(&ctx, "mm-thread");
let rt = runtime(&ctx);
let mut events = Vec::new();
run_turn(&mut session, &rt, "nest", None, &mut |e| {
events.push(e.clone())
})
.unwrap();
let middle = Session::open(&home.dir, SessionId::new("sub-mm-thread-1").unwrap()).unwrap();
let leaf = Session::open(&home.dir, SessionId::new("sub-sub-mm-thread-1-1").unwrap()).unwrap();
assert_eq!(leaf.slot(), ctx.cfg.slots.background);
let middle_text = last_tool_text(&middle);
assert!(
middle_text.starts_with("sub-sub-mm-thread-1-1 completed\n"),
"{middle_text}"
);
assert!(middle_text.contains("leaf"), "{middle_text}");
let parent_text = last_tool_text(&session);
assert!(
parent_text.starts_with("sub-mm-thread-1 completed\n"),
"{parent_text}"
);
assert!(parent_text.contains("middle"), "{parent_text}");
let sent = server.requests_to(CHAT);
assert_eq!(sent[0].json()["id_slot"], 0);
assert_eq!(sent[1].json()["id_slot"], 1);
assert_eq!(sent[2].json()["id_slot"], 1);
let parent_first = sent[0].json()["messages"].as_array().unwrap().clone();
let parent_last = sent[4].json()["messages"].as_array().unwrap().clone();
assert_eq!(&parent_last[..parent_first.len()], &parent_first[..]);
}
#[test]
fn a_child_that_ends_on_a_tool_call_is_failed() {
let home = Home::new();
home.write("compact.md", "Summarise.\n");
let server = FakeServer::start();
let mut cfg = home.config(&server.socket);
cfg.r#loop.tool_iterations = 1;
server.route(
CHAT,
vec![
sse_call_tool("subagent", serde_json::json!({"task": "loop"})),
sse_named("clock", "{}"),
sse_named("clock", "{}"),
sse_text("ok"),
],
);
let ctx = Arc::new(Context::new(
cfg.clone(),
Client::new(cfg),
Box::new(ScriptedPort::new(vec![])),
Registry::m5c(),
Box::new(NoMemory),
));
let mut session = parent(&ctx, "mm-thread");
let rt = runtime(&ctx);
let mut events = Vec::new();
run_turn(&mut session, &rt, "go", None, &mut |e| {
events.push(e.clone())
})
.unwrap();
let text = last_tool_text(&session);
assert_eq!(text, "sub-mm-thread-1 failed");
drop(home);
}
+1
View File
@@ -54,6 +54,7 @@ impl Setup {
client: &self.client,
port: &self.port,
registry: &self.registry,
ctx: None,
}
}
+1
View File
@@ -214,6 +214,7 @@ fn after_a_turn_an_opted_in_call_is_summarised() {
client: &client,
port: &port,
registry: &registry,
ctx: None,
};
let mut events = Vec::new();
let outcome = run_turn(&mut s, &rt, "hosts?", None, &mut |e| events.push(e.clone()));
+1
View File
@@ -241,6 +241,7 @@ fn a_port_that_returns_a_pending_frame_as_its_answer_is_a_failure_not_a_decision
client: &s.client,
port: &ReturnsPendingPort,
registry: &s.registry,
ctx: None,
};
let mut events = Vec::new();
let result = loopd::turn::run_turn(&mut session, &runtime, "x", None, &mut |e| {
+1
View File
@@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone.
| Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model |
|---|---|---|---|---|---|---|---|
| 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-<parent>-<n>` 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 |
| M5d2/11-loopd-heartbeat-failed-last | 2026-10-03 | done | 1 | pass | none | Copied the given test `crates/loopd/tests/scheduler_heartbeat_failed.rs` from `docs/plans/M5d2/files/`. Applied H1 in `crates/loopd/src/scheduler.rs` inside `heartbeat`: after the quoted block (found exactly once) the bookkeeping is now unconditional — `let _ = compact::finish_turn(...)` instead of binding `outcome`, then `drop(held)`, then `self.last_heartbeat = Some(now)` and `work.push(TickWork::Heartbeat)` with no `if outcome.is_ok()`. The unreadable-prompt and busy `return` paths stay untouched and never reach H1; cron `Finished`/`Failed` and the job arm are unchanged. `cargo test -p loopd --test scheduler_heartbeat_failed` failed first (`left: []`, `right: [Heartbeat]`), then passed with `scheduler_heartbeat` (6). `cargo check -p loopd` clean, no warning. `make gate` printed `gate: ok` with 1575 tests (1574 before, plus 1). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b |
+26
View File
@@ -1243,3 +1243,29 @@ session.
**Check.** `sessions/heartbeat/0.jsonl` gains a new `User` whose text starts with `It is ` and an
RFC 3339 timestamp.
## subagent-id
**What you see.** A `subagent` call is denied as invalid-arguments. The tool result names the
child id that would not fit and this entry:
```text
Denied: the arguments are not valid for this tool: child session id `sub-…` would not fit; see docs/runbook.md#subagent-id.
```
**Why.** A session id is `[a-z0-9-]`, at most 64 bytes. The child is `sub-<parent>-<n>` with
`n` the parent's next call id. A parent already near that limit, or a nested chain of
`sub-` prefixes, cannot spawn another child.
**Confirm.**
```sh
printf '%s' '<the id named in the denial>' | wc -c
```
**Fix.** Do the work in the current session, or start a shorter parent id (`bxctl chat` without
a long `--session`). Nested children share one background slot; a parallel pool is out of
scope.
**Check.** A `subagent` call from a short parent (`chat-*`, `mm-*`, `heartbeat`) completes and
writes `sessions/sub-<parent>-<n>/`.