docs: M6a implementation plan

docs/plans/M6a/files/crates/brokerd/tests/state_subagent.rs

@@ -0,0 +1,71 @@
+//! A `sub-*` session copies its parent's taint. Do not edit.
+
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use brokerd::policy::{Label, SessionState};
+use brokerd::state::StateStore;
+use bxcore::{DataClass, SessionId};
+use tmp::TempDir;
+
+fn id(text: &str) -> SessionId {
+    SessionId::new(text).unwrap()
+}
+
+fn store(home: &TempDir) -> StateStore {
+    StateStore::new(&home.path().join("broker/sessions"))
+}
+
+#[test]
+fn a_new_subagent_copies_the_parents_untrusted_flag() {
+    let home = TempDir::new("state-sub");
+    let store = store(&home);
+    let parent = id("mm-thread");
+    store
+        .raise(
+            &parent,
+            SessionState::default(),
+            Label {
+                class: DataClass::Private,
+                untrusted: true,
+            },
+        )
+        .unwrap();
+    let child = store.read(&id("sub-mm-thread-3")).unwrap();
+    assert_eq!(
+        child,
+        SessionState {
+            taint: DataClass::Private,
+            untrusted: true,
+        }
+    );
+    let path = home.path().join("broker/sessions/sub-mm-thread-3.json");
+    assert!(path.exists());
+}
+
+#[test]
+fn a_nested_subagent_copies_through_a_missing_middle_child() {
+    let home = TempDir::new("state-sub-nest");
+    let store = store(&home);
+    store
+        .raise(
+            &id("mm-thread"),
+            SessionState::default(),
+            Label {
+                class: DataClass::Secret,
+                untrusted: true,
+            },
+        )
+        .unwrap();
+    let leaf = store.read(&id("sub-sub-mm-thread-1-2")).unwrap();
+    assert_eq!(leaf.taint, DataClass::Secret);
+    assert!(leaf.untrusted);
+}
+
+#[test]
+fn a_plain_session_still_creates_nothing_on_read() {
+    let home = TempDir::new("state-plain");
+    let store = store(&home);
+    let _ = store.read(&id("chat-1")).unwrap();
+    assert!(!home.path().join("broker").exists());
+}

docs/plans/M6a/files/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:?}");
+}

docs/plans/M6a/files/crates/loopd/tests/subagent_dispatch.rs

@@ -0,0 +1,57 @@
+//! Dispatch of `subagent`: not the port, unknown keys, child ids. Do not edit.
+
+use bxcore::CallId;
+use bxcore::SessionId;
+use loopd::subagent::{self, SUBAGENT};
+use loopd::tools::{Dispatch, Registry, dispatch};
+
+#[test]
+fn call_tool_unwraps_subagent_and_does_not_send_it_to_the_port() {
+    let r = Registry::m5c();
+    match dispatch(
+        &r,
+        "call_tool",
+        r#"{"name":"subagent","arguments":{"task":"look around"}}"#,
+    ) {
+        Dispatch::Subagent { task } => assert_eq!(task, "look around"),
+        other => panic!("{other:?}"),
+    }
+    match dispatch(&r, SUBAGENT, r#"{"task":"look around"}"#) {
+        Dispatch::Subagent { task } => assert_eq!(task, "look around"),
+        other => panic!("{other:?}"),
+    }
+}
+
+#[test]
+fn unknown_keys_and_a_missing_task_are_invalid_arguments() {
+    let r = Registry::m5c();
+    let cases = [
+        (r#"{"task":"x","extra":1}"#, "unknown field: extra"),
+        (r#"{"text":"x"}"#, "unknown field: text"),
+        ("{}", "task is required"),
+        (r#"{"task":1}"#, "task must be a string"),
+        ("[]", "arguments must be a JSON object"),
+        ("not json", "arguments must be a JSON object"),
+    ];
+    for (args, want) in cases {
+        match dispatch(&r, SUBAGENT, args) {
+            Dispatch::Local(text) => {
+                assert!(
+                    text.starts_with("Denied: the arguments are not valid for this tool:"),
+                    "{args}: {text}"
+                );
+                assert!(text.contains(want), "{args}: {text}");
+            }
+            other => panic!("{args}: {other:?}"),
+        }
+    }
+}
+
+#[test]
+fn a_parent_id_that_cannot_fit_a_child_is_named_in_the_denial() {
+    let parent = SessionId::new(&"a".repeat(59)).unwrap();
+    let err = subagent::child_id(&parent, CallId(1)).unwrap_err();
+    assert!(err.len() > 64, "{err}");
+    assert!(err.starts_with("sub-"), "{err}");
+    assert!(subagent::child_id(&SessionId::new("mm-thread").unwrap(), CallId(3)).is_ok());
+}

docs/plans/M6a/files/crates/loopd/tests/subagent_registry.rs

@@ -0,0 +1,57 @@
+//! `subagent` in the M5c registry, not core. Do not edit.
+
+mod support;
+
+use loopd::baseline::Baseline;
+use loopd::subagent::{self, SUBAGENT};
+use loopd::tools::Registry;
+use std::path::Path;
+use support::Home;
+
+fn core_names(registry: &Registry) -> Vec<String> {
+    registry
+        .core_schemas()
+        .into_iter()
+        .map(|s| s.name)
+        .collect()
+}
+
+#[test]
+fn subagent_is_in_m5c_and_is_not_core() {
+    let r = Registry::m5c();
+    let entry = r.get(SUBAGENT).expect("subagent");
+    assert!(!entry.core);
+    assert_eq!(entry.schema, subagent::schema());
+    assert_eq!(
+        core_names(&r),
+        [
+            "clock",
+            "recall",
+            "memory",
+            "ask_memory",
+            "find_tool",
+            "call_tool"
+        ]
+    );
+    assert!(r.find("subagent").iter().any(|s| s.name == SUBAGENT));
+    assert!(r.find("isolated").iter().any(|s| s.name == SUBAGENT));
+}
+
+#[test]
+fn a_baseline_from_m5c_does_not_put_subagent_in_the_tools_array() {
+    let home = Home::new();
+    let cfg = home.config(Path::new("/tmp/unused.sock"));
+    let baseline = Baseline::assemble(&cfg, &Registry::m5c()).unwrap();
+    let names: Vec<&str> = baseline.tools.iter().map(|t| t.name.as_str()).collect();
+    assert_eq!(
+        names,
+        [
+            "clock",
+            "recall",
+            "memory",
+            "ask_memory",
+            "find_tool",
+            "call_tool"
+        ]
+    );
+}

docs/plans/M6a/files/crates/loopd/tests/subagent_spawn.rs

@@ -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);
+}

docs/plans/M6a/files/runbook-subagent-id.md

@@ -0,0 +1,25 @@
+## 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>/`.