Scheduled work on this computer (spec 3.5.0): Accept scheduled work, scheduled turns, the watch command; team_add names the sub-manager

remote.scheduled (lembas service scheduled on|off, lembas remote scheduled on|off) said as
limits.scheduled; a scheduled session runs in the schedule's mode (Edit when none, Plan refused,
lowered to max_mode), reports ok/silent/failed/cancelled and the final reply, waits the
schedule's wait on approvals and offers no question, plan card or show. _lembas/monitor judges
the watch command by the mode (ask is refused) and hashes the output as it streams.
Conformance areas scheduled, silent and monitor run here.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/acp.test.ts

@@ -137,7 +137,7 @@ test("the device's limits: outside the roots, untrusted, a higher mode", async (
   const root = project()
   const limits: Limits = { roots: [root], maxMode: "edit", approvalTimeoutMs: 50, requireTrust: true }
   const { c, init } = await client(limits)
-  expect(init._meta.lembas.limits).toEqual({ roots: [root], max_mode: "edit" })
+  expect(init._meta.lembas.limits).toEqual({ roots: [root], max_mode: "edit", scheduled: false })
   await expect(c.request("session/new", { cwd: tmpdir() })).rejects.toThrow("outside the directories")
   const untrusted = join(root, "..", `ph-untrusted-${Date.now()}`)
   mkdirSync(untrusted)
@@ -268,7 +268,7 @@ test("the terminal: a real shell through the link, only where remote.terminal sa
   config(fake.url)
   const root = project()
   const off = await client({ roots: [root], maxMode: "edit", approvalTimeoutMs: 50, requireTrust: true })
-  expect(off.init._meta.lembas.capabilities).toEqual(["directories", "configure", "steer", "sessions", "status", "delete", "compact", "title", "files", "commands", "attachments", "ask", "plan", "model", "turns", "context", "library", "prototypes", "plan_state", "rewind", "jobs", "show", "progress", "tree", "agents"])
+  expect(off.init._meta.lembas.capabilities).toEqual(["directories", "configure", "steer", "sessions", "status", "delete", "compact", "title", "files", "commands", "attachments", "ask", "plan", "model", "turns", "context", "library", "prototypes", "plan_state", "rewind", "jobs", "show", "progress", "tree", "agents", "scheduled"])
   await expect(off.c.request("_lembas/terminal/open", { cwd: root })).rejects.toThrow("remote.terminal is off")
 
   const { c, init } = await client({ roots: [root], maxMode: "edit", approvalTimeoutMs: 50, requireTrust: true, terminal: true })

tests/conformance.ts

@@ -34,6 +34,7 @@ import { offer, subagentModel, type ModelInfo } from "../src/team/subagents.ts"
 import { progressLine, progressText, type Progress } from "../src/session/progress.ts"
 import { checkMode, requestRefusal, tighten, whoDecides, type Grant } from "../src/team/delegation.ts"
 import { fillPrompt as fillText } from "../src/prompt/fill.ts"
+import { isSilent, judgeCase, monitorHash, scheduledMode, stripSentinel } from "../src/acp/scheduled.ts"
 
 type Case = Record<string, unknown>
 
@@ -50,6 +51,14 @@ function readableUnder(grants: Record<string, string> | undefined, path: string)
 }
 
 export const AREAS: Record<string, (c: Case) => unknown> = {
+  // Spec 3.5.0: scheduled work on a linked computer (src/acp/scheduled.ts).
+  scheduled: (c) => scheduledMode({ accept: c.accept === true, mode: c.mode, max_mode: c.max_mode as Mode, approval_wait_s: c.approval_wait_s }),
+  silent: (c) => ({ silent: isSilent(c.text as string), stripped: stripSentinel(c.text as string) }),
+  monitor: (c) => {
+    if (c.kind === "hash") return monitorHash(c.exit as number, c.output as string)
+    const r = judgeCase(c.line as string, c.mode as Mode, c.rules as PermissionConfig | undefined)
+    return "run" in r ? "run" : { refused: r.refused }
+  },
   hardline: (c) => hardlineCommand(c.line as string, BUILTIN_HARDLINE)?.id ?? null,
   split: (c) => splitCommand(c.line as string),
   plain: (c) => plainCommands(c.line as string),

tests/remote.test.ts

@@ -37,7 +37,7 @@ test("list --json: one object, the defaults applied", () => {
   const r = run("list", "--json")
   expect(r.code).toBe(0)
   expect(r.out).toHaveLength(1)
-  expect(r.json()).toEqual({ enabled: false, roots: [], max_mode: "edit", terminal: false })
+  expect(r.json()).toEqual({ enabled: false, roots: [], max_mode: "edit", terminal: false, scheduled: false })
   // Without --json, lines a person reads.
   const plain = run("list")
   expect(plain.out.join("\n")).toContain("remote work: off")
@@ -54,7 +54,7 @@ test("add: absolute real paths, turns remote work on, skips duplicates and cover
   const r = run("add", a, link, "--json")
   expect(r.code).toBe(0)
   // The link is stored as where it leads.
-  expect(r.json()).toEqual({ enabled: true, roots: [a, b], max_mode: "edit", terminal: false })
+  expect(r.json()).toEqual({ enabled: true, roots: [a, b], max_mode: "edit", terminal: false, scheduled: false })
   // Again, and a directory inside one: nothing more.
   const again = run("add", a, join(a, "sub"))
   expect(again.code).toBe(0)
@@ -212,7 +212,7 @@ test("the service links with remote work off, says no roots and refuses everythi
   // Turned on from outside, as Desktop does: it links again, saying the new roots, and serves.
   run("add", a)
   await until(() => hellos.length === 2, "a second hello")
-  expect(hellos[1].limits).toEqual({ roots: [a], max_mode: "edit", require_trust: true })
+  expect(hellos[1].limits).toEqual({ roots: [a], max_mode: "edit", require_trust: true, scheduled: false })
   expect(states.some((s) => s.detail === "the remote settings changed; linking again")).toBe(true)
   await ask("initialize", { protocolVersion: 1, clientCapabilities: { _meta: { lembas: { protocol: 2 } } } })
   expect((await ask("_lembas/directories", {})).result.places.map((p: any) => [p.path, p.accepted])).toEqual([[a, true]])

tests/scheduled.test.ts

@@ -0,0 +1,191 @@
+// Scheduled work on a linked computer (harness spec 3.5.0, acp.md): Accept scheduled work, a
+// scheduled turn's mode, approvals that wait the schedule's wait, the run's outcome in the answer,
+// and `_lembas/monitor`. A fake LLeMbas is the client; the model is a scripted endpoint.
+import { afterEach, expect, test } from "bun:test"
+import { mkdirSync, mkdtempSync, realpathSync, writeFileSync } from "node:fs"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+import { AcpAgent, type Limits } from "../src/acp/agent.ts"
+import { Peer, type Transport } from "../src/acp/rpc.ts"
+import { approvalWait, deniedText, monitorHash, runMonitor, SCHEDULED } from "../src/acp/scheduled.ts"
+import { paths } from "../src/config/paths.ts"
+import { remoteState } from "../src/config/remote.ts"
+import { setTrust } from "../src/project/root.ts"
+import { remoteCommand } from "../src/remote.ts"
+import { delta, fakeProvider, toolCall, type Fake } from "./fake-provider.ts"
+
+let fake: Fake | undefined
+afterEach(() => fake?.stop())
+
+function pair(): [Transport, Transport] {
+  const make = () => ({ msg: (_: string) => {}, end: () => {} })
+  const a = make()
+  const b = make()
+  const side = (me: typeof a, other: typeof a): Transport => ({
+    send: (t) => queueMicrotask(() => other.msg(t)),
+    onMessage: (fn) => (me.msg = fn),
+    onClose: (fn) => (me.end = fn),
+    close: () => (me.end(), other.end()),
+  })
+  return [side(a, b), side(b, a)]
+}
+
+function config(url: string) {
+  mkdirSync(paths.config, { recursive: true })
+  writeFileSync(join(paths.config, "connections.yaml"), `connections:\n  f:\n    type: webui\n    url: ${url}\n    models: { f/m: {} }\n`, { mode: 0o600 })
+  writeFileSync(join(paths.config, "config.yaml"), "model: f/m\n")
+}
+
+function project(): string {
+  const dir = realpathSync(mkdtempSync(join(tmpdir(), "ph-sched-")))
+  Bun.spawnSync(["git", "init", "-q", dir])
+  setTrust(dir, "trusted")
+  return dir
+}
+
+const limitsFor = (root: string, extra: Partial<Limits> = {}): Limits => ({ roots: [root], maxMode: "auto", approvalTimeoutMs: 3000, requireTrust: true, enabled: true, scheduled: true, ...extra })
+
+async function client(limits: Limits, answer?: (p: any) => unknown) {
+  const [ta, tb] = pair()
+  new AcpAgent(new Peer(ta), limits)
+  const c = new Peer(tb)
+  const asked: any[] = []
+  const settled: any[] = []
+  c.on("_lembas/permission/settled", (p) => void settled.push(p))
+  c.on("_lembas/event", () => {})
+  c.handle("session/request_permission", (p) => (asked.push(p), answer ? answer(p) : { outcome: { outcome: "selected", optionId: "once" } }))
+  const init: any = await c.request("initialize", { protocolVersion: 1, clientCapabilities: { _meta: { lembas: { protocol: 3 } } } })
+  return { c, init, asked, settled }
+}
+
+const sched = (extra: Record<string, unknown> = {}) => ({ schedule_id: "sched1", run_id: "run7", title: "Check", ...extra })
+
+test("capability scheduled and limits.scheduled in initialize", async () => {
+  fake = fakeProvider([])
+  config(fake.url)
+  const root = project()
+  const on = await client(limitsFor(root))
+  expect(on.init._meta.lembas.capabilities).toContain("scheduled")
+  expect(on.init._meta.lembas.limits.scheduled).toBe(true)
+  const off = await client(limitsFor(root, { scheduled: false }))
+  expect(off.init._meta.lembas.capabilities).toContain("scheduled")
+  expect(off.init._meta.lembas.limits.scheduled).toBe(false)
+})
+
+test("off: a scheduled session, prompt and monitor are refused with the fixed text and data.reason", async () => {
+  fake = fakeProvider([{ chunks: [delta({ content: "hi" })] }])
+  config(fake.url)
+  const root = project()
+  const { c } = await client(limitsFor(root, { scheduled: false }))
+  const err = await c.request("session/new", { cwd: root, _meta: { lembas: { scheduled: sched() } } }).catch((e) => e)
+  expect(err.message).toBe(SCHEDULED.texts.off)
+  expect(err.data).toEqual({ reason: "scheduled_off" })
+  const m = await c.request("_lembas/monitor", { cwd: root, command: "true" }).catch((e) => e)
+  expect(m.data).toEqual({ reason: "scheduled_off" })
+  // An ordinary agent chat is not affected.
+  const s: any = await c.request("session/new", { cwd: root })
+  const p = await c.request("session/prompt", { sessionId: s.sessionId, prompt: [{ type: "text", text: "go" }], _meta: { lembas: { scheduled: sched() } } }).catch((e) => e)
+  expect(p.data).toEqual({ reason: "scheduled_off" })
+})
+
+test("the mode: Edit when none chosen, Auto only when the schedule says it, Plan refused, _meta.mode ignored", async () => {
+  fake = fakeProvider([])
+  config(fake.url)
+  const root = project()
+  const { c } = await client(limitsFor(root))
+  const none: any = await c.request("session/new", { cwd: root, _meta: { mode: "auto", lembas: { scheduled: sched() } } })
+  expect(none.modes.currentModeId).toBe("edit")
+  const auto: any = await c.request("session/new", { cwd: root, _meta: { lembas: { scheduled: sched({ mode: "auto" }) } } })
+  expect(auto.modes.currentModeId).toBe("auto")
+  const plan = await c.request("session/new", { cwd: root, _meta: { lembas: { scheduled: sched({ mode: "plan" }) } } }).catch((e) => e)
+  expect(plan.message).toBe(SCHEDULED.texts.plan)
+  const { c: c2 } = await client(limitsFor(root, { maxMode: "manual" }))
+  const low: any = await c2.request("session/new", { cwd: root, _meta: { lembas: { scheduled: sched({ mode: "auto" }) } } })
+  expect(low.modes.currentModeId).toBe("manual")
+  // A folder outside the roots: the usual reason, marked.
+  const out = await c.request("session/new", { cwd: tmpdir(), _meta: { lembas: { scheduled: sched() } } }).catch((e) => e)
+  expect(out.data).toEqual({ reason: "folder" })
+})
+
+test("a scheduled turn reports ok, silent, and the final reply", async () => {
+  fake = fakeProvider([{ chunks: [delta({ content: "All the waybread is wrapped." })] }, { chunks: [delta({ content: "Nothing new.\n[silent]." })] }])
+  config(fake.url)
+  const root = project()
+  const { c } = await client(limitsFor(root))
+  const s: any = await c.request("session/new", { cwd: root, _meta: { lembas: { scheduled: sched() } } })
+  const r1: any = await c.request("session/prompt", { sessionId: s.sessionId, prompt: [{ type: "text", text: "check" }], _meta: { lembas: { turnId: "run7", scheduled: sched() } } })
+  expect(r1._meta.lembas.scheduled).toEqual({ run_id: "run7", status: "ok", text: "All the waybread is wrapped.", error: null })
+  const r2: any = await c.request("session/prompt", { sessionId: s.sessionId, prompt: [{ type: "text", text: "check" }], _meta: { lembas: { turnId: "run8", scheduled: sched({ run_id: "run8" }) } } })
+  expect(r2._meta.lembas.scheduled.status).toBe("silent")
+  expect(r2._meta.lembas.scheduled.run_id).toBe("run8")
+  // Sent again after a dropped link: the run's answer, not a second run.
+  const again: any = await c.request("session/prompt", { sessionId: s.sessionId, prompt: [{ type: "text", text: "check" }], _meta: { lembas: { turnId: "run7", scheduled: sched() } } })
+  expect(again._meta.lembas.scheduled.text).toBe("All the waybread is wrapped.")
+})
+
+test("approvals in a scheduled turn carry `scheduled` and the schedule's wait; unanswered, the model reads the fixed text", async () => {
+  fake = fakeProvider([{ chunks: [toolCall(0, "b1", "bash", JSON.stringify({ command: "rm -rf build" }))] }, { chunks: [delta({ content: "could not clean" })] }])
+  config(fake.url)
+  const root = project()
+  const { c, asked } = await client(limitsFor(root), () => ({ outcome: { outcome: "selected", optionId: "deny" } }))
+  const s: any = await c.request("session/new", { cwd: root, _meta: { lembas: { scheduled: sched({ mode: "edit" }) } } })
+  const r: any = await c.request("session/prompt", { sessionId: s.sessionId, prompt: [{ type: "text", text: "clean" }], _meta: { lembas: { turnId: "run9", scheduled: sched({ run_id: "run9", approval_wait_s: 90 }) } } })
+  expect(asked[0]._meta.lembas.scheduled).toEqual({ schedule_id: "sched1", run_id: "run9", wait_s: 90 })
+  expect(asked[0]._meta.lembas.patient).toBeUndefined()
+  expect(r._meta.lembas.scheduled.status).toBe("ok")
+  expect(deniedText(30 * 60_000)).toBe("Nobody answered this approval within 30 minutes, so it counts as denied.")
+  expect(approvalWait(undefined)).toBe(1800)
+})
+
+test("ask_user in a scheduled turn: nobody to ask", async () => {
+  const ASK = JSON.stringify({ questions: [{ header: "DB", question: "Which?", options: [{ label: "A" }, { label: "B" }] }] })
+  fake = fakeProvider([{ chunks: [toolCall(0, "q1", "ask_user", ASK)] }, { chunks: [delta({ content: "assumed A" })] }])
+  config(fake.url)
+  const root = project()
+  const { c } = await client(limitsFor(root))
+  let questioned = false
+  c.handle("_lembas/question", () => ((questioned = true), { outcome: "dismissed" }))
+  const s: any = await c.request("session/new", { cwd: root, _meta: { lembas: { scheduled: sched() } } })
+  const r: any = await c.request("session/prompt", { sessionId: s.sessionId, prompt: [{ type: "text", text: "pick" }], _meta: { lembas: { turnId: "r1", scheduled: sched() } } })
+  expect(questioned).toBe(false)
+  expect(r._meta.lembas.scheduled.text).toBe("assumed A")
+})
+
+test("_lembas/monitor: runs what the mode allows, refuses what it would ask about, hashes the output", async () => {
+  fake = fakeProvider([])
+  config(fake.url)
+  const root = project()
+  writeFileSync(join(root, "feed.txt"), "one\n")
+  const { c } = await client(limitsFor(root))
+  const r: any = await c.request("_lembas/monitor", { cwd: root, command: "cat feed.txt", mode: "edit" })
+  expect(r).toMatchObject({ exit: 0, timed_out: false, excerpt: "one\n", truncated: false, bytes: 4 })
+  expect(r.hash).toBe(monitorHash(0, "one\n"))
+  const asks = await c.request("_lembas/monitor", { cwd: root, command: "rm -rf build", mode: "edit" }).catch((e) => e)
+  expect(asks.data).toEqual({ reason: "permission" })
+  expect(asks.message).toContain("edit mode")
+  const plan = await c.request("_lembas/monitor", { cwd: root, command: "cat feed.txt", mode: "plan" }).catch((e) => e)
+  expect(plan.data).toEqual({ reason: "plan" })
+})
+
+test("runMonitor: a timeout stops the group and has no hash; a long output is hashed whole", async () => {
+  const root = project()
+  const slow = await runMonitor("echo start; sleep 5", root, 1)
+  expect(slow).toMatchObject({ hash: null, exit: null, timed_out: true })
+  expect(slow.excerpt).toBe("start\n")
+  const big = await runMonitor(`head -c ${SCHEDULED.monitor.output_max_bytes + 10} /dev/zero`, root, 10)
+  expect(big.bytes).toBe(SCHEDULED.monitor.output_max_bytes + 10)
+  expect(big.hash).toBe(monitorHash(0, new Uint8Array(SCHEDULED.monitor.output_max_bytes + 10)))
+  expect(big.truncated).toBe(true)
+}, 20_000)
+
+test("lembas remote scheduled on|off writes remote.scheduled, and --json says it", () => {
+  mkdirSync(paths.config, { recursive: true })
+  writeFileSync(join(paths.config, "config.yaml"), "remote:\n  enabled: true\n")
+  const out: string[] = []
+  expect(remoteCommand(["scheduled", "on", "--json"], { out: (l) => out.push(l), err: () => {} })).toBe(0)
+  expect(JSON.parse(out.at(-1)!).scheduled).toBe(true)
+  expect(remoteState().scheduled).toBe(true)
+  expect(remoteCommand(["scheduled", "off"], { out: () => {}, err: () => {} })).toBe(0)
+  expect(remoteState().scheduled).toBe(false)
+  expect(remoteCommand(["scheduled", "maybe"], { out: () => {}, err: () => {} })).toBe(2)
+})