Harness spec 2.0.0: background work is a job — commands and subagents alike, seen and steered
One registry per process, every job owned by a session: bash's background commands (and a command still running at its timeout, which now goes to the background instead of being killed) and subagents started with task background: true. job_list, job_output, job_stop (the former bash_list, bash_output, bash_kill) and the new job_send and report; a finished job wakes the owning session in every front end with its marked note. The TUI counts jobs, lists them (/jobs), opens a command's output, enters a subagent and messages it, moves a command to the background (ctrl+b) and asks before quitting with jobs running. The link gains the jobs capability. Subagents get a session row of their own, the spec's preamble and agent texts, and their errors reach the report. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
75 files changed
+3614
-437
No files matched your search
+66
-7
@@ -25,6 +25,7 @@ import { NOBODY_TO_ASK, type Answer, type QuestionReply, type QuestionRequest }
|
||||
import { nobodyToApprove, type PlanReply, type PlanRequest } from "../tool/plan_submit.ts"
|
||||
import { realPath } from "../permission/evaluate.ts"
|
||||
import type { Plan } from "../plan/plan.ts"
|
||||
import type { JobInfo } from "../jobs/registry.ts"
|
||||
import { PLAN_META } from "../session/engine.ts"
|
||||
import { ToolError } from "../tool/tool.ts"
|
||||
import { dropAttachments, preparePrompt, storedAttachments } from "./prompt.ts"
|
||||
@@ -40,6 +41,7 @@ import { CWD_META, REMOTE_META, Store } from "../session/store.ts"
|
||||
import type { Message } from "../provider/types.ts"
|
||||
import type { Hub } from "./hub.ts"
|
||||
import { INVALID_PARAMS, METHOD_NOT_FOUND, RpcError, type Peer } from "./rpc.ts"
|
||||
import { jobList, jobOutput, jobSend, jobStop, jobTranscript, storedJobs } from "./jobs.ts"
|
||||
|
||||
export const ACP_VERSION = 1
|
||||
/** The `_lembas/*` extension's version (LLeMbas `api/devices.py:PROTOCOL`). 2: every
|
||||
@@ -154,8 +156,10 @@ export function toolKind(name: string): string {
|
||||
if (["read", "list", "view_image"].includes(name)) return "read"
|
||||
if (["write", "edit", "multiedit", "apply_patch"].includes(name)) return "edit"
|
||||
if (["glob", "grep", "session_search", "knowledge_search", "notes_search"].includes(name)) return "search"
|
||||
if (["bash", "bash_output", "bash_kill"].includes(name)) return "execute"
|
||||
if (["bash", "job_stop"].includes(name)) return "execute"
|
||||
if (["job_output", "job_list"].includes(name)) return "read"
|
||||
if (["web_fetch", "web_search"].includes(name)) return "fetch"
|
||||
// A message between agents (job_send, report) is nothing the kinds name.
|
||||
return "other"
|
||||
}
|
||||
|
||||
@@ -266,6 +270,10 @@ export const CAPABILITIES = [
|
||||
"prototypes",
|
||||
"plan_state",
|
||||
"rewind",
|
||||
// Harness spec 2.0.0: the session's background jobs — the `job` and `job_note` events, the
|
||||
// `_lembas/jobs` and `_lembas/job/*` methods, `jobs` in `status`, `job` on an approval, and a
|
||||
// note's turn marked `machine`.
|
||||
"jobs",
|
||||
] as const
|
||||
export const STOP: Record<string, string> = { stop: "end_turn", steps: "max_turn_requests", budget: "max_tokens", cancelled: "cancelled", error: "end_turn" }
|
||||
|
||||
@@ -291,7 +299,16 @@ export function permissionParams(sessionId: string, req: AskRequest, o: { patien
|
||||
...(offerAlways ? [{ optionId: "session", kind: "allow_always", name: "Always, this session" }] : []),
|
||||
{ optionId: "deny", kind: "reject_once", name: "Deny" },
|
||||
],
|
||||
_meta: { lembas: { reason: req.decision.reason, purpose: req.purpose, always: req.decision.always ?? req.request.always, ...(o.patient ? { patient: true } : {}) } },
|
||||
_meta: {
|
||||
lembas: {
|
||||
reason: req.decision.reason,
|
||||
purpose: req.purpose,
|
||||
always: req.decision.always ?? req.request.always,
|
||||
...(o.patient ? { patient: true } : {}),
|
||||
// Which subagent asks (capability jobs), so the card can say — it may ask while the session itself is idle.
|
||||
...(req.source ? { job: { id: req.source.jobId, agent: req.source.agent, description: req.source.description } } : {}),
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -442,6 +459,11 @@ export class AcpAgent {
|
||||
peer.handle("_lembas/terminal/open", (p) => this.terminalOpen(p))
|
||||
peer.handle("_lembas/files/search", (p) => this.filesSearch(p))
|
||||
peer.handle("_lembas/commands", (p) => this.commands(p))
|
||||
peer.handle("_lembas/jobs", (p) => this.jobs(p))
|
||||
peer.handle("_lembas/job/output", (p) => this.job("output", p))
|
||||
peer.handle("_lembas/job/stop", (p) => this.job("stop", p))
|
||||
peer.handle("_lembas/job/send", (p) => this.job("send", p))
|
||||
peer.handle("_lembas/job/transcript", (p) => this.job("transcript", p))
|
||||
peer.on("_lembas/terminal/input", (p) => this.terminalInput(p))
|
||||
peer.on("_lembas/terminal/resize", (p) => this.terminalResize(p))
|
||||
peer.on("_lembas/terminal/close", (p) => this.terminalClose(String(p?.terminalId ?? "")))
|
||||
@@ -963,9 +985,9 @@ export class AcpAgent {
|
||||
if (t) {
|
||||
// The terminal knows its turns; what the hub saw of them, when it cannot be asked.
|
||||
const turns = await t.client
|
||||
.request<{ turnId?: string; queued?: string[]; lastTurnId?: string; plan?: Plan | null }>("hub/status", { sessionId: t.sessionId }, 3000)
|
||||
.catch(() => ({ ...(t.turnId ? { turnId: t.turnId } : {}), queued: [], ...(t.lastTurnId ? { lastTurnId: t.lastTurnId } : {}), plan: this.storedPlan(id) }))
|
||||
return { sessionId: id, exists: true, held: "terminal", busy: t.busy, title: t.title, ...turns, queued: turns.queued ?? [], plan: turns.plan ?? null }
|
||||
.request<{ turnId?: string; queued?: string[]; lastTurnId?: string; plan?: Plan | null; jobs?: JobInfo[] }>("hub/status", { sessionId: t.sessionId }, 3000)
|
||||
.catch(() => ({ ...(t.turnId ? { turnId: t.turnId } : {}), queued: [], ...(t.lastTurnId ? { lastTurnId: t.lastTurnId } : {}), plan: this.storedPlan(id), jobs: this.storedJobsOf(id) }))
|
||||
return { sessionId: id, exists: true, held: "terminal", busy: t.busy, title: t.title, ...turns, queued: turns.queued ?? [], plan: turns.plan ?? null, jobs: turns.jobs ?? [] }
|
||||
}
|
||||
const s = this.sessions.get(id)
|
||||
if (s)
|
||||
@@ -979,11 +1001,46 @@ export class AcpAgent {
|
||||
queued: s.turns.queued(),
|
||||
...(s.turns.last ? { lastTurnId: s.turns.last } : {}),
|
||||
plan: s.app.engine.plan ?? null,
|
||||
jobs: jobList(s.app),
|
||||
}
|
||||
const row = (this.store ??= new Store()).session(id)
|
||||
// Not open: the last turn it finished, as kept across a restart, and its plan as stored.
|
||||
// Not open: the last turn it finished, as kept across a restart, its plan and its jobs as stored.
|
||||
const last = row ? turnStore(id).load().at(-1)?.[0] : undefined
|
||||
return { sessionId: id, exists: Boolean(row), held: "none", busy: false, title: row?.title ?? "", queued: [], ...(last ? { lastTurnId: last } : {}), plan: row ? this.storedPlan(id) : null }
|
||||
return { sessionId: id, exists: Boolean(row), held: "none", busy: false, title: row?.title ?? "", queued: [], ...(last ? { lastTurnId: last } : {}), plan: row ? this.storedPlan(id) : null, jobs: row ? this.storedJobsOf(id) : [] }
|
||||
}
|
||||
|
||||
/** A stored session's jobs (capability `jobs`), or none. */
|
||||
private storedJobsOf(id: string): JobInfo[] {
|
||||
try {
|
||||
return storedJobs((this.store ??= new Store()), id)
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
}
|
||||
|
||||
/** `_lembas/jobs`: the session's background jobs, running first — the terminal's through the
|
||||
* hub when a terminal has it, as `status` goes. */
|
||||
private async jobs(p: any) {
|
||||
const t = this.terminalFor(p)
|
||||
if (t) return t.client.request("hub/job", { sessionId: t.sessionId, op: "list" }, 15_000)
|
||||
return { jobs: jobList(this.get(p).app) }
|
||||
}
|
||||
|
||||
/** `_lembas/job/output`, `/stop`, `/send`, `/transcript` (acp/jobs.ts): one job of the session. */
|
||||
private async job(op: "output" | "stop" | "send" | "transcript", p: any) {
|
||||
const t = this.terminalFor(p)
|
||||
if (t) return t.client.request("hub/job", { sessionId: t.sessionId, op, id: p?.id, since: p?.since, text: p?.text }, 15_000)
|
||||
const app = this.get(p).app
|
||||
switch (op) {
|
||||
case "output":
|
||||
return jobOutput(app, p?.id, p?.since)
|
||||
case "stop":
|
||||
return jobStop(app, p?.id)
|
||||
case "send":
|
||||
return jobSend(app, p?.id, p?.text)
|
||||
case "transcript":
|
||||
return jobTranscript(app, p?.id)
|
||||
}
|
||||
}
|
||||
|
||||
/** A stored session's plan (the engine keeps it under PLAN_META), or null. */
|
||||
@@ -1028,6 +1085,8 @@ export class AcpAgent {
|
||||
const s = this.sessions.get(id)
|
||||
if (s) {
|
||||
s.app.cancel()
|
||||
// Its jobs go with it: nothing is left running for a session that no longer exists.
|
||||
s.app.registry.stopAll(id)
|
||||
await s.queue?.catch(() => {})
|
||||
this.sessions.delete(id)
|
||||
await s.app.close().catch(() => {})
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
// The `jobs` capability of the `_lembas/*` extension (harness/acp.md): a session's background jobs
|
||||
// shown and controlled from a client. The same answers whichever process holds the session — the
|
||||
// agent (acp/agent.ts) for one it runs or opens from the store, the terminal through the hub
|
||||
// (acp/share.ts `hub/job`) for one it has open — so both call here.
|
||||
import type { App } from "../app.ts"
|
||||
import { jobRegistry, type Job, type JobInfo } from "../jobs/registry.ts"
|
||||
import type { Store } from "../session/store.ts"
|
||||
import { historyTurns } from "./agent.ts"
|
||||
import { INVALID_PARAMS, RpcError } from "./rpc.ts"
|
||||
|
||||
/** How long `job/stop` waits for the stop to take before answering with the state as it is. */
|
||||
const STOP_WAIT_MS = 2500
|
||||
|
||||
/** `_lembas/jobs`: the session's jobs, running first (the registry's order). */
|
||||
export function jobList(app: App): JobInfo[] {
|
||||
return app.jobs().map((j) => app.registry.info(j))
|
||||
}
|
||||
|
||||
function find(app: App, id: unknown): Job {
|
||||
const job = typeof id === "string" ? app.registry.get(app.jobsSessionId(), id) : undefined
|
||||
if (!job) throw new RpcError(INVALID_PARAMS, `no job ${String(id ?? "")} in this session`)
|
||||
return job
|
||||
}
|
||||
|
||||
/** `_lembas/job/output`: the output from `since` (a command's output, a subagent's transcript
|
||||
* lines), how much there is now, and the state. The model's own read mark is not moved. */
|
||||
export function jobOutput(app: App, id: unknown, since: unknown): { text: string; length: number; state: string } {
|
||||
const job = find(app, id)
|
||||
const from = typeof since === "number" && Number.isFinite(since) ? Math.max(0, Math.floor(since)) : 0
|
||||
return { text: job.output.slice(from), length: job.output.length, state: job.state }
|
||||
}
|
||||
|
||||
/** `_lembas/job/stop`: stop it (TERM then KILL for a command, cancel for a subagent), and answer
|
||||
* once the stop has taken — or, after a moment, with the state as it is. The end note wakes the
|
||||
* owner as any end does. */
|
||||
export async function jobStop(app: App, id: unknown): Promise<{ state: string }> {
|
||||
const job = find(app, id)
|
||||
if (job.state !== "running") return { state: job.state }
|
||||
app.registry.stop(job)
|
||||
const until = Date.now() + STOP_WAIT_MS
|
||||
while (Date.now() < until && job.state === "running") await Bun.sleep(25)
|
||||
return { state: job.state }
|
||||
}
|
||||
|
||||
/** `_lembas/job/send`: a message to a subagent — its inbox while it runs, a new prompt under the
|
||||
* same id once it has finished (a new `job` start event follows). */
|
||||
export function jobSend(app: App, id: unknown, text: unknown): { state: string } {
|
||||
const job = find(app, id)
|
||||
if (job.kind !== "subagent") throw new RpcError(INVALID_PARAMS, `${job.id} is a command; only a subagent takes a message`)
|
||||
const t = typeof text === "string" ? text.trim() : ""
|
||||
if (!t) throw new RpcError(INVALID_PARAMS, "nothing to send")
|
||||
if (!job.subagent) throw new RpcError(INVALID_PARAMS, `${job.id} cannot be messaged any more — an earlier process ran it`)
|
||||
if (job.state === "running") job.subagent.send(t)
|
||||
else {
|
||||
app.registry.restartSubagent(job)
|
||||
void job.subagent.restart(t)
|
||||
}
|
||||
return { state: job.state }
|
||||
}
|
||||
|
||||
/** `_lembas/job/transcript`: a subagent's own conversation as `session/history` turns. */
|
||||
export function jobTranscript(app: App, id: unknown): { id: string; turns: ReturnType<typeof historyTurns> } {
|
||||
const job = find(app, id)
|
||||
if (job.kind !== "subagent") throw new RpcError(INVALID_PARAMS, `${job.id} is a command; its output is job/output`)
|
||||
const messages = job.subagent?.engine.messages ?? (job.childSessionId && app.store ? app.store.messages(job.childSessionId) : [])
|
||||
return { id: job.id, turns: historyTurns(messages, app.engine.o.toolCtx.cwd) }
|
||||
}
|
||||
|
||||
/** The jobs of a session nobody has open, as the store keeps them (for `status` of a session
|
||||
* `held: "none"`): one that was running when its process died reads `lost`. Running first. */
|
||||
export function storedJobs(store: Store, sessionId: string): JobInfo[] {
|
||||
const list: JobInfo[] = []
|
||||
for (const row of store.jobs(sessionId)) {
|
||||
try {
|
||||
const job = JSON.parse(row.json) as Job
|
||||
if (job.state === "running") job.state = "lost"
|
||||
list.push(jobRegistry().info(job))
|
||||
} catch {}
|
||||
}
|
||||
return [...list.filter((j) => j.state === "running"), ...list.filter((j) => j.state !== "running")]
|
||||
}
|
||||
+21
-1
@@ -15,6 +15,7 @@ import { commandSet, sourcesOf } from "../session/commands.ts"
|
||||
import type { PlanReply, PlanRequest } from "../tool/plan_submit.ts"
|
||||
import type { QuestionReply, QuestionRequest } from "../tool/ask_user.ts"
|
||||
import { clampMode, effortOf, markRemote, permissionParams, planParams, planReplyFrom, questionParams, questionReplyFrom, replyFrom, STOP, type PromptResult } from "./agent.ts"
|
||||
import { jobList, jobOutput, jobSend, jobStop, jobTranscript } from "./jobs.ts"
|
||||
import type { SessionContext } from "../app.ts"
|
||||
import { sessionTurnFile, sessionTurns, type TurnLog } from "./turns.ts"
|
||||
import { dropAttachments, preparePrompt } from "./prompt.ts"
|
||||
@@ -501,7 +502,26 @@ export class Share {
|
||||
// The terminal's turns: `_lembas/session/status` for a shared session asks here.
|
||||
peer.handle("hub/status", (p) => {
|
||||
this.mine(p)
|
||||
return { ...(this.turns.running ? { turnId: this.turns.running } : {}), queued: this.turns.queued(), ...(this.turns.last ? { lastTurnId: this.turns.last } : {}), plan: this.app.engine.plan ?? null }
|
||||
return { ...(this.turns.running ? { turnId: this.turns.running } : {}), queued: this.turns.queued(), ...(this.turns.last ? { lastTurnId: this.turns.last } : {}), plan: this.app.engine.plan ?? null, jobs: jobList(this.app) }
|
||||
})
|
||||
// The session's jobs (capability `jobs`): `_lembas/jobs` and `_lembas/job/*` for a session
|
||||
// this terminal has open are answered here, by the same code the agent uses for its own.
|
||||
peer.handle("hub/job", async (p) => {
|
||||
this.mine(p)
|
||||
switch (p?.op) {
|
||||
case "list":
|
||||
return { jobs: jobList(this.app) }
|
||||
case "output":
|
||||
return jobOutput(this.app, p?.id, p?.since)
|
||||
case "stop":
|
||||
return jobStop(this.app, p?.id)
|
||||
case "send":
|
||||
return jobSend(this.app, p?.id, p?.text)
|
||||
case "transcript":
|
||||
return jobTranscript(this.app, p?.id)
|
||||
default:
|
||||
throw new RpcError(INVALID_PARAMS, `no such job operation: ${String(p?.op ?? "")}`)
|
||||
}
|
||||
})
|
||||
// The web UI is editing or regenerating from a turn: that turn and the later ones go, as /undo
|
||||
// drops one — the files stay.
|
||||
|
||||
Reference in new issue
Block a user