From b683f9e755172f234f81b4af3340646d6fb1def5 Mon Sep 17 00:00:00 2001 From: Toray Altas Date: Wed, 5 Aug 2026 11:50:26 -0400 Subject: [PATCH 1/8] feat(cli): add Zoo automation client --- apps/zoo/eslint-suppressions.json | 1 + apps/zoo/eslint.config.mjs | 4 + apps/zoo/package.json | 33 +++ apps/zoo/src/__tests__/fixtures/fake-host.mjs | 148 ++++++++++ apps/zoo/src/__tests__/options.test.ts | 17 ++ apps/zoo/src/__tests__/projection.test.ts | 34 +++ apps/zoo/src/__tests__/subprocess.test.ts | 85 ++++++ apps/zoo/src/automation.ts | 274 ++++++++++++++++++ apps/zoo/src/index.ts | 85 ++++++ apps/zoo/src/options.ts | 50 ++++ apps/zoo/src/projection.ts | 74 +++++ apps/zoo/src/render.ts | 95 ++++++ apps/zoo/src/supervisor.ts | 233 +++++++++++++++ apps/zoo/tsconfig.json | 6 + apps/zoo/tsup.config.ts | 12 + packages/types/src/api.ts | 2 + packages/types/src/events.ts | 1 + packages/zoo-host/src/__tests__/host.test.ts | 22 ++ packages/zoo-host/src/child.ts | 9 + packages/zoo-host/src/dispatcher.ts | 26 +- packages/zoo-host/src/events.ts | 162 +++++++++++ packages/zoo-host/src/index.ts | 1 + pnpm-lock.yaml | 31 ++ src/extension/api.ts | 7 + 24 files changed, 1411 insertions(+), 1 deletion(-) create mode 100644 apps/zoo/eslint-suppressions.json create mode 100644 apps/zoo/eslint.config.mjs create mode 100644 apps/zoo/package.json create mode 100644 apps/zoo/src/__tests__/fixtures/fake-host.mjs create mode 100644 apps/zoo/src/__tests__/options.test.ts create mode 100644 apps/zoo/src/__tests__/projection.test.ts create mode 100644 apps/zoo/src/__tests__/subprocess.test.ts create mode 100644 apps/zoo/src/automation.ts create mode 100644 apps/zoo/src/index.ts create mode 100644 apps/zoo/src/options.ts create mode 100644 apps/zoo/src/projection.ts create mode 100644 apps/zoo/src/render.ts create mode 100644 apps/zoo/src/supervisor.ts create mode 100644 apps/zoo/tsconfig.json create mode 100644 apps/zoo/tsup.config.ts create mode 100644 packages/zoo-host/src/events.ts diff --git a/apps/zoo/eslint-suppressions.json b/apps/zoo/eslint-suppressions.json new file mode 100644 index 0000000000..0967ef424b --- /dev/null +++ b/apps/zoo/eslint-suppressions.json @@ -0,0 +1 @@ +{} diff --git a/apps/zoo/eslint.config.mjs b/apps/zoo/eslint.config.mjs new file mode 100644 index 0000000000..694bf73664 --- /dev/null +++ b/apps/zoo/eslint.config.mjs @@ -0,0 +1,4 @@ +import { config } from "@roo-code/config-eslint/base" + +/** @type {import("eslint").Linter.Config} */ +export default [...config] diff --git a/apps/zoo/package.json b/apps/zoo/package.json new file mode 100644 index 0000000000..bdec28a356 --- /dev/null +++ b/apps/zoo/package.json @@ -0,0 +1,33 @@ +{ + "name": "@zoo-code/cli", + "version": "0.1.0", + "description": "Zoo Code command-line interface", + "type": "module", + "main": "dist/index.js", + "bin": { + "zoo": "dist/index.js" + }, + "files": [ + "dist" + ], + "scripts": { + "build": "tsup", + "check-types": "tsc --noEmit", + "lint": "eslint src --ext .ts --max-warnings=0", + "test": "pnpm build && vitest run", + "clean": "rimraf dist .turbo" + }, + "dependencies": { + "@roo-code/zoo-host": "workspace:^", + "@roo-code/zoo-protocol": "workspace:^", + "commander": "^12.1.0" + }, + "devDependencies": { + "@roo-code/config-eslint": "workspace:^", + "@roo-code/config-typescript": "workspace:^", + "@types/node": "22.20.1", + "rimraf": "6.0.1", + "tsup": "8.5.1", + "vitest": "4.1.9" + } +} diff --git a/apps/zoo/src/__tests__/fixtures/fake-host.mjs b/apps/zoo/src/__tests__/fixtures/fake-host.mjs new file mode 100644 index 0000000000..1042c003d8 --- /dev/null +++ b/apps/zoo/src/__tests__/fixtures/fake-host.mjs @@ -0,0 +1,148 @@ +import process from "node:process" +import { setImmediate } from "node:timers" + +let hostSequence = 0 +let publicSequence = 0 +const hostId = "fake-host" +const scenario = process.env.ZOO_FAKE_SCENARIO ?? "completed" +const send = (event) => process.send({ v: 1, seq: ++hostSequence, hostId, ...event }) +const stream = (event) => + send({ + type: "event", + event: { + v: 1, + seq: ++publicSequence, + timestamp: new Date().toISOString(), + hostId, + ...event, + }, + }) + +process.send({ + type: "hello", + hostId, + supportedVersions: [1], + capabilities: { + 1: ["task:start", "task:resume", "task:cancel", "history:list", "host:shutdown", "checkpoint:unavailable"], + }, + buildVersion: "test", +}) + +process.on("message", (message) => { + if (message.type === "hello.select") { + stream({ + type: "system.init", + protocol: "zoo-stream", + hostProtocolVersion: 1, + capabilities: [ + "task:start", + "task:resume", + "task:cancel", + "history:list", + "host:shutdown", + "checkpoint:unavailable", + ], + clientVersion: message.clientVersion, + hostVersion: "test", + }) + return + } + send({ type: "command.ack", commandId: message.id }) + if (message.type === "task.start") { + send({ + type: "command.done", + commandId: message.id, + data: { commandType: "task.start", task: { rootTaskId: "root-1", taskId: "root-1" } }, + }) + if (scenario === "crash") { + setImmediate(() => process.exit(70)) + return + } + if (scenario === "hang") return + if (scenario === "needs-input") { + stream({ + type: "ask.required", + rootTaskId: "root-1", + taskId: "root-1", + askId: "ask-1", + category: "followup", + subject: "Choose a target", + }) + return + } + stream({ + type: "task.result", + rootTaskId: "root-1", + taskId: "root-1", + result: { + schemaVersion: 1, + protocol: "zoo-run-result", + success: true, + outcome: "completed", + rootTaskId: "root-1", + currentTaskId: "root-1", + workspace: message.workspace, + resumable: false, + content: "finished", + elapsedMs: 5, + }, + }) + } else if (message.type === "task.cancel") { + send({ type: "command.done", commandId: message.id, data: { commandType: "task.cancel", rootTaskId: message.rootTaskId } }) + if (scenario === "hang") { + stream({ + type: "task.result", + rootTaskId: message.rootTaskId, + taskId: message.rootTaskId, + result: { + schemaVersion: 1, + protocol: "zoo-run-result", + success: false, + outcome: "cancelled", + rootTaskId: message.rootTaskId, + currentTaskId: message.rootTaskId, + workspace: process.cwd(), + resumable: true, + cancellationReason: message.reason, + elapsedMs: 5, + }, + }) + } + } else if (message.type === "history.list") { + send({ + type: "command.done", + commandId: message.id, + data: { + commandType: "history.list", + workspace: message.workspace, + tasks: [{ rootTaskId: "root-1", currentTaskId: "root-1", workspace: message.workspace, state: "interrupted" }], + }, + }) + } else if (message.type === "task.resume") { + send({ + type: "command.done", + commandId: message.id, + data: { commandType: "task.resume", task: { rootTaskId: message.rootTaskId, taskId: message.taskId } }, + }) + stream({ + type: "task.result", + rootTaskId: message.rootTaskId, + taskId: message.rootTaskId, + result: { + schemaVersion: 1, + protocol: "zoo-run-result", + success: true, + outcome: "completed", + rootTaskId: message.rootTaskId, + currentTaskId: message.taskId, + workspace: process.cwd(), + resumable: false, + content: "resumed", + elapsedMs: 5, + }, + }) + } else if (message.type === "host.shutdown") { + send({ type: "command.done", commandId: message.id, data: { commandType: "host.shutdown" } }) + setImmediate(() => process.disconnect()) + } +}) diff --git a/apps/zoo/src/__tests__/options.test.ts b/apps/zoo/src/__tests__/options.test.ts new file mode 100644 index 0000000000..bcabd4a7e2 --- /dev/null +++ b/apps/zoo/src/__tests__/options.test.ts @@ -0,0 +1,17 @@ +import { describe, expect, it } from "vitest" + +import { parseDuration, runOverrides } from "../options.js" + +describe("automation options", () => { + it("parses bounded duration units", () => { + expect(parseDuration("250ms")).toBe(250) + expect(parseDuration("2m")).toBe(120_000) + expect(() => parseDuration("2 minutes")).toThrow("Invalid duration") + }) + + it("rejects mutually exclusive provider sources", () => { + expect(() => runOverrides({ provider: "anthropic", profile: "work" })).toThrow( + "profile and provider are mutually exclusive", + ) + }) +}) diff --git a/apps/zoo/src/__tests__/projection.test.ts b/apps/zoo/src/__tests__/projection.test.ts new file mode 100644 index 0000000000..c411726261 --- /dev/null +++ b/apps/zoo/src/__tests__/projection.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from "vitest" + +import { initialProjection, reduceSession } from "../projection.js" + +const base = { v: 1 as const, seq: 1, timestamp: "2026-08-05T12:00:00.000Z", hostId: "host" } + +describe("SessionProjection", () => { + it("upserts messages by canonical identity", () => { + const created = reduceSession(initialProjection(), { + ...base, + type: "message.upsert", + rootTaskId: "root", + taskId: "root", + messageId: "message-1", + role: "assistant", + content: "hel", + complete: false, + }) + const updated = reduceSession(created, { + ...base, + seq: 2, + type: "message.upsert", + rootTaskId: "root", + taskId: "root", + messageId: "message-1", + role: "assistant", + content: "hello", + complete: true, + }) + + expect([...updated.messages.values()]).toEqual([{ role: "assistant", content: "hello", complete: true }]) + expect(created.messages.get("message-1")?.content).toBe("hel") + }) +}) diff --git a/apps/zoo/src/__tests__/subprocess.test.ts b/apps/zoo/src/__tests__/subprocess.test.ts new file mode 100644 index 0000000000..e013e1be61 --- /dev/null +++ b/apps/zoo/src/__tests__/subprocess.test.ts @@ -0,0 +1,85 @@ +import { spawn } from "node:child_process" +import path from "node:path" +import { fileURLToPath } from "node:url" + +import { describe, expect, it } from "vitest" + +const packageRoot = path.resolve(fileURLToPath(new URL("../../", import.meta.url))) +const cliPath = path.join(packageRoot, "dist/index.js") +const hostPath = fileURLToPath(new URL("./fixtures/fake-host.mjs", import.meta.url)) + +async function runCli(args: string[], scenario = "completed", signal?: NodeJS.Signals) { + return new Promise<{ stdout: string; stderr: string; code: number }>((resolve, reject) => { + const child = spawn(process.execPath, [cliPath, ...args], { + env: { ...process.env, ZOO_HOST_PATH: hostPath, ZOO_FAKE_SCENARIO: scenario }, + stdio: ["ignore", "pipe", "pipe"], + }) + let stdout = "" + let stderr = "" + child.stdout.setEncoding("utf8").on("data", (chunk: string) => (stdout += chunk)) + child.stderr.setEncoding("utf8").on("data", (chunk: string) => (stderr += chunk)) + child.once("error", reject) + child.once("close", (code) => resolve({ stdout, stderr, code: code ?? 70 })) + if (signal) setTimeout(() => child.kill(signal), 100) + }) +} + +describe("packaged zoo automation", () => { + it("negotiates with the host and writes exactly one final JSON value", async () => { + const { stdout, stderr } = await runCli(["run", "finish the task", "--format", "json", "-C", packageRoot]) + const result = JSON.parse(stdout) as { outcome: string; content: string } + + expect(result).toMatchObject({ outcome: "completed", content: "finished" }) + expect(stdout.trim().split("\n")).toHaveLength(1) + expect(stderr).toBe("") + }) + + it("settles a host crash as one runtime-failure JSON value", async () => { + const { stdout, code } = await runCli(["run", "crash", "--format", "json", "-C", packageRoot], "crash") + const lines = stdout.trim().split("\n") + + expect(code).toBe(70) + expect(lines).toHaveLength(1) + expect(JSON.parse(lines[0]!)).toMatchObject({ outcome: "failed", error: { code: "host_crashed" } }) + }) + + it("returns immediately and resumably for a safe unattended ask", async () => { + const { stdout, code } = await runCli( + ["run", "ask", "--format", "stream-json", "--approval", "safe", "-C", packageRoot], + "needs-input", + ) + const events = stdout + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { type: string; result?: { outcome: string } }) + + expect(code).toBe(3) + expect(events.at(-1)).toMatchObject({ type: "task.result", result: { outcome: "needs_input" } }) + expect(events.filter((event) => event.type === "task.result")).toHaveLength(1) + }) + + it("enforces the whole-task timeout and exit code", async () => { + const { stdout, code } = await runCli( + ["run", "wait", "--format", "json", "--timeout", "50ms", "-C", packageRoot], + "hang", + ) + + expect(code).toBe(124) + expect(JSON.parse(stdout)).toMatchObject({ outcome: "timed_out", error: { code: "task_timed_out" } }) + }) + + it("cancels canonically on SIGTERM and preserves the signal exit code", async () => { + const { stdout, code } = await runCli(["run", "wait", "--format", "json", "-C", packageRoot], "hang", "SIGTERM") + + expect(code).toBe(143) + expect(JSON.parse(stdout)).toMatchObject({ outcome: "cancelled", cancellationReason: "signal" }) + }) + + it("lists and resumes the latest workspace session", async () => { + const listed = await runCli(["sessions", "list", "--format", "json", "-C", packageRoot]) + const resumed = await runCli(["resume", "--format", "json", "-C", packageRoot]) + + expect(JSON.parse(listed.stdout)).toMatchObject([{ rootTaskId: "root-1", state: "interrupted" }]) + expect(JSON.parse(resumed.stdout)).toMatchObject({ outcome: "completed", content: "resumed" }) + }) +}) diff --git a/apps/zoo/src/automation.ts b/apps/zoo/src/automation.ts new file mode 100644 index 0000000000..21f7e6c725 --- /dev/null +++ b/apps/zoo/src/automation.ts @@ -0,0 +1,274 @@ +import fs from "node:fs" +import os from "node:os" +import path from "node:path" +import { fileURLToPath } from "node:url" + +import { + exitCodeFor, + ZOO_PUBLIC_SCHEMA_VERSION, + zooRunResultSchema, + type HostCommand, + type ZooErrorCode, + type ZooRunResult, + type ZooStreamEvent, +} from "@roo-code/zoo-protocol" + +import { runOverrides, type OutputFormat, type SharedOptions } from "./options.js" +import { initialProjection, reduceSession } from "./projection.js" +import { createRenderer } from "./render.js" +import { defaultStorageRoot, HostClient } from "./supervisor.js" + +type AutomationOptions = Omit & { + format: OutputFormat + quiet: boolean + workspace: string + timeout?: number +} + +export async function runAutomation( + request: { type: "run"; prompt: string } | { type: "resume"; taskId?: string }, + options: AutomationOptions, +): Promise { + if (options.format === "stream-json" && options.quiet) throw new Error("--quiet cannot be used with stream-json") + if (options.approval === "interactive") throw new Error("--approval interactive cannot be used in automation") + const overrides = runOverrides({ ...options, approval: options.approval ?? "safe" }) + const startedAt = Date.now() + const storageRoot = options.ephemeral + ? fs.mkdtempSync(path.join(os.tmpdir(), "zoo-")) + : path.join(defaultStorageRoot(), "state") + fs.mkdirSync(storageRoot, { recursive: true }) + const renderer = createRenderer(options.format, options.quiet) + let projection = initialProjection() + let settled = false + let settleResult: ((result: ZooRunResult) => void) | undefined + const resultPromise = new Promise((resolve) => { + settleResult = (result) => { + if (settled) return + settled = true + resolve(result) + } + }) + let lastEvent: ZooStreamEvent | undefined + const makeResult = ( + outcome: "needs_input" | "cancelled" | "timed_out" | "failed", + rootTaskId: string, + input: { currentTaskId?: string; resumable: boolean; code?: ZooErrorCode; message?: string; content?: string }, + ): ZooRunResult => + zooRunResultSchema.parse({ + schemaVersion: ZOO_PUBLIC_SCHEMA_VERSION, + protocol: "zoo-run-result", + success: false, + outcome, + rootTaskId, + currentTaskId: input.currentTaskId, + workspace: options.workspace, + resumable: input.resumable, + content: input.content, + error: + outcome === "failed" || outcome === "timed_out" + ? { code: input.code ?? "task_failed", message: input.message ?? "Task failed", kind: "runtime" } + : undefined, + elapsedMs: Date.now() - startedAt, + cancellationReason: outcome === "cancelled" ? "signal" : undefined, + }) + const client = new HostClient({ + workspace: options.workspace, + storageRoot, + extensionRoot: process.env.ZOO_EXTENSION_PATH ?? fileURLToPath(new URL("../../../src/dist", import.meta.url)), + debug: options.debug, + onEvent(event) { + const normalizedEvent: ZooStreamEvent = + event.type === "task.result" && + event.result.outcome === "cancelled" && + event.result.cancellationReason === "timeout" + ? { + ...event, + result: makeResult("timed_out", event.result.rootTaskId, { + currentTaskId: event.result.currentTaskId, + resumable: event.result.resumable, + code: "task_timed_out", + message: "Task deadline exceeded", + }), + } + : event + lastEvent = normalizedEvent + projection = reduceSession(projection, normalizedEvent) + renderer.event(normalizedEvent) + if (normalizedEvent.type === "task.result") settleResult?.(normalizedEvent.result) + if (normalizedEvent.type === "ask.required" && options.approval !== "auto") { + settleResult?.( + makeResult("needs_input", normalizedEvent.rootTaskId, { + currentTaskId: normalizedEvent.taskId, + resumable: true, + content: normalizedEvent.subject, + }), + ) + } + }, + }) + let signal: "SIGINT" | "SIGTERM" | undefined + let notifySignal: ((value: "SIGINT" | "SIGTERM") => void) | undefined + const signalPromise = new Promise<"SIGINT" | "SIGTERM">((resolve) => (notifySignal = resolve)) + let rootTaskId: string | undefined + const onSignal = (received: "SIGINT" | "SIGTERM") => { + if (signal) { + void client.stop() + if (rootTaskId) { + settleResult?.( + makeResult("cancelled", rootTaskId, { currentTaskId: projection.currentTaskId, resumable: false }), + ) + } + return + } + signal = received + notifySignal?.(received) + } + process.on("SIGINT", onSignal) + process.on("SIGTERM", onSignal) + let timeout: NodeJS.Timeout | undefined + let notifyTimeout: (() => void) | undefined + const timeoutPromise = new Promise<"timeout">((resolve) => (notifyTimeout = () => resolve("timeout"))) + if (options.timeout !== undefined) timeout = setTimeout(() => notifyTimeout?.(), options.timeout) + try { + const startup = await Promise.race([ + client.start().then(() => "started" as const), + timeoutPromise, + signalPromise.then(() => "signal" as const), + ]) + if (startup === "timeout") throw new Error("Task deadline exceeded during host startup") + if (startup === "signal") throw new Error(`Received ${signal} during host startup`) + let command: HostCommand extends infer Command + ? Command extends HostCommand + ? Omit + : never + : never + if (request.type === "run") { + command = { + type: "task.start", + workspace: options.workspace, + prompt: request.prompt, + overrides, + } + } else { + let taskId = request.taskId + if (!taskId) { + const history = await client.command({ type: "history.list", workspace: options.workspace }) + if (history.data.commandType !== "history.list" || history.data.tasks.length === 0) { + throw new Error("No session exists for this workspace") + } + taskId = history.data.tasks[0]!.rootTaskId + } + command = { type: "task.resume", taskId, rootTaskId: taskId, overrides } + } + const accepted = await client.command(command) + if (accepted.data.commandType !== "task.start" && accepted.data.commandType !== "task.resume") { + throw new Error("Host returned an invalid task acceptance") + } + rootTaskId = accepted.data.task.rootTaskId + if (signal) void client.command({ type: "task.cancel", rootTaskId, reason: "signal" }).catch(() => undefined) + const localSettlement = Promise.race([ + timeoutPromise.then(async () => { + await client + .command({ type: "task.cancel", rootTaskId: rootTaskId!, reason: "timeout" }, 5_000) + .catch(() => undefined) + return makeResult("timed_out", rootTaskId!, { + currentTaskId: projection.currentTaskId, + resumable: true, + code: "task_timed_out", + message: "Task deadline exceeded", + }) + }), + signalPromise.then(async () => { + await client + .command({ type: "task.cancel", rootTaskId: rootTaskId!, reason: "signal" }, 5_000) + .catch(() => undefined) + await new Promise((resolve) => setTimeout(resolve, 1_000)) + return makeResult("cancelled", rootTaskId!, { + currentTaskId: projection.currentTaskId, + resumable: false, + }) + }), + renderer.outputClosed.then(async () => { + await client + .command({ type: "task.cancel", rootTaskId: rootTaskId!, reason: "user" }, 5_000) + .catch(() => undefined) + return makeResult("failed", rootTaskId!, { + currentTaskId: projection.currentTaskId, + resumable: false, + code: "output_closed", + message: "Output stream closed", + }) + }), + client.failed.catch((error: Error) => + makeResult("failed", rootTaskId!, { + currentTaskId: projection.currentTaskId, + resumable: false, + code: "host_crashed", + message: error.message, + }), + ), + ]) + const result = await Promise.race([resultPromise, localSettlement]) + if (options.format === "stream-json" && lastEvent?.type !== "task.result") { + const event: ZooStreamEvent = { + v: 1, + seq: (lastEvent?.seq ?? 0) + 1, + timestamp: new Date().toISOString(), + hostId: lastEvent?.hostId ?? "parent", + type: "task.result", + rootTaskId: result.rootTaskId, + taskId: result.rootTaskId, + result, + } + renderer.event(event) + } + renderer.result(result) + const failedCode = + result.error?.code === "task_timed_out" || result.error?.code === "cleanup_timed_out" + ? "task_failed" + : (result.error?.code ?? "task_failed") + return exitCodeFor( + result.outcome === "failed" + ? { outcome: "failed", errorCode: failedCode } + : result.outcome === "cancelled" + ? { outcome: "cancelled", signal } + : result.outcome === "timed_out" + ? { outcome: "timed_out", errorCode: "task_timed_out" } + : { outcome: result.outcome }, + ) + } finally { + if (timeout) clearTimeout(timeout) + process.removeListener("SIGINT", onSignal) + process.removeListener("SIGTERM", onSignal) + await client.stop() + renderer.dispose() + if (options.ephemeral) fs.rmSync(storageRoot, { recursive: true, force: true }) + } +} + +export async function listSessions(options: { + workspace: string + format: "text" | "json" + ephemeral?: boolean + debug?: boolean +}) { + if (options.ephemeral) return options.format === "json" ? process.stdout.write("[]\n") : undefined + const client = new HostClient({ + workspace: options.workspace, + storageRoot: path.join(defaultStorageRoot(), "state"), + extensionRoot: process.env.ZOO_EXTENSION_PATH ?? fileURLToPath(new URL("../../../src/dist", import.meta.url)), + debug: options.debug, + onEvent: () => undefined, + }) + try { + await client.start() + const response = await client.command({ type: "history.list", workspace: options.workspace }) + if (response.data.commandType !== "history.list") throw new Error("Host returned an invalid history response") + if (options.format === "json") process.stdout.write(`${JSON.stringify(response.data.tasks)}\n`) + else + for (const task of response.data.tasks) + process.stdout.write(`${task.rootTaskId}\t${task.state}\t${task.currentTaskId}\n`) + } finally { + await client.stop() + } +} diff --git a/apps/zoo/src/index.ts b/apps/zoo/src/index.ts new file mode 100644 index 0000000000..95c9456dc6 --- /dev/null +++ b/apps/zoo/src/index.ts @@ -0,0 +1,85 @@ +import { Command, Option } from "commander" + +import { listSessions, runAutomation } from "./automation.js" +import { parseDuration, resolveWorkspace, type OutputFormat, type SharedOptions } from "./options.js" + +const program = new Command().name("zoo").description("Run Zoo Code from the terminal").version("0.1.0") + +async function readPipedStdin(): Promise { + if (process.stdin.isTTY) return "" + let input = "" + process.stdin.setEncoding("utf8") + for await (const chunk of process.stdin) input += chunk + return input.trim() +} + +function shared(command: Command): Command { + return command + .option("-C, --cwd ", "effective workspace", process.cwd()) + .option("--provider ", "ephemeral provider selection") + .option("--profile ", "ephemeral provider profile") + .option("--model ", "ephemeral model selection") + .option("--mode ", "ephemeral mode selection") + .addOption( + new Option("--reasoning-effort ").choices([ + "disabled", + "none", + "minimal", + "low", + "medium", + "high", + "xhigh", + "max", + ]), + ) + .addOption(new Option("--approval ").choices(["interactive", "safe", "auto"])) + .option("--ephemeral", "remove invocation state after cleanup") + .option("--timeout ", "whole-task deadline") + .option("--debug", "write redacted diagnostics to stderr") +} + +function automation(command: Command): Command { + return shared(command) + .addOption(new Option("--format ").choices(["text", "json", "stream-json"]).default("text")) + .option("--quiet", "print only the final answer or result") +} + +function normalized(options: SharedOptions & { format: OutputFormat; quiet: boolean }) { + return { ...options, workspace: resolveWorkspace(options.cwd), timeout: parseDuration(options.timeout) } +} + +automation(program.command("run [prompt...]").description("run one task without an interactive UI")).action( + async (words: string[] | undefined, options: SharedOptions & { format: OutputFormat; quiet: boolean }) => { + const positional = words?.join(" ").trim() + const piped = await readPipedStdin() + if (positional && piped) throw new Error("Use either a positional prompt or piped stdin, not both") + const prompt = positional || piped + if (!prompt) throw new Error("zoo run requires a prompt or piped stdin") + process.exitCode = await runAutomation({ type: "run", prompt }, normalized(options)) + }, +) + +automation(program.command("resume [task-id]").description("resume a workspace session")).action( + async (taskId: string | undefined, options: SharedOptions & { format: OutputFormat; quiet: boolean }) => { + process.exitCode = await runAutomation({ type: "resume", taskId }, normalized(options)) + }, +) + +const sessions = program.command("sessions").description("inspect canonical sessions") +shared( + sessions.command("list").addOption(new Option("--format ").choices(["text", "json"]).default("text")), +).action(async (options: SharedOptions & { format: "text" | "json" }) => { + await listSessions({ ...options, workspace: resolveWorkspace(options.cwd) }) +}) + +program.action(() => { + if (!process.stdin.isTTY || !process.stdout.isTTY) throw new Error("Interactive Zoo requires TTY stdin and stdout") + throw new Error("Interactive Zoo is not available in this build") +}) + +program.showSuggestionAfterError().showHelpAfterError() +await program.parseAsync().catch((error) => { + const message = error instanceof Error ? error.message : String(error) + process.stderr.write(`zoo: ${message}\n`) + process.exitCode = message.includes("SIGINT") ? 130 : message.includes("SIGTERM") ? 143 : 2 +}) diff --git a/apps/zoo/src/options.ts b/apps/zoo/src/options.ts new file mode 100644 index 0000000000..b1115cf622 --- /dev/null +++ b/apps/zoo/src/options.ts @@ -0,0 +1,50 @@ +import fs from "node:fs" +import path from "node:path" + +import { runOverridesSchema, type RunOverrides } from "@roo-code/zoo-protocol" + +export type OutputFormat = "text" | "json" | "stream-json" + +export type SharedOptions = { + cwd: string + provider?: string + profile?: string + model?: string + mode?: string + reasoningEffort?: RunOverrides["reasoningEffort"] + approval?: RunOverrides["approval"] + ephemeral?: boolean + timeout?: string + debug?: boolean +} + +export function resolveWorkspace(input: string): string { + const resolved = fs.realpathSync(path.resolve(input)) + if (!fs.statSync(resolved).isDirectory()) throw new Error(`Workspace is not a directory: ${input}`) + return resolved +} + +export function parseDuration(value: string | undefined): number | undefined { + if (value === undefined) return undefined + const match = /^(\d+)(ms|s|m|h)$/.exec(value) + if (!match) throw new Error(`Invalid duration: ${value}`) + const amount = Number(match[1]) + const multiplier = { ms: 1, s: 1_000, m: 60_000, h: 3_600_000 }[match[2] as "ms" | "s" | "m" | "h"] + const duration = amount * multiplier + if (!Number.isSafeInteger(duration) || duration <= 0) throw new Error(`Invalid duration: ${value}`) + return duration +} + +export function runOverrides( + options: Pick, +): RunOverrides | undefined { + const parsed = runOverridesSchema.parse({ + provider: options.provider, + profile: options.profile, + model: options.model, + mode: options.mode, + reasoningEffort: options.reasoningEffort, + approval: options.approval, + }) + return Object.values(parsed).some((value) => value !== undefined) ? parsed : undefined +} diff --git a/apps/zoo/src/projection.ts b/apps/zoo/src/projection.ts new file mode 100644 index 0000000000..a215d91e00 --- /dev/null +++ b/apps/zoo/src/projection.ts @@ -0,0 +1,74 @@ +import type { ZooRunResult, ZooStreamEvent } from "@roo-code/zoo-protocol" + +export type ProjectedMessage = { role: "assistant" | "user" | "reasoning"; content: string; complete: boolean } +export type SessionProjection = { + rootTaskId?: string + currentTaskId?: string + tasks: ReadonlyMap + messages: ReadonlyMap + pendingAsks: ReadonlyMap + tools: ReadonlyMap + usage?: { inputTokens?: number; outputTokens?: number; cacheReads?: number; cacheWrites?: number } + cost?: number + result?: ZooRunResult +} + +export const initialProjection = (): SessionProjection => ({ + tasks: new Map(), + messages: new Map(), + pendingAsks: new Map(), + tools: new Map(), +}) + +export function reduceSession(state: SessionProjection, event: ZooStreamEvent): SessionProjection { + const task = "taskId" in event ? { rootTaskId: event.rootTaskId, currentTaskId: event.taskId } : {} + switch (event.type) { + case "task.created": + case "task.delegated": + case "task.lifecycle": { + const tasks = new Map(state.tasks) + const current = tasks.get(event.taskId) ?? {} + tasks.set(event.taskId, { + ...current, + ...(event.type === "task.delegated" ? { parentTaskId: event.parentTaskId } : {}), + ...(event.type === "task.lifecycle" ? { state: event.state } : {}), + }) + return { ...state, ...task, tasks } + } + case "message.upsert": { + const messages = new Map(state.messages) + messages.set(event.messageId, { role: event.role, content: event.content, complete: event.complete }) + return { ...state, ...task, messages } + } + case "ask.required": { + const pendingAsks = new Map(state.pendingAsks) + pendingAsks.set(event.askId, { taskId: event.taskId, category: event.category, subject: event.subject }) + return { ...state, ...task, pendingAsks } + } + case "ask.resolved": + case "ask.abandoned": { + const pendingAsks = new Map(state.pendingAsks) + pendingAsks.delete(event.askId) + return { ...state, ...task, pendingAsks } + } + case "tool.started": + case "tool.updated": + case "tool.completed": + case "tool.failed": { + const tools = new Map(state.tools) + tools.set(event.toolCallId, { + name: event.name, + state: + event.type === "tool.failed" ? "failed" : event.type === "tool.completed" ? "completed" : "active", + output: event.output, + }) + return { ...state, ...task, tools } + } + case "usage.updated": + return { ...state, ...task, usage: event.usage, cost: event.cost } + case "task.result": + return { ...state, ...task, result: event.result } + default: + return { ...state, ...task } + } +} diff --git a/apps/zoo/src/render.ts b/apps/zoo/src/render.ts new file mode 100644 index 0000000000..805ec7140c --- /dev/null +++ b/apps/zoo/src/render.ts @@ -0,0 +1,95 @@ +import type { ZooRunResult, ZooStreamEvent } from "@roo-code/zoo-protocol" + +export type Renderer = { + event: (event: ZooStreamEvent) => void + result: (result: ZooRunResult) => void + outputClosed: Promise + dispose: () => void +} + +export function createRenderer(format: "text" | "json" | "stream-json", quiet: boolean): Renderer { + let closed = false + let lastAssistantContent: string | undefined + let closeOutput: (() => void) | undefined + const outputClosed = new Promise((resolve) => (closeOutput = resolve)) + const onError = (error: NodeJS.ErrnoException) => { + if (error.code !== "EPIPE") throw error + closed = true + closeOutput?.() + } + process.stdout.on("error", onError) + const write = (value: string): void => { + if (!closed) process.stdout.write(value) + } + const dispose = () => process.stdout.removeListener("error", onError) + if (format === "json") { + return { + event: () => undefined, + result: (result) => write(`${JSON.stringify(result)}\n`), + outputClosed, + dispose, + } + } + if (format === "stream-json") { + return { event: (event) => write(`${JSON.stringify(event)}\n`), result: () => undefined, outputClosed, dispose } + } + return { + event(event) { + if (quiet && event.type !== "task.result" && event.type !== "system.warning") return + switch (event.type) { + case "system.init": + if (!quiet) write(`Zoo ${event.clientVersion} · ${event.capabilities.join(", ")}\n`) + break + case "system.warning": + write(`warning: ${event.message}\n`) + break + case "message.upsert": + if (event.complete && event.role !== "user") { + write(`${event.content}\n`) + if (event.role === "assistant") lastAssistantContent = event.content + } + break + case "task.delegated": + if (!quiet) write(`Delegated to ${event.childTaskId}\n`) + break + case "ask.required": + if (!quiet) write(`Input required: ${event.subject}\n`) + break + case "tool.started": + if (!quiet) write(`Tool ${event.name} started\n`) + break + case "tool.completed": + if (!quiet) write(`Tool ${event.name} completed${event.output ? `: ${event.output}` : ""}\n`) + break + case "tool.failed": + write(`Tool ${event.name} failed: ${event.error.message}\n`) + break + case "terminal.output": + if (!quiet) write(event.delta) + break + case "terminal.status": + if (!quiet && (event.state === "exited" || event.state === "killed")) { + write(`\nCommand ${event.state} (${event.exitCode ?? "unknown"})\n`) + } + break + case "mcp.started": + if (!quiet) write(`MCP ${event.server}/${event.operation} started\n`) + break + case "mcp.completed": + if (!quiet) write(`MCP ${event.server}/${event.operation} completed\n`) + break + case "mcp.failed": + write(`MCP ${event.server}/${event.operation} failed: ${event.error.message}\n`) + break + } + }, + result(result) { + if (result.content && (quiet || result.content !== lastAssistantContent)) { + write(`${result.content}\n`) + } + if (!result.success) write(`${result.error?.message ?? result.outcome}\n`) + }, + outputClosed, + dispose, + } +} diff --git a/apps/zoo/src/supervisor.ts b/apps/zoo/src/supervisor.ts new file mode 100644 index 0000000000..cc65689151 --- /dev/null +++ b/apps/zoo/src/supervisor.ts @@ -0,0 +1,233 @@ +import { fork, type ChildProcess } from "node:child_process" +import { randomUUID } from "node:crypto" +import os from "node:os" +import path from "node:path" +import { fileURLToPath } from "node:url" + +import { + createHostEventStreamParser, + hostCommandSchema, + hostHelloSchema, + negotiateProtocol, + parentHelloSchema, + validateNegotiatedStreamSession, + ZOO_HOST_PROTOCOL_VERSION, + type HostCommand, + type HostEvent, + type HostHello, + type ParentHello, + type ZooCapability, + type ZooStreamEvent, + redactText, +} from "@roo-code/zoo-protocol" + +type PendingCommand = { + acknowledged: boolean + resolve: (event: Extract) => void + reject: (error: Error) => void +} +type OutboundHostCommand = HostCommand extends infer Command + ? Command extends HostCommand + ? Omit + : never + : never +type HostClientOptions = { + workspace: string + storageRoot: string + extensionRoot: string + onEvent: (event: ZooStreamEvent) => void + debug?: boolean +} + +const requiredCapabilities: ZooCapability[] = [ + "task:start", + "task:resume", + "task:cancel", + "history:list", + "host:shutdown", +] + +export class HostClient { + private child: ChildProcess | undefined + private parser: ReturnType | undefined + private readonly pending = new Map() + private failure: Error | undefined + private rejectFailure: ((error: Error) => void) | undefined + public readonly failed = new Promise((_, reject) => (this.rejectFailure = reject)) + private hello: HostHello | undefined + private selection: ParentHello | undefined + private resolveInitialized: (() => void) | undefined + private initialized = new Promise((resolve) => (this.resolveInitialized = resolve)) + private lastHeartbeat = Date.now() + private watchdog: NodeJS.Timeout | undefined + + constructor(private readonly options: HostClientOptions) {} + + public async start(): Promise { + const hostPath = + process.env.ZOO_HOST_PATH ?? + fileURLToPath(new URL("../../../packages/zoo-host/dist/child.js", import.meta.url)) + const child = fork(hostPath, [], { + stdio: ["ignore", "pipe", "pipe", "ipc"], + env: { + ...process.env, + ZOO_HOST_CONFIG: JSON.stringify({ + extensionRoot: this.options.extensionRoot, + workspaceRoot: this.options.workspace, + storageRoot: this.options.storageRoot, + appRoot: this.options.extensionRoot, + buildVersion: "0.1.0", + }), + }, + }) + this.child = child + child.stdout?.resume() + child.stderr?.on("data", (chunk: Buffer) => { + if (this.options.debug) process.stderr.write(redactText(chunk.subarray(0, 16 * 1024).toString("utf8"))) + }) + child.once("exit", (code, signal) => this.fail(new Error(`Zoo host exited (${signal ?? code ?? "unknown"})`))) + child.once("error", (error) => this.fail(error)) + + const hello = await Promise.race([this.waitForHello(child, 15_000), this.failed]) + this.hello = hello + const negotiation = negotiateProtocol(hello, [ZOO_HOST_PROTOCOL_VERSION], requiredCapabilities) + if (!negotiation.ok) throw new Error(negotiation.message) + this.parser = createHostEventStreamParser({ hostId: hello.hostId }) + child.on("message", (message) => this.receive(message)) + this.selection = parentHelloSchema.parse({ + type: "hello.select", + version: negotiation.version, + clientVersion: "0.1.0", + requiredCapabilities, + }) + child.send(this.selection) + let initializationTimer: NodeJS.Timeout | undefined + try { + await Promise.race([ + this.initialized, + this.failed, + new Promise((_, reject) => { + initializationTimer = setTimeout( + () => reject(new Error("Zoo host initialization timed out")), + 30_000, + ) + }), + ]) + } finally { + if (initializationTimer) clearTimeout(initializationTimer) + } + this.lastHeartbeat = Date.now() + this.watchdog = setInterval(() => { + if (Date.now() - this.lastHeartbeat > 5_000) this.fail(new Error("Zoo host heartbeat timed out")) + }, 1_000) + this.watchdog.unref() + } + + public async command(input: OutboundHostCommand, timeoutMs = 15_000) { + if (!this.child?.connected) throw this.failure ?? new Error("Zoo host is unavailable") + const id = randomUUID() + const command = hostCommandSchema.parse({ v: ZOO_HOST_PROTOCOL_VERSION, id, ...input }) + return new Promise>((resolve, reject) => { + const timer = setTimeout(() => { + this.pending.delete(id) + reject(new Error(`Host command timed out: ${command.type}`)) + }, timeoutMs) + this.pending.set(id, { + acknowledged: false, + resolve: (event) => { + clearTimeout(timer) + resolve(event) + }, + reject: (error) => { + clearTimeout(timer) + reject(error) + }, + }) + this.child!.send(command, (error) => { + if (error) this.pending.get(id)?.reject(error) + }) + }) + } + + public async stop(): Promise { + const child = this.child + if (!child) return + if (this.watchdog) clearInterval(this.watchdog) + this.watchdog = undefined + if (child.connected && !this.failure) + await this.command({ type: "host.shutdown" }, 5_000).catch(() => undefined) + child.disconnect() + await new Promise((resolve) => { + if (child.exitCode !== null || child.signalCode !== null) return resolve() + const timer = setTimeout(() => { + child.kill("SIGKILL") + resolve() + }, 2_000) + child.once("exit", () => { + clearTimeout(timer) + resolve() + }) + }) + } + + private waitForHello(child: ChildProcess, timeoutMs: number) { + return new Promise>((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("Zoo host startup timed out")), timeoutMs) + child.once("message", (message) => { + clearTimeout(timer) + try { + resolve(hostHelloSchema.parse(message)) + } catch (error) { + reject(error) + } + }) + }) + } + + private receive(input: unknown): void { + try { + for (const event of this.parser?.push(input) ?? []) { + if (event.type === "host.heartbeat") this.lastHeartbeat = Date.now() + if (event.type === "event") { + if (event.event.type === "system.init") { + if (!this.hello || !this.selection) + throw new Error("Host initialized before protocol negotiation") + const validation = validateNegotiatedStreamSession(this.hello, this.selection, [event.event]) + if (!validation.ok) throw new Error(validation.message) + this.resolveInitialized?.() + } + this.options.onEvent(event.event) + } + if (event.type === "command.ack") { + const pending = this.pending.get(event.commandId) + if (!pending || pending.acknowledged) throw new Error(`Invalid ACK for command ${event.commandId}`) + pending.acknowledged = true + } + if (event.type === "command.done") { + const pending = this.pending.get(event.commandId) + if (!pending?.acknowledged) throw new Error(`DONE preceded ACK for command ${event.commandId}`) + pending.resolve(event) + this.pending.delete(event.commandId) + } + if (event.type === "command.error") { + const pending = this.pending.get(event.commandId) + if (!pending?.acknowledged) throw new Error(`ERROR preceded ACK for command ${event.commandId}`) + pending.reject(new Error(`${event.error.code}: ${event.error.message}`)) + this.pending.delete(event.commandId) + } + } + } catch (error) { + this.fail(error instanceof Error ? error : new Error(String(error))) + } + } + + private fail(error: Error): void { + if (this.failure) return + this.failure = error + this.rejectFailure?.(error) + for (const pending of this.pending.values()) pending.reject(error) + this.pending.clear() + } +} + +export const defaultStorageRoot = () => path.join(os.homedir(), ".zoo") diff --git a/apps/zoo/tsconfig.json b/apps/zoo/tsconfig.json new file mode 100644 index 0000000000..99027cfa10 --- /dev/null +++ b/apps/zoo/tsconfig.json @@ -0,0 +1,6 @@ +{ + "extends": "@roo-code/config-typescript/base.json", + "compilerOptions": { "outDir": "dist" }, + "include": ["src", "*.config.ts"], + "exclude": ["node_modules"] +} diff --git a/apps/zoo/tsup.config.ts b/apps/zoo/tsup.config.ts new file mode 100644 index 0000000000..f016edb917 --- /dev/null +++ b/apps/zoo/tsup.config.ts @@ -0,0 +1,12 @@ +import { defineConfig } from "tsup" + +export default defineConfig({ + entry: ["src/index.ts"], + format: ["esm"], + clean: true, + sourcemap: true, + target: "node22", + platform: "node", + banner: { js: "#!/usr/bin/env node" }, + noExternal: ["@roo-code/zoo-protocol"], +}) diff --git a/packages/types/src/api.ts b/packages/types/src/api.ts index 3ce3594e26..cea1d2ce1c 100644 --- a/packages/types/src/api.ts +++ b/packages/types/src/api.ts @@ -44,6 +44,7 @@ export type HeadlessTaskResult = { currentTaskId: string outcome: "completed" | "cancelled" | "failed" resumable: boolean + cancellationReason?: "user" | "signal" | "timeout" content?: string error?: { code: "task_failed" | "cancel_failed" | "shutdown"; message: string } tokenUsage?: TokenUsage @@ -71,6 +72,7 @@ export interface RooCodeAPI extends EventEmitter { }): Promise getHeadlessTaskResult(rootTaskId: string): Promise waitForHeadlessTaskResult(rootTaskId: string): Promise + listHeadlessTaskHistory(workspace: string): Promise shutdownHeadless(): Promise /** * Starts a new task with an optional initial message and images. diff --git a/packages/types/src/events.ts b/packages/types/src/events.ts index 62c792e842..abcdba29ea 100644 --- a/packages/types/src/events.ts +++ b/packages/types/src/events.ts @@ -149,6 +149,7 @@ export const rooCodeEventsSchema = z.object({ currentTaskId: z.string(), outcome: z.enum(["completed", "cancelled", "failed"]), resumable: z.boolean(), + cancellationReason: z.enum(["user", "signal", "timeout"]).optional(), content: z.string().optional(), historyItem: historyItemSchema.optional(), }), diff --git a/packages/zoo-host/src/__tests__/host.test.ts b/packages/zoo-host/src/__tests__/host.test.ts index d3be05e929..02ef07505d 100644 --- a/packages/zoo-host/src/__tests__/host.test.ts +++ b/packages/zoo-host/src/__tests__/host.test.ts @@ -171,4 +171,26 @@ exports.deactivate = async () => {} { seq: 2, type: "command.error", error: { code: "task_failed" } }, ]) }) + + it("lists canonical root sessions for only the pinned workspace", async () => { + const sent: unknown[] = [] + const transport = new HostTransport("host-1", async (message) => void sent.push(message)) + const api = { + listHeadlessTaskHistory: vi.fn().mockResolvedValue([ + { id: "root", rootTaskId: "root", workspace: "/workspace", status: "interrupted" }, + { id: "child", rootTaskId: "root", parentTaskId: "root", workspace: "/workspace", status: "completed" }, + ]), + } as never + const dispatcher = new HostCommandDispatcher(api, transport, "/workspace") + await dispatcher.dispatch({ v: 1, id: "cmd-1", type: "history.list", workspace: "/workspace" }) + + expect(sent).toMatchObject([ + { seq: 1, type: "command.ack" }, + { + seq: 2, + type: "command.done", + data: { commandType: "history.list", tasks: [{ rootTaskId: "root", state: "interrupted" }] }, + }, + ]) + }) }) diff --git a/packages/zoo-host/src/child.ts b/packages/zoo-host/src/child.ts index 83c389ded8..2dece3d376 100644 --- a/packages/zoo-host/src/child.ts +++ b/packages/zoo-host/src/child.ts @@ -10,6 +10,7 @@ import { import { activateExtensionHost } from "./bootstrap.js" import { HostCommandDispatcher } from "./dispatcher.js" +import { HostEventBridge } from "./events.js" import { validateHostRoots, type HostRoots } from "./roots.js" import { createSystemVaultBackend, VaultSecretStorage } from "./security.js" import { HostTransport } from "./transport.js" @@ -70,6 +71,14 @@ export async function runChild(config: ChildConfig): Promise { const secretStorage = new VaultSecretStorage(createSystemVaultBackend()) const extension = await activateExtensionHost(roots, secretStorage) const transport = new HostTransport(hostId, sendProcessMessage) + const bridge = new HostEventBridge( + extension.api, + transport, + roots.workspaceRoot, + selection.clientVersion, + config.buildVersion, + ) + await bridge.initialize() const dispatcher = new HostCommandDispatcher(extension.api, transport, roots.workspaceRoot) transport.startHeartbeat() process.on("message", (message) => { diff --git a/packages/zoo-host/src/dispatcher.ts b/packages/zoo-host/src/dispatcher.ts index e3d36b167e..e9ada4c3e9 100644 --- a/packages/zoo-host/src/dispatcher.ts +++ b/packages/zoo-host/src/dispatcher.ts @@ -48,6 +48,11 @@ export class HostCommandDispatcher { return { commandType: command.type, task } } case "task.resume": { + const history = await this.api.getTaskHistoryItem(command.taskId) + if (!history) throw new Error(`Unknown session ${command.taskId}`) + if (history.workspace !== this.workspace) { + throw new Error(`Session ${command.taskId} belongs to workspace ${history.workspace ?? "unknown"}`) + } const task = await this.api.resumeHeadlessTask(command.taskId, command.overrides) this.activeRootTaskId = task.rootTaskId return { commandType: command.type, task } @@ -78,7 +83,26 @@ export class HostCommandDispatcher { await this.api.shutdownHeadless() return { commandType: command.type } case "history.list": - return { commandType: command.type, workspace: command.workspace, tasks: [] } + if (command.workspace !== this.workspace) throw new Error("Host workspace identity cannot change") + return { + commandType: command.type, + workspace: command.workspace, + tasks: (await this.api.listHeadlessTaskHistory(command.workspace)) + .filter((item) => item.parentTaskId === undefined) + .map((item) => ({ + rootTaskId: item.rootTaskId ?? item.id, + currentTaskId: item.delegatedToId ?? item.id, + workspace: command.workspace, + state: + item.status === "completed" + ? ("completed" as const) + : item.status === "interrupted" + ? ("interrupted" as const) + : item.status === "delegated" + ? ("waiting" as const) + : ("running" as const), + })), + } } } } diff --git a/packages/zoo-host/src/events.ts b/packages/zoo-host/src/events.ts new file mode 100644 index 0000000000..4fe8465e9b --- /dev/null +++ b/packages/zoo-host/src/events.ts @@ -0,0 +1,162 @@ +import type { RooCodeAPI, HeadlessTaskResult } from "@roo-code/types" +import { RooCodeEventName } from "@roo-code/types" +import { ZOO_PUBLIC_SCHEMA_VERSION, type ZooStreamEvent } from "@roo-code/zoo-protocol" + +import { HostTransport } from "./transport.js" + +export class HostEventBridge { + private publicSequence = 0 + private readonly roots = new Map() + private readonly startedAt = new Map() + private readonly pendingCreated = new Set() + + constructor( + private readonly api: RooCodeAPI, + private readonly transport: HostTransport, + private readonly workspace: string, + private readonly clientVersion: string, + private readonly hostVersion: string, + ) {} + + public async initialize(): Promise { + await this.emit({ + type: "system.init", + protocol: "zoo-stream", + hostProtocolVersion: 1, + capabilities: [ + "task:start", + "task:resume", + "task:input", + "task:cancel", + "ask:respond", + "history:list", + "host:snapshot", + "host:shutdown", + "checkpoint:unavailable", + ], + clientVersion: this.clientVersion, + hostVersion: this.hostVersion, + }) + this.api.on(RooCodeEventName.TaskCreated, (taskId) => { + this.pendingCreated.add(taskId) + }) + this.api.on(RooCodeEventName.TaskStarted, (taskId) => { + if (this.pendingCreated.delete(taskId)) { + this.roots.set(taskId, taskId) + void this.emitTask("task.created", taskId, {}) + } + this.startedAt.set(this.roots.get(taskId) ?? taskId, Date.now()) + void this.emitTask("task.started", taskId, {}) + void this.emitTask("task.lifecycle", taskId, { state: "running" }) + }) + this.api.on(RooCodeEventName.TaskDelegated, (parentTaskId, childTaskId) => { + const rootTaskId = this.roots.get(parentTaskId) ?? parentTaskId + this.roots.set(childTaskId, rootTaskId) + if (this.pendingCreated.delete(childTaskId)) { + void this.emitTask("task.created", childTaskId, { parentTaskId }, rootTaskId) + } + void this.emitTask("task.delegated", childTaskId, { parentTaskId, childTaskId }, rootTaskId) + }) + this.api.on(RooCodeEventName.Message, ({ taskId, message }) => { + if (message.type !== "say" || !message.say || message.say === "api_req_started") return + const role = message.say === "reasoning" ? "reasoning" : "assistant" + void this.emitTask("message.upsert", taskId, { + messageId: String(message.ts), + role, + content: message.text ?? "", + complete: message.partial !== true, + }) + }) + this.api.on(RooCodeEventName.HeadlessAsk, (ask) => { + this.roots.set(ask.taskId, ask.rootTaskId) + void (async () => { + await this.emitTask( + "ask.required", + ask.taskId, + { askId: ask.askId, category: ask.ask, subject: ask.text ?? ask.ask }, + ask.rootTaskId, + ) + await this.emitTask("task.lifecycle", ask.taskId, { state: "waiting" }, ask.rootTaskId) + })() + }) + this.api.on(RooCodeEventName.HeadlessTaskResult, (result) => void this.emitResult(result)) + } + + private async emitResult(event: { + rootTaskId: string + currentTaskId: string + outcome: "completed" | "cancelled" | "failed" + resumable: boolean + cancellationReason?: "user" | "signal" | "timeout" + content?: string + }): Promise { + const detailed = (await this.api.getHeadlessTaskResult(event.rootTaskId)) as HeadlessTaskResult | undefined + const outcome = event.outcome + await this.emitTask( + "task.lifecycle", + event.currentTaskId, + { + state: outcome === "completed" ? "completed" : outcome === "failed" ? "failed" : "interrupted", + cause: outcome === "cancelled" ? "cancelled" : outcome === "failed" ? "failed" : undefined, + }, + event.rootTaskId, + ) + await this.emit({ + type: "task.result", + rootTaskId: event.rootTaskId, + taskId: event.rootTaskId, + result: { + schemaVersion: 1, + protocol: "zoo-run-result", + success: outcome === "completed", + outcome, + rootTaskId: event.rootTaskId, + currentTaskId: event.currentTaskId, + workspace: this.workspace, + resumable: event.resumable, + content: event.content ?? detailed?.content, + error: + outcome === "failed" + ? { + code: + detailed?.error?.code === "shutdown" + ? "task_failed" + : (detailed?.error?.code ?? "task_failed"), + message: detailed?.error?.message ?? "Task failed", + kind: "runtime", + } + : undefined, + usage: detailed?.tokenUsage + ? { + inputTokens: detailed.tokenUsage.totalTokensIn, + outputTokens: detailed.tokenUsage.totalTokensOut, + cacheReads: detailed.tokenUsage.totalCacheReads, + cacheWrites: detailed.tokenUsage.totalCacheWrites, + } + : undefined, + cost: detailed?.tokenUsage?.totalCost, + elapsedMs: Date.now() - (this.startedAt.get(event.rootTaskId) ?? Date.now()), + cancellationReason: + outcome === "cancelled" + ? (detailed?.cancellationReason ?? event.cancellationReason ?? "user") + : undefined, + }, + }) + this.startedAt.delete(event.rootTaskId) + } + + private emitTask(type: string, taskId: string, data: Record, rootTaskId?: string): Promise { + return this.emit({ type, rootTaskId: rootTaskId ?? this.roots.get(taskId) ?? taskId, taskId, ...data }) + } + + private async emit(event: Record): Promise { + const normalized = { + v: ZOO_PUBLIC_SCHEMA_VERSION, + seq: ++this.publicSequence, + timestamp: new Date().toISOString(), + hostId: this.transport.hostId, + ...event, + } as ZooStreamEvent + await this.transport.send({ type: "event", event: normalized }) + } +} diff --git a/packages/zoo-host/src/index.ts b/packages/zoo-host/src/index.ts index 55a49d1077..141ffd5280 100644 --- a/packages/zoo-host/src/index.ts +++ b/packages/zoo-host/src/index.ts @@ -1,5 +1,6 @@ export * from "./bootstrap.js" export * from "./dispatcher.js" +export * from "./events.js" export * from "./roots.js" export * from "./security.js" export * from "./transport.js" diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index ab86cce4b6..4d53e1cee9 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -195,6 +195,37 @@ importers: specifier: workspace:^ version: link:../../packages/build + apps/zoo: + dependencies: + '@roo-code/zoo-host': + specifier: workspace:^ + version: link:../../packages/zoo-host + '@roo-code/zoo-protocol': + specifier: workspace:^ + version: link:../../packages/zoo-protocol + commander: + specifier: ^12.1.0 + version: 12.1.0 + devDependencies: + '@roo-code/config-eslint': + specifier: workspace:^ + version: link:../../packages/config-eslint + '@roo-code/config-typescript': + specifier: workspace:^ + version: link:../../packages/config-typescript + '@types/node': + specifier: 22.20.1 + version: 22.20.1 + rimraf: + specifier: 6.0.1 + version: 6.0.1 + tsup: + specifier: 8.5.1 + version: 8.5.1(jiti@2.7.0)(postcss@8.5.15)(tsx@4.22.4)(typescript@5.9.3)(yaml@2.9.0) + vitest: + specifier: 4.1.9 + version: 4.1.9(@opentelemetry/api@1.9.0)(@types/node@22.20.1)(@vitest/coverage-v8@4.1.9)(@vitest/ui@4.1.9)(jsdom@26.1.0)(vite@8.1.0(@types/node@22.20.1)(esbuild@0.28.1)(jiti@2.7.0)(tsx@4.22.4)(yaml@2.9.0)) + packages/build: dependencies: zod: diff --git a/src/extension/api.ts b/src/extension/api.ts index 3c47a218e2..632868de04 100644 --- a/src/extension/api.ts +++ b/src/extension/api.ts @@ -257,6 +257,7 @@ export class API extends EventEmitter implements RooCodeAPI { public async cancelHeadlessTask({ rootTaskId, + reason, }: { rootTaskId: string reason: "user" | "signal" | "timeout" @@ -277,6 +278,7 @@ export class API extends EventEmitter implements RooCodeAPI { currentTaskId: currentTask.taskId, outcome: "cancelled", resumable, + cancellationReason: reason, }) return { rootTaskId, resumable, status: "interrupted" } } catch (error) { @@ -341,6 +343,7 @@ export class API extends EventEmitter implements RooCodeAPI { currentTaskId: result.currentTaskId, outcome: result.outcome, resumable: result.resumable, + cancellationReason: result.cancellationReason, content: result.content, historyItem: this.sidebarProvider.taskHistoryStore.get(result.rootTaskId), }) @@ -458,6 +461,10 @@ export class API extends EventEmitter implements RooCodeAPI { return item ? structuredClone(item) : undefined } + public async listHeadlessTaskHistory(workspace: string) { + return structuredClone(this.sidebarProvider.taskHistoryStore.getByWorkspace(workspace)) + } + public async getTaskApiConversationHistoryLength(taskId: string): Promise { try { const { apiConversationHistory } = await this.sidebarProvider.getTaskWithId(taskId) From e7e0bc57209cea1b1cca12e67c4bc9d01d03c88e Mon Sep 17 00:00:00 2001 From: Toray Altas Date: Wed, 5 Aug 2026 12:18:58 -0400 Subject: [PATCH 2/8] no-mistakes(review): Harden Zoo CLI automation safety boundaries --- apps/zoo/src/__tests__/fixtures/fake-host.mjs | 44 ++++++ apps/zoo/src/automation.ts | 126 +++++++++++------ apps/zoo/src/supervisor.ts | 68 +++++++-- packages/types/src/api.ts | 3 +- packages/types/src/events.ts | 2 +- packages/zoo-host/src/child.ts | 2 +- packages/zoo-host/src/dispatcher.ts | 17 +++ packages/zoo-host/src/events.ts | 129 ++++++++++++++++-- packages/zoo-host/src/security.ts | 2 +- packages/zoo-protocol/src/redaction.ts | 55 +++++++- src/core/auto-approval/index.ts | 5 +- src/core/task/Task.ts | 13 +- src/core/tools/RunSlashCommandTool.ts | 2 +- src/core/tools/SwitchModeTool.ts | 6 +- src/extension/api.ts | 20 +++ 15 files changed, 414 insertions(+), 80 deletions(-) diff --git a/apps/zoo/src/__tests__/fixtures/fake-host.mjs b/apps/zoo/src/__tests__/fixtures/fake-host.mjs index 1042c003d8..eaec1d045f 100644 --- a/apps/zoo/src/__tests__/fixtures/fake-host.mjs +++ b/apps/zoo/src/__tests__/fixtures/fake-host.mjs @@ -54,6 +54,9 @@ process.on("message", (message) => { commandId: message.id, data: { commandType: "task.start", task: { rootTaskId: "root-1", taskId: "root-1" } }, }) + stream({ type: "task.created", requestId: message.id, rootTaskId: "root-1", taskId: "root-1" }) + stream({ type: "task.started", rootTaskId: "root-1", taskId: "root-1" }) + stream({ type: "task.lifecycle", rootTaskId: "root-1", taskId: "root-1", state: "running" }) if (scenario === "crash") { setImmediate(() => process.exit(70)) return @@ -68,10 +71,31 @@ process.on("message", (message) => { category: "followup", subject: "Choose a target", }) + stream({ type: "task.lifecycle", rootTaskId: "root-1", taskId: "root-1", state: "waiting" }) + stream({ + type: "task.result", + requestId: message.id, + rootTaskId: "root-1", + taskId: "root-1", + result: { + schemaVersion: 1, + protocol: "zoo-run-result", + success: false, + outcome: "needs_input", + rootTaskId: "root-1", + currentTaskId: "root-1", + workspace: message.workspace, + resumable: true, + content: "Choose a target", + elapsedMs: 5, + }, + }) return } + stream({ type: "task.lifecycle", rootTaskId: "root-1", taskId: "root-1", state: "completed" }) stream({ type: "task.result", + requestId: message.id, rootTaskId: "root-1", taskId: "root-1", result: { @@ -90,8 +114,16 @@ process.on("message", (message) => { } else if (message.type === "task.cancel") { send({ type: "command.done", commandId: message.id, data: { commandType: "task.cancel", rootTaskId: message.rootTaskId } }) if (scenario === "hang") { + stream({ + type: "task.lifecycle", + rootTaskId: message.rootTaskId, + taskId: message.rootTaskId, + state: "interrupted", + cause: "cancelled", + }) stream({ type: "task.result", + requestId: message.id, rootTaskId: message.rootTaskId, taskId: message.rootTaskId, result: { @@ -124,8 +156,20 @@ process.on("message", (message) => { commandId: message.id, data: { commandType: "task.resume", task: { rootTaskId: message.rootTaskId, taskId: message.taskId } }, }) + stream({ type: "task.created", rootTaskId: message.rootTaskId, taskId: message.taskId }) + stream({ type: "task.lifecycle", rootTaskId: message.rootTaskId, taskId: message.taskId, state: "interrupted" }) + stream({ + type: "task.resumed", + requestId: message.id, + rootTaskId: message.rootTaskId, + taskId: message.taskId, + previousState: "interrupted", + }) + stream({ type: "task.started", rootTaskId: message.rootTaskId, taskId: message.taskId }) + stream({ type: "task.lifecycle", rootTaskId: message.rootTaskId, taskId: message.taskId, state: "completed" }) stream({ type: "task.result", + requestId: message.id, rootTaskId: message.rootTaskId, taskId: message.rootTaskId, result: { diff --git a/apps/zoo/src/automation.ts b/apps/zoo/src/automation.ts index 21f7e6c725..b0ee272151 100644 --- a/apps/zoo/src/automation.ts +++ b/apps/zoo/src/automation.ts @@ -6,6 +6,7 @@ import { fileURLToPath } from "node:url" import { exitCodeFor, ZOO_PUBLIC_SCHEMA_VERSION, + zooErrorCodeSchema, zooRunResultSchema, type HostCommand, type ZooErrorCode, @@ -49,6 +50,7 @@ export async function runAutomation( } }) let lastEvent: ZooStreamEvent | undefined + let finalRendered = false const makeResult = ( outcome: "needs_input" | "cancelled" | "timed_out" | "failed", rootTaskId: string, @@ -93,23 +95,53 @@ export async function runAutomation( : event lastEvent = normalizedEvent projection = reduceSession(projection, normalizedEvent) - renderer.event(normalizedEvent) + if (normalizedEvent.type !== "task.result") renderer.event(normalizedEvent) if (normalizedEvent.type === "task.result") settleResult?.(normalizedEvent.result) - if (normalizedEvent.type === "ask.required" && options.approval !== "auto") { - settleResult?.( - makeResult("needs_input", normalizedEvent.rootTaskId, { - currentTaskId: normalizedEvent.taskId, - resumable: true, - content: normalizedEvent.subject, - }), - ) - } }, }) let signal: "SIGINT" | "SIGTERM" | undefined let notifySignal: ((value: "SIGINT" | "SIGTERM") => void) | undefined const signalPromise = new Promise<"SIGINT" | "SIGTERM">((resolve) => (notifySignal = resolve)) let rootTaskId: string | undefined + let clientStopped = false + const deadline = options.timeout === undefined ? undefined : startedAt + options.timeout + const remainingDeadline = () => (deadline === undefined ? 7_000 : Math.max(0, deadline - Date.now())) + const renderFinal = (result: ZooRunResult) => { + if (finalRendered) return + finalRendered = true + if (options.format === "stream-json") { + const event: ZooStreamEvent = + lastEvent?.type === "task.result" + ? { ...lastEvent, result } + : { + v: 1, + seq: (lastEvent?.seq ?? 0) + 1, + timestamp: new Date().toISOString(), + hostId: lastEvent?.hostId ?? "parent", + type: "task.result", + rootTaskId: result.rootTaskId, + taskId: result.rootTaskId, + result, + } + renderer.event(event) + } + renderer.result(result) + } + const resultExitCode = (result: ZooRunResult) => { + const failedCode = + result.error?.code === "task_timed_out" || result.error?.code === "cleanup_timed_out" + ? "task_failed" + : (result.error?.code ?? "task_failed") + return exitCodeFor( + result.outcome === "failed" + ? { outcome: "failed", errorCode: failedCode } + : result.outcome === "cancelled" + ? { outcome: "cancelled", signal } + : result.outcome === "timed_out" + ? { outcome: "timed_out", errorCode: result.error?.code === "cleanup_timed_out" ? "cleanup_timed_out" : "task_timed_out" } + : { outcome: result.outcome }, + ) + } const onSignal = (received: "SIGINT" | "SIGTERM") => { if (signal) { void client.stop() @@ -167,10 +199,8 @@ export async function runAutomation( rootTaskId = accepted.data.task.rootTaskId if (signal) void client.command({ type: "task.cancel", rootTaskId, reason: "signal" }).catch(() => undefined) const localSettlement = Promise.race([ - timeoutPromise.then(async () => { - await client - .command({ type: "task.cancel", rootTaskId: rootTaskId!, reason: "timeout" }, 5_000) - .catch(() => undefined) + timeoutPromise.then(() => { + void client.command({ type: "task.cancel", rootTaskId: rootTaskId!, reason: "timeout" }, 1).catch(() => undefined) return makeResult("timed_out", rootTaskId!, { currentTaskId: projection.currentTaskId, resumable: true, @@ -209,38 +239,50 @@ export async function runAutomation( ), ]) const result = await Promise.race([resultPromise, localSettlement]) - if (options.format === "stream-json" && lastEvent?.type !== "task.result") { - const event: ZooStreamEvent = { - v: 1, - seq: (lastEvent?.seq ?? 0) + 1, - timestamp: new Date().toISOString(), - hostId: lastEvent?.hostId ?? "parent", - type: "task.result", - rootTaskId: result.rootTaskId, - taskId: result.rootTaskId, - result, - } - renderer.event(event) - } - renderer.result(result) - const failedCode = - result.error?.code === "task_timed_out" || result.error?.code === "cleanup_timed_out" - ? "task_failed" - : (result.error?.code ?? "task_failed") - return exitCodeFor( - result.outcome === "failed" - ? { outcome: "failed", errorCode: failedCode } - : result.outcome === "cancelled" - ? { outcome: "cancelled", signal } - : result.outcome === "timed_out" - ? { outcome: "timed_out", errorCode: "task_timed_out" } - : { outcome: result.outcome }, - ) + const cleanupCompleted = await client.stop(remainingDeadline()) + clientStopped = true + const finalResult = + !cleanupCompleted && result.outcome !== "timed_out" + ? makeResult("timed_out", result.rootTaskId, { + currentTaskId: result.currentTaskId, + resumable: result.resumable, + code: "cleanup_timed_out", + message: "Task cleanup exceeded the deadline", + }) + : result + renderFinal(finalResult) + return resultExitCode(finalResult) + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + const parsedCode = zooErrorCodeSchema.safeParse(message.split(":", 1)[0]) + const code: ZooErrorCode = parsedCode.success + ? parsedCode.data + : message.includes("protocol") || message.includes("negotiat") + ? "protocol_incompatible" + : rootTaskId + ? "task_failed" + : "host_start_failed" + const result = + message.includes("deadline") + ? makeResult("timed_out", rootTaskId ?? "unavailable", { + resumable: Boolean(rootTaskId), + code: "task_timed_out", + message, + }) + : signal + ? makeResult("cancelled", rootTaskId ?? "unavailable", { resumable: false }) + : makeResult("failed", rootTaskId ?? "unavailable", { + resumable: false, + code, + message, + }) + renderFinal(result) + return resultExitCode(result) } finally { if (timeout) clearTimeout(timeout) process.removeListener("SIGINT", onSignal) process.removeListener("SIGTERM", onSignal) - await client.stop() + if (!clientStopped) await client.stop(remainingDeadline()) renderer.dispose() if (options.ephemeral) fs.rmSync(storageRoot, { recursive: true, force: true }) } diff --git a/apps/zoo/src/supervisor.ts b/apps/zoo/src/supervisor.ts index cc65689151..6e24fe23a0 100644 --- a/apps/zoo/src/supervisor.ts +++ b/apps/zoo/src/supervisor.ts @@ -6,11 +6,13 @@ import { fileURLToPath } from "node:url" import { createHostEventStreamParser, + createTextRedactor, hostCommandSchema, hostHelloSchema, negotiateProtocol, parentHelloSchema, validateNegotiatedStreamSession, + validateStreamLifecycle, ZOO_HOST_PROTOCOL_VERSION, type HostCommand, type HostEvent, @@ -18,7 +20,6 @@ import { type ParentHello, type ZooCapability, type ZooStreamEvent, - redactText, } from "@roo-code/zoo-protocol" type PendingCommand = { @@ -60,6 +61,11 @@ export class HostClient { private initialized = new Promise((resolve) => (this.resolveInitialized = resolve)) private lastHeartbeat = Date.now() private watchdog: NodeJS.Timeout | undefined + private readonly commands: HostCommand[] = [] + private readonly hostEvents: HostEvent[] = [] + private readonly publicEvents: ZooStreamEvent[] = [] + private initiatingCommandId: string | undefined + private pendingResult: ZooStreamEvent | undefined constructor(private readonly options: HostClientOptions) {} @@ -82,8 +88,12 @@ export class HostClient { }) this.child = child child.stdout?.resume() + const stderrRedactor = createTextRedactor() child.stderr?.on("data", (chunk: Buffer) => { - if (this.options.debug) process.stderr.write(redactText(chunk.subarray(0, 16 * 1024).toString("utf8"))) + if (this.options.debug) process.stderr.write(stderrRedactor.push(chunk.toString("utf8"))) + }) + child.stderr?.once("end", () => { + if (this.options.debug) process.stderr.write(stderrRedactor.flush()) }) child.once("exit", (code, signal) => this.fail(new Error(`Zoo host exited (${signal ?? code ?? "unknown"})`))) child.once("error", (error) => this.fail(error)) @@ -127,6 +137,8 @@ export class HostClient { if (!this.child?.connected) throw this.failure ?? new Error("Zoo host is unavailable") const id = randomUUID() const command = hostCommandSchema.parse({ v: ZOO_HOST_PROTOCOL_VERSION, id, ...input }) + this.commands.push(command) + if (command.type === "task.start" || command.type === "task.resume") this.initiatingCommandId = command.id return new Promise>((resolve, reject) => { const timer = setTimeout(() => { this.pending.delete(id) @@ -149,23 +161,35 @@ export class HostClient { }) } - public async stop(): Promise { + public async stop(timeoutMs = 7_000): Promise { const child = this.child - if (!child) return + if (!child) return true if (this.watchdog) clearInterval(this.watchdog) this.watchdog = undefined + const deadline = Date.now() + Math.max(0, timeoutMs) + if (timeoutMs <= 0) { + child.kill("SIGKILL") + return false + } if (child.connected && !this.failure) - await this.command({ type: "host.shutdown" }, 5_000).catch(() => undefined) + await this.command({ type: "host.shutdown" }, Math.max(1, Math.min(5_000, deadline - Date.now()))).catch( + () => undefined, + ) child.disconnect() - await new Promise((resolve) => { - if (child.exitCode !== null || child.signalCode !== null) return resolve() + return new Promise((resolve) => { + if (child.exitCode !== null || child.signalCode !== null) return resolve(true) + const remaining = Math.max(0, deadline - Date.now()) + if (remaining === 0) { + child.kill("SIGKILL") + return resolve(false) + } const timer = setTimeout(() => { child.kill("SIGKILL") - resolve() - }, 2_000) + resolve(false) + }, remaining) child.once("exit", () => { clearTimeout(timer) - resolve() + resolve(true) }) }) } @@ -187,8 +211,10 @@ export class HostClient { private receive(input: unknown): void { try { for (const event of this.parser?.push(input) ?? []) { + this.hostEvents.push(event) if (event.type === "host.heartbeat") this.lastHeartbeat = Date.now() if (event.type === "event") { + this.publicEvents.push(event.event) if (event.event.type === "system.init") { if (!this.hello || !this.selection) throw new Error("Host initialized before protocol negotiation") @@ -196,7 +222,13 @@ export class HostClient { if (!validation.ok) throw new Error(validation.message) this.resolveInitialized?.() } - this.options.onEvent(event.event) + if (event.event.type === "task.result") { + if (this.pendingResult) throw new Error("Stream emitted multiple task results") + this.pendingResult = event.event + this.flushResult() + } else { + this.options.onEvent(event.event) + } } if (event.type === "command.ack") { const pending = this.pending.get(event.commandId) @@ -208,6 +240,7 @@ export class HostClient { if (!pending?.acknowledged) throw new Error(`DONE preceded ACK for command ${event.commandId}`) pending.resolve(event) this.pending.delete(event.commandId) + this.flushResult() } if (event.type === "command.error") { const pending = this.pending.get(event.commandId) @@ -221,6 +254,19 @@ export class HostClient { } } + private flushResult(): void { + if (!this.pendingResult || !this.initiatingCommandId) return + if (this.commands.some((command) => this.pending.has(command.id))) return + const validation = validateStreamLifecycle(this.publicEvents, this.commands, this.hostEvents, { + initiatingCommandId: this.initiatingCommandId, + commandIds: this.commands.map((command) => command.id), + }) + if (!validation.ok) throw new Error(`${validation.code}: ${validation.message}`) + const result = this.pendingResult + this.pendingResult = undefined + this.options.onEvent(result) + } + private fail(error: Error): void { if (this.failure) return this.failure = error diff --git a/packages/types/src/api.ts b/packages/types/src/api.ts index cea1d2ce1c..4bc6338b0f 100644 --- a/packages/types/src/api.ts +++ b/packages/types/src/api.ts @@ -42,7 +42,7 @@ export type HeadlessCancelSettlement = { export type HeadlessTaskResult = { rootTaskId: string currentTaskId: string - outcome: "completed" | "cancelled" | "failed" + outcome: "completed" | "needs_input" | "cancelled" | "failed" resumable: boolean cancellationReason?: "user" | "signal" | "timeout" content?: string @@ -66,6 +66,7 @@ export interface RooCodeAPI extends EventEmitter { }): Promise resumeHeadlessTask(taskId: string, overrides?: RunOverrides): Promise respondToHeadlessAsk(input: { taskId: string; askId: string; response: HeadlessAskResponse }): Promise + settleHeadlessNeedsInput(input: { rootTaskId: string; taskId: string; content?: string }): Promise cancelHeadlessTask(input: { rootTaskId: string reason: "user" | "signal" | "timeout" diff --git a/packages/types/src/events.ts b/packages/types/src/events.ts index abcdba29ea..487608bb25 100644 --- a/packages/types/src/events.ts +++ b/packages/types/src/events.ts @@ -147,7 +147,7 @@ export const rooCodeEventsSchema = z.object({ z.object({ rootTaskId: z.string(), currentTaskId: z.string(), - outcome: z.enum(["completed", "cancelled", "failed"]), + outcome: z.enum(["completed", "needs_input", "cancelled", "failed"]), resumable: z.boolean(), cancellationReason: z.enum(["user", "signal", "timeout"]).optional(), content: z.string().optional(), diff --git a/packages/zoo-host/src/child.ts b/packages/zoo-host/src/child.ts index 2dece3d376..e2b0f52226 100644 --- a/packages/zoo-host/src/child.ts +++ b/packages/zoo-host/src/child.ts @@ -79,7 +79,7 @@ export async function runChild(config: ChildConfig): Promise { config.buildVersion, ) await bridge.initialize() - const dispatcher = new HostCommandDispatcher(extension.api, transport, roots.workspaceRoot) + const dispatcher = new HostCommandDispatcher(extension.api, transport, roots.workspaceRoot, bridge) transport.startHeartbeat() process.on("message", (message) => { void dispatcher.dispatch(message).catch(async (error) => { diff --git a/packages/zoo-host/src/dispatcher.ts b/packages/zoo-host/src/dispatcher.ts index e9ada4c3e9..e8023b9d05 100644 --- a/packages/zoo-host/src/dispatcher.ts +++ b/packages/zoo-host/src/dispatcher.ts @@ -2,6 +2,7 @@ import type { RooCodeAPI } from "@roo-code/types" import { hostCommandSchema, type HostCommand } from "@roo-code/zoo-protocol" import { HostTransport } from "./transport.js" +import { HostEventBridge } from "./events.js" export class HostCommandDispatcher { private queue = Promise.resolve() @@ -11,6 +12,7 @@ export class HostCommandDispatcher { private readonly api: RooCodeAPI, private readonly transport: HostTransport, private readonly workspace: string, + private readonly bridge?: HostEventBridge, ) {} public dispatch(input: unknown): Promise { @@ -43,6 +45,7 @@ export class HostCommandDispatcher { switch (command.type) { case "task.start": { if (command.workspace !== this.workspace) throw new Error("Host workspace identity cannot change") + this.bridge?.prepareStart(command.id, command.overrides?.approval ?? "safe") const task = await this.api.startHeadlessTask({ text: command.prompt, overrides: command.overrides }) this.activeRootTaskId = task.rootTaskId return { commandType: command.type, task } @@ -53,6 +56,13 @@ export class HostCommandDispatcher { if (history.workspace !== this.workspace) { throw new Error(`Session ${command.taskId} belongs to workspace ${history.workspace ?? "unknown"}`) } + this.bridge?.prepareResume( + command.id, + command.taskId, + command.rootTaskId, + command.overrides?.approval ?? "safe", + history.status === "delegated" ? "waiting" : "interrupted", + ) const task = await this.api.resumeHeadlessTask(command.taskId, command.overrides) this.activeRootTaskId = task.rootTaskId return { commandType: command.type, task } @@ -61,6 +71,12 @@ export class HostCommandDispatcher { await this.api.sendMessage(command.text, command.images) return { commandType: command.type, taskId: command.taskId } case "ask.respond": + this.bridge?.prepareAskResponse( + command.id, + command.taskId, + command.askId, + command.response === "approve" ? "approve" : command.response === "reject" ? "reject" : "needs_input", + ) await this.api.respondToHeadlessAsk({ taskId: command.taskId, askId: command.askId, @@ -71,6 +87,7 @@ export class HostCommandDispatcher { }) return { commandType: command.type, taskId: command.taskId, askId: command.askId } case "task.cancel": + this.bridge?.prepareCancellation(command.id, command.rootTaskId) await this.api.cancelHeadlessTask({ rootTaskId: command.rootTaskId, reason: command.reason }) return { commandType: command.type, rootTaskId: command.rootTaskId } case "host.snapshot": diff --git a/packages/zoo-host/src/events.ts b/packages/zoo-host/src/events.ts index 4fe8465e9b..9dd40badeb 100644 --- a/packages/zoo-host/src/events.ts +++ b/packages/zoo-host/src/events.ts @@ -9,6 +9,22 @@ export class HostEventBridge { private readonly roots = new Map() private readonly startedAt = new Map() private readonly pendingCreated = new Set() + private readonly initiatingRequests = new Map() + private readonly approvalModes = new Map() + private readonly pendingAsks = new Map() + private readonly pendingResponses = new Map() + private readonly cancellationRequests = new Map() + private pendingInitiation: + | { type: "start"; requestId: string; approval: "interactive" | "safe" | "auto" } + | { + type: "resume" + requestId: string + approval: "interactive" | "safe" | "auto" + taskId: string + rootTaskId: string + previousState: "waiting" | "interrupted" + } + | undefined constructor( private readonly api: RooCodeAPI, @@ -18,6 +34,33 @@ export class HostEventBridge { private readonly hostVersion: string, ) {} + public prepareStart(requestId: string, approval: "interactive" | "safe" | "auto"): void { + this.pendingInitiation = { type: "start", requestId, approval } + } + + public prepareResume( + requestId: string, + taskId: string, + rootTaskId: string, + approval: "interactive" | "safe" | "auto", + previousState: "waiting" | "interrupted", + ): void { + this.pendingInitiation = { type: "resume", requestId, taskId, rootTaskId, approval, previousState } + } + + public prepareAskResponse( + requestId: string, + taskId: string, + askId: string, + decision: "approve" | "reject" | "needs_input", + ): void { + this.pendingResponses.set(taskId, { requestId, askId, decision }) + } + + public prepareCancellation(requestId: string, rootTaskId: string): void { + this.cancellationRequests.set(rootTaskId, requestId) + } + public async initialize(): Promise { await this.emit({ type: "system.init", @@ -42,12 +85,30 @@ export class HostEventBridge { }) this.api.on(RooCodeEventName.TaskStarted, (taskId) => { if (this.pendingCreated.delete(taskId)) { - this.roots.set(taskId, taskId) - void this.emitTask("task.created", taskId, {}) + const initiation = this.pendingInitiation + const rootTaskId = initiation?.type === "resume" ? initiation.rootTaskId : taskId + this.roots.set(taskId, rootTaskId) + if (initiation) { + this.initiatingRequests.set(rootTaskId, initiation.requestId) + this.approvalModes.set(rootTaskId, initiation.approval) + } + void (async () => { + await this.emitTask("task.created", taskId, { requestId: initiation?.requestId }, rootTaskId) + if (initiation?.type === "resume") { + await this.emitTask("task.lifecycle", taskId, { state: initiation.previousState }, rootTaskId) + await this.emitTask( + "task.resumed", + taskId, + { requestId: initiation.requestId, previousState: initiation.previousState }, + rootTaskId, + ) + } + await this.emitTask("task.started", taskId, {}, rootTaskId) + await this.emitTask("task.lifecycle", taskId, { state: "running" }, rootTaskId) + })() + this.pendingInitiation = undefined } this.startedAt.set(this.roots.get(taskId) ?? taskId, Date.now()) - void this.emitTask("task.started", taskId, {}) - void this.emitTask("task.lifecycle", taskId, { state: "running" }) }) this.api.on(RooCodeEventName.TaskDelegated, (parentTaskId, childTaskId) => { const rootTaskId = this.roots.get(parentTaskId) ?? parentTaskId @@ -69,6 +130,7 @@ export class HostEventBridge { }) this.api.on(RooCodeEventName.HeadlessAsk, (ask) => { this.roots.set(ask.taskId, ask.rootTaskId) + this.pendingAsks.set(ask.taskId, { askId: ask.askId, subject: ask.text ?? ask.ask }) void (async () => { await this.emitTask( "ask.required", @@ -77,6 +139,28 @@ export class HostEventBridge { ask.rootTaskId, ) await this.emitTask("task.lifecycle", ask.taskId, { state: "waiting" }, ask.rootTaskId) + if (this.approvalModes.get(ask.rootTaskId) !== "interactive") { + await this.api.settleHeadlessNeedsInput({ + rootTaskId: ask.rootTaskId, + taskId: ask.taskId, + content: ask.text ?? ask.ask, + }) + } + })() + }) + this.api.on(RooCodeEventName.TaskAskResponded, (taskId) => { + const response = this.pendingResponses.get(taskId) + if (!response) return + this.pendingResponses.delete(taskId) + this.pendingAsks.delete(taskId) + void (async () => { + await this.emitTask("ask.resolved", taskId, { + requestId: response.requestId, + askId: response.askId, + decision: response.decision, + source: "user", + }) + await this.emitTask("task.lifecycle", taskId, { state: "running", requestId: response.requestId }) })() }) this.api.on(RooCodeEventName.HeadlessTaskResult, (result) => void this.emitResult(result)) @@ -85,24 +169,40 @@ export class HostEventBridge { private async emitResult(event: { rootTaskId: string currentTaskId: string - outcome: "completed" | "cancelled" | "failed" + outcome: "completed" | "needs_input" | "cancelled" | "failed" resumable: boolean cancellationReason?: "user" | "signal" | "timeout" content?: string }): Promise { const detailed = (await this.api.getHeadlessTaskResult(event.rootTaskId)) as HeadlessTaskResult | undefined const outcome = event.outcome - await this.emitTask( - "task.lifecycle", - event.currentTaskId, - { - state: outcome === "completed" ? "completed" : outcome === "failed" ? "failed" : "interrupted", - cause: outcome === "cancelled" ? "cancelled" : outcome === "failed" ? "failed" : undefined, - }, - event.rootTaskId, - ) + const pendingAsk = this.pendingAsks.get(event.currentTaskId) + if (pendingAsk && outcome !== "needs_input") { + await this.emitTask( + "ask.abandoned", + event.currentTaskId, + { + askId: pendingAsk.askId, + reason: outcome === "failed" ? "failed" : "cancelled", + }, + event.rootTaskId, + ) + this.pendingAsks.delete(event.currentTaskId) + } + if (outcome !== "needs_input") { + await this.emitTask( + "task.lifecycle", + event.currentTaskId, + { + state: outcome === "completed" ? "completed" : outcome === "failed" ? "failed" : "interrupted", + cause: outcome === "cancelled" ? "cancelled" : outcome === "failed" ? "failed" : undefined, + }, + event.rootTaskId, + ) + } await this.emit({ type: "task.result", + requestId: this.cancellationRequests.get(event.rootTaskId) ?? this.initiatingRequests.get(event.rootTaskId), rootTaskId: event.rootTaskId, taskId: event.rootTaskId, result: { @@ -143,6 +243,7 @@ export class HostEventBridge { }, }) this.startedAt.delete(event.rootTaskId) + this.cancellationRequests.delete(event.rootTaskId) } private emitTask(type: string, taskId: string, data: Record, rootTaskId?: string): Promise { diff --git a/packages/zoo-host/src/security.ts b/packages/zoo-host/src/security.ts index 8dd0655ef8..9c0e11e822 100644 --- a/packages/zoo-host/src/security.ts +++ b/packages/zoo-host/src/security.ts @@ -46,7 +46,7 @@ export function createSystemVaultBackend(service = "Zoo Code CLI", platform = pr } }, async store(account, value) { - await execFile("security", ["add-generic-password", "-U", "-s", service, "-a", account, "-w", value]) + await spawnWithInput("security", ["add-generic-password", "-U", "-s", service, "-a", account, "-w"], `${value}\n`) }, async delete(account) { try { diff --git a/packages/zoo-protocol/src/redaction.ts b/packages/zoo-protocol/src/redaction.ts index f225db8c3d..b858b8b9e5 100644 --- a/packages/zoo-protocol/src/redaction.ts +++ b/packages/zoo-protocol/src/redaction.ts @@ -20,7 +20,7 @@ const unsafeTerminalEditing = new RegExp( "i", ) const secretPatterns: ReadonlyArray = [ - /\b(?:Authorization|Proxy-Authorization|Cookie|Set-Cookie)\s*:\s*[^\r\n]+/gi, + /\b(?:Authorization|Proxy-Authorization|Cookie|Set-Cookie)\s*:\s*[^\r\n]+(?:\r?\n[ \t]+[^\r\n]*)*/gi, new RegExp(`--${sensitiveKeyName}(?:\\s*=\\s*|\\s+)${cliSecretValue}`, "gi"), new RegExp(`(? { + if (!logicalLine && !logicalOverflowed) return "" + const output = logicalOverflowed ? `${REDACTED}\n` : redactText(logicalLine) + logicalLine = "" + logicalOverflowed = false + return output + } + const consume = (value: string, flush: boolean): string => { + fragment += value + let output = "" + for (;;) { + const newline = fragment.indexOf("\n") + if (newline < 0) break + const line = fragment.slice(0, newline + 1) + fragment = fragment.slice(newline + 1) + if (fragmentOverflowed) { + output += flushLogicalLine() + output += `${REDACTED}\n` + fragmentOverflowed = false + } else if (/^[ \t]/.test(line) && (logicalLine || logicalOverflowed)) { + logicalLine += line + } else { + output += flushLogicalLine() + logicalLine = line + } + if (logicalLine.length > maxBufferedLength) { + logicalLine = "" + logicalOverflowed = true + } + } + if (fragment.length > maxBufferedLength) { + fragment = "" + fragmentOverflowed = true + } + if (flush) { + output += flushLogicalLine() + if (fragment || fragmentOverflowed) output += fragmentOverflowed ? REDACTED : redactText(fragment) + fragment = "" + fragmentOverflowed = false + } + return output + } + return { + push: (value: string) => consume(value, false), + flush: () => consume("", true), + } +} + export function redactValue(value: Record): Record export function redactValue(value: JsonValue[]): JsonValue[] export function redactValue(value: JsonValue): JsonValue diff --git a/src/core/auto-approval/index.ts b/src/core/auto-approval/index.ts index 3d24c7497f..1563f5f184 100644 --- a/src/core/auto-approval/index.ts +++ b/src/core/auto-approval/index.ts @@ -121,6 +121,9 @@ export async function checkAutoApproval({ } if (state.alwaysAllowExecute === true) { + const decision = getCommandDecision(text, state.allowedCommands || [], state.deniedCommands || []) + if (decision === "auto_deny") return { decision: "deny" } + // Execute commands immediately when DCG allows them. ExecuteCommandTool // marks commands blocked by DCG as protected before reaching this check, // which keeps the explicit user approval prompt for those commands. When @@ -130,8 +133,6 @@ export async function checkAutoApproval({ return { decision: "approve" } } - const decision = getCommandDecision(text, state.allowedCommands || [], state.deniedCommands || []) - if (decision === "auto_approve") { return { decision: "approve" } } else if (decision === "auto_deny") { diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 3ffb70e7cb..bfebcab4c1 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -71,7 +71,7 @@ import { combineCommandSequences } from "../../shared/combineCommandSequences" import { t } from "../../i18n" import { getApiMetrics, hasTokenUsageChanged, hasToolUsageChanged } from "../../shared/getApiMetrics" import { ClineAskResponse } from "../../shared/WebviewMessage" -import { defaultModeSlug, getModeBySlug } from "../../shared/modes" +import { type Mode, defaultModeSlug, getModeBySlug } from "../../shared/modes" import { DiffStrategy, type ToolUse, type ToolParamName, toolParamNames } from "../../shared/tools" import { getModelMaxOutputTokens } from "../../shared/api" @@ -821,6 +821,15 @@ export class Task extends EventEmitter implements TaskLike { return this.taskModeReady } + public async switchMode(newMode: string): Promise { + if (!this.isolateRunConfiguration) { + await this.providerRef.deref()?.handleModeSwitch(newMode as Mode) + return + } + this._taskMode = newMode + this.emit(RooCodeEventName.TaskModeSwitched, this.taskId, newMode) + } + /** * Get the task mode asynchronously, ensuring it's properly initialized. * This is the recommended way to access the task mode as it guarantees @@ -2718,7 +2727,7 @@ export class Task extends EventEmitter implements TaskLike { const state = await this.getEffectiveState() const targetMode = getModeBySlug(slashCommandMode, state?.customModes) if (targetMode) { - await provider.handleModeSwitch(slashCommandMode) + await this.switchMode(slashCommandMode) } } } diff --git a/src/core/tools/RunSlashCommandTool.ts b/src/core/tools/RunSlashCommandTool.ts index c6fd48665d..421e7841fc 100644 --- a/src/core/tools/RunSlashCommandTool.ts +++ b/src/core/tools/RunSlashCommandTool.ts @@ -103,7 +103,7 @@ export class RunSlashCommandTool extends BaseTool<"run_slash_command"> { const provider = task.providerRef.deref() const targetMode = getModeBySlug(command.mode, (await provider?.getState())?.customModes) if (targetMode) { - await provider?.handleModeSwitch(command.mode) + await task.switchMode(command.mode) } } diff --git a/src/core/tools/SwitchModeTool.ts b/src/core/tools/SwitchModeTool.ts index a60ce63bde..ffccb2e5c6 100644 --- a/src/core/tools/SwitchModeTool.ts +++ b/src/core/tools/SwitchModeTool.ts @@ -2,7 +2,7 @@ import delay from "delay" import { Task } from "../task/Task" import { formatResponse } from "../prompts/responses" -import { defaultModeSlug, getModeBySlug } from "../../shared/modes" +import { getModeBySlug } from "../../shared/modes" import { BaseTool, ToolCallbacks } from "./BaseTool" import type { ToolUse } from "../../shared/tools" @@ -39,7 +39,7 @@ export class SwitchModeTool extends BaseTool<"switch_mode"> { } // Check if already in requested mode - const currentMode = (await task.providerRef.deref()?.getState())?.mode ?? defaultModeSlug + const currentMode = await task.getTaskMode() if (currentMode === mode_slug) { task.recordToolError("switch_mode") @@ -56,7 +56,7 @@ export class SwitchModeTool extends BaseTool<"switch_mode"> { } // Switch the mode using shared handler - await task.providerRef.deref()?.handleModeSwitch(mode_slug) + await task.switchMode(mode_slug) pushToolResult( `Successfully switched from ${getModeBySlug(currentMode)?.name ?? currentMode} mode to ${ diff --git a/src/extension/api.ts b/src/extension/api.ts index 632868de04..6a38f75d76 100644 --- a/src/extension/api.ts +++ b/src/extension/api.ts @@ -255,6 +255,26 @@ export class API extends EventEmitter implements RooCodeAPI { if (!accepted) throw new Error(`Ask ${askId} is not pending on task ${taskId}`) } + public async settleHeadlessNeedsInput({ + rootTaskId, + taskId, + content, + }: { + rootTaskId: string + taskId: string + content?: string + }): Promise { + const run = this.headlessRuns.get(rootTaskId) + if (!run || run.result) return + this.settleHeadlessRun(run, { + rootTaskId, + currentTaskId: taskId, + outcome: "needs_input", + resumable: true, + content, + }) + } + public async cancelHeadlessTask({ rootTaskId, reason, From 5c2371f577b9e75dcddb6caa4f644738573d9f3e Mon Sep 17 00:00:00 2001 From: Toray Altas Date: Wed, 5 Aug 2026 13:48:47 -0400 Subject: [PATCH 3/8] fix(cli): complete automation safety boundaries --- apps/zoo/src/__tests__/fixtures/fake-host.mjs | 1 - apps/zoo/src/automation.ts | 86 +++++++++++---- apps/zoo/src/supervisor.ts | 34 ++++-- packages/types/src/api.ts | 21 ++++ packages/zoo-host/src/dispatcher.ts | 49 +++++++-- packages/zoo-host/src/events.ts | 104 +++++++++++++----- src/core/auto-approval/index.ts | 5 +- src/core/task/Task.ts | 20 ++++ src/core/webview/ClineProvider.ts | 74 +++++++++---- .../ClineProvider.run-overrides.spec.ts | 15 +++ src/extension/api.ts | 17 +++ 11 files changed, 331 insertions(+), 95 deletions(-) diff --git a/apps/zoo/src/__tests__/fixtures/fake-host.mjs b/apps/zoo/src/__tests__/fixtures/fake-host.mjs index eaec1d045f..b8958ec630 100644 --- a/apps/zoo/src/__tests__/fixtures/fake-host.mjs +++ b/apps/zoo/src/__tests__/fixtures/fake-host.mjs @@ -56,7 +56,6 @@ process.on("message", (message) => { }) stream({ type: "task.created", requestId: message.id, rootTaskId: "root-1", taskId: "root-1" }) stream({ type: "task.started", rootTaskId: "root-1", taskId: "root-1" }) - stream({ type: "task.lifecycle", rootTaskId: "root-1", taskId: "root-1", state: "running" }) if (scenario === "crash") { setImmediate(() => process.exit(70)) return diff --git a/apps/zoo/src/automation.ts b/apps/zoo/src/automation.ts index b0ee272151..3fd2d05db9 100644 --- a/apps/zoo/src/automation.ts +++ b/apps/zoo/src/automation.ts @@ -17,7 +17,7 @@ import { import { runOverrides, type OutputFormat, type SharedOptions } from "./options.js" import { initialProjection, reduceSession } from "./projection.js" import { createRenderer } from "./render.js" -import { defaultStorageRoot, HostClient } from "./supervisor.js" +import { defaultStorageRoot, HostClient, ZooClientError } from "./supervisor.js" type AutomationOptions = Omit & { format: OutputFormat @@ -54,7 +54,15 @@ export async function runAutomation( const makeResult = ( outcome: "needs_input" | "cancelled" | "timed_out" | "failed", rootTaskId: string, - input: { currentTaskId?: string; resumable: boolean; code?: ZooErrorCode; message?: string; content?: string }, + input: { + currentTaskId?: string + resumable: boolean + code?: ZooErrorCode + message?: string + content?: string + kind?: "configuration" | "provider" | "runtime" + phase?: string + }, ): ZooRunResult => zooRunResultSchema.parse({ schemaVersion: ZOO_PUBLIC_SCHEMA_VERSION, @@ -68,7 +76,12 @@ export async function runAutomation( content: input.content, error: outcome === "failed" || outcome === "timed_out" - ? { code: input.code ?? "task_failed", message: input.message ?? "Task failed", kind: "runtime" } + ? { + code: input.code ?? "task_failed", + message: input.message ?? "Task failed", + kind: input.kind ?? "runtime", + phase: input.phase, + } : undefined, elapsedMs: Date.now() - startedAt, cancellationReason: outcome === "cancelled" ? "signal" : undefined, @@ -106,6 +119,18 @@ export async function runAutomation( let clientStopped = false const deadline = options.timeout === undefined ? undefined : startedAt + options.timeout const remainingDeadline = () => (deadline === undefined ? 7_000 : Math.max(0, deadline - Date.now())) + const commandBudget = () => { + const remaining = remainingDeadline() + if (remaining <= 0) throw new ZooClientError("task_timed_out", "Task deadline exceeded") + return Math.max(1, Math.min(15_000, remaining)) + } + const runCommand = (operation: Promise) => + Promise.race([ + operation, + signalPromise.then(() => { + throw new ZooClientError("cancel_failed", `Received ${signal}`) + }), + ]) const renderFinal = (result: ZooRunResult) => { if (finalRendered) return finalRendered = true @@ -138,7 +163,11 @@ export async function runAutomation( : result.outcome === "cancelled" ? { outcome: "cancelled", signal } : result.outcome === "timed_out" - ? { outcome: "timed_out", errorCode: result.error?.code === "cleanup_timed_out" ? "cleanup_timed_out" : "task_timed_out" } + ? { + outcome: "timed_out", + errorCode: + result.error?.code === "cleanup_timed_out" ? "cleanup_timed_out" : "task_timed_out", + } : { outcome: result.outcome }, ) } @@ -163,7 +192,7 @@ export async function runAutomation( if (options.timeout !== undefined) timeout = setTimeout(() => notifyTimeout?.(), options.timeout) try { const startup = await Promise.race([ - client.start().then(() => "started" as const), + client.start(remainingDeadline()).then(() => "started" as const), timeoutPromise, signalPromise.then(() => "signal" as const), ]) @@ -182,17 +211,22 @@ export async function runAutomation( overrides, } } else { - let taskId = request.taskId - if (!taskId) { - const history = await client.command({ type: "history.list", workspace: options.workspace }) - if (history.data.commandType !== "history.list" || history.data.tasks.length === 0) { - throw new Error("No session exists for this workspace") - } - taskId = history.data.tasks[0]!.rootTaskId + const history = await runCommand( + client.command({ type: "history.list", workspace: options.workspace }, commandBudget()), + ) + if (history.data.commandType !== "history.list" || history.data.tasks.length === 0) { + throw new ZooClientError("invalid_session", "No session exists for this workspace") } - command = { type: "task.resume", taskId, rootTaskId: taskId, overrides } + const selected = request.taskId + ? history.data.tasks.find( + (task) => task.rootTaskId === request.taskId || task.currentTaskId === request.taskId, + ) + : history.data.tasks[0] + if (!selected) throw new ZooClientError("invalid_session", `Unknown session ${request.taskId}`) + const taskId = request.taskId === selected.currentTaskId ? request.taskId : selected.currentTaskId + command = { type: "task.resume", taskId, rootTaskId: selected.rootTaskId, overrides } } - const accepted = await client.command(command) + const accepted = await runCommand(client.command(command, commandBudget())) if (accepted.data.commandType !== "task.start" && accepted.data.commandType !== "task.resume") { throw new Error("Host returned an invalid task acceptance") } @@ -200,10 +234,12 @@ export async function runAutomation( if (signal) void client.command({ type: "task.cancel", rootTaskId, reason: "signal" }).catch(() => undefined) const localSettlement = Promise.race([ timeoutPromise.then(() => { - void client.command({ type: "task.cancel", rootTaskId: rootTaskId!, reason: "timeout" }, 1).catch(() => undefined) + void client + .command({ type: "task.cancel", rootTaskId: rootTaskId!, reason: "timeout" }, 1) + .catch(() => undefined) return makeResult("timed_out", rootTaskId!, { currentTaskId: projection.currentTaskId, - resumable: true, + resumable: false, code: "task_timed_out", message: "Task deadline exceeded", }) @@ -233,8 +269,10 @@ export async function runAutomation( makeResult("failed", rootTaskId!, { currentTaskId: projection.currentTaskId, resumable: false, - code: "host_crashed", + code: error instanceof ZooClientError ? error.code : "host_crashed", message: error.message, + kind: error instanceof ZooClientError ? error.detail?.kind : "runtime", + phase: error instanceof ZooClientError ? error.detail?.phase : undefined, }), ), ]) @@ -254,16 +292,14 @@ export async function runAutomation( return resultExitCode(finalResult) } catch (error) { const message = error instanceof Error ? error.message : String(error) - const parsedCode = zooErrorCodeSchema.safeParse(message.split(":", 1)[0]) + const parsedCode = zooErrorCodeSchema.safeParse(error instanceof ZooClientError ? error.code : undefined) const code: ZooErrorCode = parsedCode.success ? parsedCode.data - : message.includes("protocol") || message.includes("negotiat") - ? "protocol_incompatible" - : rootTaskId - ? "task_failed" - : "host_start_failed" + : rootTaskId + ? "task_failed" + : "host_start_failed" const result = - message.includes("deadline") + code === "task_timed_out" || message.includes("deadline") ? makeResult("timed_out", rootTaskId ?? "unavailable", { resumable: Boolean(rootTaskId), code: "task_timed_out", @@ -275,6 +311,8 @@ export async function runAutomation( resumable: false, code, message, + kind: error instanceof ZooClientError ? error.detail?.kind : "runtime", + phase: error instanceof ZooClientError ? error.detail?.phase : undefined, }) renderFinal(result) return resultExitCode(result) diff --git a/apps/zoo/src/supervisor.ts b/apps/zoo/src/supervisor.ts index 6e24fe23a0..1c78727931 100644 --- a/apps/zoo/src/supervisor.ts +++ b/apps/zoo/src/supervisor.ts @@ -19,6 +19,8 @@ import { type HostHello, type ParentHello, type ZooCapability, + type ZooError, + type ZooErrorCode, type ZooStreamEvent, } from "@roo-code/zoo-protocol" @@ -48,6 +50,17 @@ const requiredCapabilities: ZooCapability[] = [ "host:shutdown", ] +export class ZooClientError extends Error { + constructor( + public readonly code: ZooErrorCode, + message: string, + public readonly detail?: ZooError, + ) { + super(message) + this.name = "ZooClientError" + } +} + export class HostClient { private child: ChildProcess | undefined private parser: ReturnType | undefined @@ -69,7 +82,8 @@ export class HostClient { constructor(private readonly options: HostClientOptions) {} - public async start(): Promise { + public async start(timeoutMs = 45_000): Promise { + const deadline = Date.now() + timeoutMs const hostPath = process.env.ZOO_HOST_PATH ?? fileURLToPath(new URL("../../../packages/zoo-host/dist/child.js", import.meta.url)) @@ -98,10 +112,13 @@ export class HostClient { child.once("exit", (code, signal) => this.fail(new Error(`Zoo host exited (${signal ?? code ?? "unknown"})`))) child.once("error", (error) => this.fail(error)) - const hello = await Promise.race([this.waitForHello(child, 15_000), this.failed]) + const hello = await Promise.race([ + this.waitForHello(child, Math.max(1, Math.min(15_000, deadline - Date.now()))), + this.failed, + ]) this.hello = hello const negotiation = negotiateProtocol(hello, [ZOO_HOST_PROTOCOL_VERSION], requiredCapabilities) - if (!negotiation.ok) throw new Error(negotiation.message) + if (!negotiation.ok) throw new ZooClientError("protocol_incompatible", negotiation.message) this.parser = createHostEventStreamParser({ hostId: hello.hostId }) child.on("message", (message) => this.receive(message)) this.selection = parentHelloSchema.parse({ @@ -119,7 +136,7 @@ export class HostClient { new Promise((_, reject) => { initializationTimer = setTimeout( () => reject(new Error("Zoo host initialization timed out")), - 30_000, + Math.max(1, Math.min(30_000, deadline - Date.now())), ) }), ]) @@ -142,7 +159,7 @@ export class HostClient { return new Promise>((resolve, reject) => { const timer = setTimeout(() => { this.pending.delete(id) - reject(new Error(`Host command timed out: ${command.type}`)) + reject(new ZooClientError("task_timed_out", `Host command timed out: ${command.type}`)) }, timeoutMs) this.pending.set(id, { acknowledged: false, @@ -219,7 +236,7 @@ export class HostClient { if (!this.hello || !this.selection) throw new Error("Host initialized before protocol negotiation") const validation = validateNegotiatedStreamSession(this.hello, this.selection, [event.event]) - if (!validation.ok) throw new Error(validation.message) + if (!validation.ok) throw new ZooClientError(validation.code, validation.message) this.resolveInitialized?.() } if (event.event.type === "task.result") { @@ -245,8 +262,9 @@ export class HostClient { if (event.type === "command.error") { const pending = this.pending.get(event.commandId) if (!pending?.acknowledged) throw new Error(`ERROR preceded ACK for command ${event.commandId}`) - pending.reject(new Error(`${event.error.code}: ${event.error.message}`)) + pending.reject(new ZooClientError(event.error.code, event.error.message, event.error)) this.pending.delete(event.commandId) + this.flushResult() } } } catch (error) { @@ -261,7 +279,7 @@ export class HostClient { initiatingCommandId: this.initiatingCommandId, commandIds: this.commands.map((command) => command.id), }) - if (!validation.ok) throw new Error(`${validation.code}: ${validation.message}`) + if (!validation.ok) throw new ZooClientError(validation.code, validation.message) const result = this.pendingResult this.pendingResult = undefined this.options.onEvent(result) diff --git a/packages/types/src/api.ts b/packages/types/src/api.ts index 4bc6338b0f..1681f478e6 100644 --- a/packages/types/src/api.ts +++ b/packages/types/src/api.ts @@ -26,6 +26,26 @@ export type RunOverrides = { approval?: "interactive" | "safe" | "auto" } +export type HeadlessApiErrorCode = + | "invalid_provider" + | "invalid_profile" + | "invalid_model" + | "invalid_mode" + | "invalid_session" + | "credentials_missing" + | "cancel_failed" + +export class HeadlessApiError extends Error { + constructor( + public readonly code: HeadlessApiErrorCode, + message: string, + public readonly kind: "configuration" | "provider" | "runtime" = "runtime", + ) { + super(message) + this.name = "HeadlessApiError" + } +} + export type HeadlessTaskReference = { taskId: string; rootTaskId: string } export type HeadlessAskResponse = @@ -66,6 +86,7 @@ export interface RooCodeAPI extends EventEmitter { }): Promise resumeHeadlessTask(taskId: string, overrides?: RunOverrides): Promise respondToHeadlessAsk(input: { taskId: string; askId: string; response: HeadlessAskResponse }): Promise + submitHeadlessTaskInput(input: { taskId: string; text?: string; images?: string[] }): Promise settleHeadlessNeedsInput(input: { rootTaskId: string; taskId: string; content?: string }): Promise cancelHeadlessTask(input: { rootTaskId: string diff --git a/packages/zoo-host/src/dispatcher.ts b/packages/zoo-host/src/dispatcher.ts index e8023b9d05..8e06c7ad0e 100644 --- a/packages/zoo-host/src/dispatcher.ts +++ b/packages/zoo-host/src/dispatcher.ts @@ -1,4 +1,4 @@ -import type { RooCodeAPI } from "@roo-code/types" +import { HeadlessApiError, type RooCodeAPI } from "@roo-code/types" import { hostCommandSchema, type HostCommand } from "@roo-code/zoo-protocol" import { HostTransport } from "./transport.js" @@ -28,14 +28,22 @@ export class HostCommandDispatcher { const data = await this.executeCommand(command) await this.transport.send({ type: "command.done", commandId: command.id, data }) } catch (error) { + const detail = + error instanceof HeadlessApiError + ? { code: error.code, kind: error.kind, message: error.message } + : { + code: "task_failed" as const, + kind: "runtime" as const, + message: error instanceof Error ? error.message : String(error), + } await this.transport.send({ type: "command.error", commandId: command.id, error: { - code: "task_failed", - kind: "runtime", + code: detail.code, + kind: detail.kind, phase: command.type, - message: error instanceof Error ? error.message : String(error), + message: detail.message, }, }) } @@ -52,9 +60,14 @@ export class HostCommandDispatcher { } case "task.resume": { const history = await this.api.getTaskHistoryItem(command.taskId) - if (!history) throw new Error(`Unknown session ${command.taskId}`) + if (!history) + throw new HeadlessApiError("invalid_session", `Unknown session ${command.taskId}`, "configuration") if (history.workspace !== this.workspace) { - throw new Error(`Session ${command.taskId} belongs to workspace ${history.workspace ?? "unknown"}`) + throw new HeadlessApiError( + "invalid_session", + `Session ${command.taskId} belongs to workspace ${history.workspace ?? "unknown"}`, + "configuration", + ) } this.bridge?.prepareResume( command.id, @@ -68,14 +81,23 @@ export class HostCommandDispatcher { return { commandType: command.type, task } } case "task.input": - await this.api.sendMessage(command.text, command.images) + await this.api.submitHeadlessTaskInput({ + taskId: command.taskId, + text: command.text, + images: command.images, + }) + this.bridge?.recordTaskInput(command.id, command.taskId, command.text ?? "") return { commandType: command.type, taskId: command.taskId } case "ask.respond": this.bridge?.prepareAskResponse( command.id, command.taskId, command.askId, - command.response === "approve" ? "approve" : command.response === "reject" ? "reject" : "needs_input", + command.response === "approve" + ? "approve" + : command.response === "reject" + ? "reject" + : "needs_input", ) await this.api.respondToHeadlessAsk({ taskId: command.taskId, @@ -86,10 +108,17 @@ export class HostCommandDispatcher { : { response: command.response }, }) return { commandType: command.type, taskId: command.taskId, askId: command.askId } - case "task.cancel": + case "task.cancel": { this.bridge?.prepareCancellation(command.id, command.rootTaskId) - await this.api.cancelHeadlessTask({ rootTaskId: command.rootTaskId, reason: command.reason }) + const settlement = await this.api.cancelHeadlessTask({ + rootTaskId: command.rootTaskId, + reason: command.reason, + }) + if (settlement.status === "failed") { + throw new HeadlessApiError("cancel_failed", `Failed to cancel task ${command.rootTaskId}`) + } return { commandType: command.type, rootTaskId: command.rootTaskId } + } case "host.snapshot": return { commandType: command.type, diff --git a/packages/zoo-host/src/events.ts b/packages/zoo-host/src/events.ts index 9dd40badeb..93168a5985 100644 --- a/packages/zoo-host/src/events.ts +++ b/packages/zoo-host/src/events.ts @@ -12,8 +12,13 @@ export class HostEventBridge { private readonly initiatingRequests = new Map() private readonly approvalModes = new Map() private readonly pendingAsks = new Map() - private readonly pendingResponses = new Map() + private readonly pendingResponses = new Map< + string, + { requestId: string; askId: string; decision: "approve" | "reject" | "needs_input" } + >() private readonly cancellationRequests = new Map() + private readonly startedTasks = new Set() + private eventQueue = Promise.resolve() private pendingInitiation: | { type: "start"; requestId: string; approval: "interactive" | "safe" | "auto" } | { @@ -23,7 +28,7 @@ export class HostEventBridge { taskId: string rootTaskId: string previousState: "waiting" | "interrupted" - } + } | undefined constructor( @@ -84,6 +89,9 @@ export class HostEventBridge { this.pendingCreated.add(taskId) }) this.api.on(RooCodeEventName.TaskStarted, (taskId) => { + let creation: + | { requestId?: string; rootTaskId: string; previousState?: "waiting" | "interrupted" } + | undefined if (this.pendingCreated.delete(taskId)) { const initiation = this.pendingInitiation const rootTaskId = initiation?.type === "resume" ? initiation.rootTaskId : taskId @@ -92,46 +100,69 @@ export class HostEventBridge { this.initiatingRequests.set(rootTaskId, initiation.requestId) this.approvalModes.set(rootTaskId, initiation.approval) } - void (async () => { - await this.emitTask("task.created", taskId, { requestId: initiation?.requestId }, rootTaskId) - if (initiation?.type === "resume") { - await this.emitTask("task.lifecycle", taskId, { state: initiation.previousState }, rootTaskId) + creation = { + requestId: initiation?.requestId, + rootTaskId, + previousState: initiation?.type === "resume" ? initiation.previousState : undefined, + } + this.pendingInitiation = undefined + } + this.startedAt.set(this.roots.get(taskId) ?? taskId, Date.now()) + this.enqueue(async () => { + if (creation) { + await this.emitTask("task.created", taskId, { requestId: creation.requestId }, creation.rootTaskId) + if (creation.previousState) { + await this.emitTask( + "task.lifecycle", + taskId, + { state: creation.previousState }, + creation.rootTaskId, + ) await this.emitTask( "task.resumed", taskId, - { requestId: initiation.requestId, previousState: initiation.previousState }, - rootTaskId, + { requestId: creation.requestId, previousState: creation.previousState }, + creation.rootTaskId, ) } - await this.emitTask("task.started", taskId, {}, rootTaskId) - await this.emitTask("task.lifecycle", taskId, { state: "running" }, rootTaskId) - })() - this.pendingInitiation = undefined - } - this.startedAt.set(this.roots.get(taskId) ?? taskId, Date.now()) + } + if (!this.startedTasks.has(taskId)) { + this.startedTasks.add(taskId) + await this.emitTask("task.started", taskId, {}) + } + }) }) this.api.on(RooCodeEventName.TaskDelegated, (parentTaskId, childTaskId) => { const rootTaskId = this.roots.get(parentTaskId) ?? parentTaskId this.roots.set(childTaskId, rootTaskId) - if (this.pendingCreated.delete(childTaskId)) { - void this.emitTask("task.created", childTaskId, { parentTaskId }, rootTaskId) + const created = this.pendingCreated.delete(childTaskId) + this.enqueue(async () => { + if (created) await this.emitTask("task.created", childTaskId, { parentTaskId }, rootTaskId) + await this.emitTask("task.delegated", childTaskId, { parentTaskId, childTaskId }, rootTaskId) + }) + }) + this.api.on(RooCodeEventName.TaskCompleted, (taskId) => { + const rootTaskId = this.roots.get(taskId) + if (rootTaskId && taskId !== rootTaskId) { + this.enqueue(() => this.emitTask("task.lifecycle", taskId, { state: "completed" }, rootTaskId)) } - void this.emitTask("task.delegated", childTaskId, { parentTaskId, childTaskId }, rootTaskId) }) this.api.on(RooCodeEventName.Message, ({ taskId, message }) => { if (message.type !== "say" || !message.say || message.say === "api_req_started") return const role = message.say === "reasoning" ? "reasoning" : "assistant" - void this.emitTask("message.upsert", taskId, { - messageId: String(message.ts), - role, - content: message.text ?? "", - complete: message.partial !== true, - }) + this.enqueue(() => + this.emitTask("message.upsert", taskId, { + messageId: String(message.ts), + role, + content: message.text ?? "", + complete: message.partial !== true, + }), + ) }) this.api.on(RooCodeEventName.HeadlessAsk, (ask) => { this.roots.set(ask.taskId, ask.rootTaskId) this.pendingAsks.set(ask.taskId, { askId: ask.askId, subject: ask.text ?? ask.ask }) - void (async () => { + this.enqueue(async () => { await this.emitTask( "ask.required", ask.taskId, @@ -146,14 +177,14 @@ export class HostEventBridge { content: ask.text ?? ask.ask, }) } - })() + }) }) this.api.on(RooCodeEventName.TaskAskResponded, (taskId) => { const response = this.pendingResponses.get(taskId) if (!response) return this.pendingResponses.delete(taskId) this.pendingAsks.delete(taskId) - void (async () => { + this.enqueue(async () => { await this.emitTask("ask.resolved", taskId, { requestId: response.requestId, askId: response.askId, @@ -161,9 +192,26 @@ export class HostEventBridge { source: "user", }) await this.emitTask("task.lifecycle", taskId, { state: "running", requestId: response.requestId }) - })() + }) }) - this.api.on(RooCodeEventName.HeadlessTaskResult, (result) => void this.emitResult(result)) + this.api.on(RooCodeEventName.HeadlessTaskResult, (result) => this.enqueue(() => this.emitResult(result))) + } + + public recordTaskInput(requestId: string, taskId: string, text: string): void { + this.enqueue(async () => { + await this.emitTask("message.upsert", taskId, { + requestId, + messageId: `input-${requestId}`, + role: "user", + content: text, + complete: true, + }) + await this.emitTask("task.lifecycle", taskId, { requestId, state: "running" }) + }) + } + + private enqueue(operation: () => Promise): void { + this.eventQueue = this.eventQueue.then(operation).catch(() => undefined) } private async emitResult(event: { diff --git a/src/core/auto-approval/index.ts b/src/core/auto-approval/index.ts index 1563f5f184..fa60db0e1e 100644 --- a/src/core/auto-approval/index.ts +++ b/src/core/auto-approval/index.ts @@ -127,16 +127,13 @@ export async function checkAutoApproval({ // Execute commands immediately when DCG allows them. ExecuteCommandTool // marks commands blocked by DCG as protected before reaching this check, // which keeps the explicit user approval prompt for those commands. When - // enabled, DCG is the authoritative command policy, so Zoo's allow and deny - // lists are intentionally bypassed for commands that DCG allows. + // enabled, DCG authorizes commands only after explicit denials are enforced. if (state.destructiveCommandGuardEnabled === true) { return { decision: "approve" } } if (decision === "auto_approve") { return { decision: "approve" } - } else if (decision === "auto_deny") { - return { decision: "deny" } } else { return { decision: "ask" } } diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index bfebcab4c1..e20bbec62b 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -22,6 +22,7 @@ import { type TaskMetadata, type TaskEvents, type ProviderSettings, + type RunOverrides, type TokenUsage, type ToolUsage, type ToolName, @@ -167,6 +168,7 @@ export interface TaskOptions extends CreateTaskOptions { runApiConfigName?: string runApprovalMode?: "interactive" | "safe" | "auto" isolateRunConfiguration?: boolean + runOverrides?: RunOverrides } export class Task extends EventEmitter implements TaskLike { @@ -213,6 +215,7 @@ export class Task extends EventEmitter implements TaskLike { private _taskMode: string | undefined private readonly runApprovalMode: "interactive" | "safe" | "auto" | undefined private readonly isolateRunConfiguration: boolean + private readonly runOverrides: RunOverrides | undefined /** * Promise that resolves when the task mode has been initialized. @@ -490,6 +493,7 @@ export class Task extends EventEmitter implements TaskLike { runApiConfigName, runApprovalMode, isolateRunConfiguration, + runOverrides, }: TaskOptions) { super() @@ -541,6 +545,7 @@ export class Task extends EventEmitter implements TaskLike { this.apiConfiguration = apiConfiguration this.runApprovalMode = runApprovalMode this.isolateRunConfiguration = isolateRunConfiguration ?? false + this.runOverrides = runOverrides ? structuredClone(runOverrides) : undefined this.api = buildApiHandler(this.apiConfiguration) this.rateLimitClock = rateLimitClock ?? createRateLimitClock() this.autoApprovalHandler = new AutoApprovalHandler() @@ -775,6 +780,8 @@ export class Task extends EventEmitter implements TaskLike { alwaysAllowModeSwitch: false, alwaysAllowSubtasks: false, alwaysAllowExecute: false, + alwaysAllowFollowupQuestions: false, + followupAutoApproveTimeoutMs: 0, } : this.runApprovalMode === "auto" ? { @@ -826,10 +833,23 @@ export class Task extends EventEmitter implements TaskLike { await this.providerRef.deref()?.handleModeSwitch(newMode as Mode) return } + const provider = this.providerRef.deref() + const resolved = await provider?.resolveTaskRunOverrides( + { ...this.runOverrides, mode: newMode }, + this.apiConfiguration, + ) + if (resolved) { + this.updateApiConfiguration(resolved.apiConfiguration) + this._taskApiConfigName = resolved.profile + } this._taskMode = newMode this.emit(RooCodeEventName.TaskModeSwitched, this.taskId, newMode) } + public getDelegatedRunOverrides(mode: string): RunOverrides | undefined { + return this.runOverrides ? { ...this.runOverrides, mode } : undefined + } + /** * Get the task mode asynchronously, ensuring it's properly initialized. * This is the recommended way to access the task mode as it guarantees diff --git a/src/core/webview/ClineProvider.ts b/src/core/webview/ClineProvider.ts index 26c6868cbe..e8e2afabe0 100644 --- a/src/core/webview/ClineProvider.ts +++ b/src/core/webview/ClineProvider.ts @@ -35,6 +35,7 @@ import { type TokenUsage, type ToolUsage, type RunOverrides, + HeadlessApiError, type ExtensionMessage, type ExtensionState, type MarketplaceInstalledMetadata, @@ -1212,6 +1213,7 @@ export class ClineProvider runApiConfigName: resolvedRun.profile, runApprovalMode: options?.runOverrides?.approval, isolateRunConfiguration: !!options?.runOverrides, + runOverrides: options?.runOverrides, }) if (isRehydratingCurrentTask) { @@ -3143,17 +3145,42 @@ export class ClineProvider ): Promise<{ apiConfiguration: ProviderSettings; mode?: string; profile?: string }> { if (!overrides) return { apiConfiguration: baseline } if (overrides.profile && overrides.provider) - throw new Error("profile and provider overrides are mutually exclusive") + throw new HeadlessApiError( + "invalid_profile", + "profile and provider overrides are mutually exclusive", + "configuration", + ) + const mode = overrides.mode ?? history.mode ?? defaultModeSlug + if (!getModeBySlug(mode, await this.customModesManager.getCustomModes())) { + throw new HeadlessApiError("invalid_mode", `Unknown mode override: ${mode}`, "configuration") + } let apiConfiguration = structuredClone(baseline) + const modeProfileId = await this.providerSettingsManager.getModeConfigId(mode) + if (modeProfileId) { + const modeProfile = await this.providerSettingsManager.getProfile({ id: modeProfileId }) + if (modeProfile.apiProvider) apiConfiguration = modeProfile + } let profile = history.profile if (overrides.profile) { - apiConfiguration = await this.providerSettingsManager.getProfile({ name: overrides.profile }) + try { + apiConfiguration = await this.providerSettingsManager.getProfile({ name: overrides.profile }) + } catch { + throw new HeadlessApiError( + "invalid_profile", + `Unknown profile override: ${overrides.profile}`, + "configuration", + ) + } profile = overrides.profile } if (overrides.provider) { if (!Object.values(providerIdentifiers).includes(overrides.provider as ProviderName)) { - throw new Error(`Unknown provider override: ${overrides.provider}`) + throw new HeadlessApiError( + "invalid_provider", + `Unknown provider override: ${overrides.provider}`, + "configuration", + ) } apiConfiguration = { ...apiConfiguration, apiProvider: overrides.provider as ProviderName } profile = undefined @@ -3174,13 +3201,13 @@ export class ClineProvider reasoningEffort: overrides.reasoningEffort === "disabled" ? undefined : overrides.reasoningEffort, } } - const mode = overrides.mode ?? history.mode ?? defaultModeSlug - if (!getModeBySlug(mode, await this.customModesManager.getCustomModes())) { - throw new Error(`Unknown mode override: ${mode}`) - } return { apiConfiguration, mode, profile } } + public resolveTaskRunOverrides(overrides: RunOverrides, baseline: ProviderSettings) { + return this.resolveRunOverrides(overrides, baseline) + } + // When initializing a new task, (not from history but from a tool command // new_task) there is no need to remove the previous task since the new // task is a subtask of the previous one, and when it finishes it is removed @@ -3282,6 +3309,7 @@ export class ClineProvider runApiConfigName: resolvedRun.profile, runApprovalMode: runOverrides?.approval, isolateRunConfiguration: !!runOverrides, + runOverrides, ...options, rateLimitClock: this.rateLimitClock, }) @@ -3685,14 +3713,17 @@ export class ClineProvider // This ensures the child's system prompt and configuration are based on the correct mode. // The mode switch must happen before createTask() because the Task constructor // initializes its mode from provider.getState() during initializeTaskMode(). - try { - await this.handleModeSwitch(mode as any) - } catch (e) { - this.log( - `[delegateParentAndOpenChild] handleModeSwitch failed for mode '${mode}': ${ - (e as Error)?.message ?? String(e) - }`, - ) + const delegatedRunOverrides = parent.getDelegatedRunOverrides(mode) + if (!delegatedRunOverrides) { + try { + await this.handleModeSwitch(mode as any) + } catch (e) { + this.log( + `[delegateParentAndOpenChild] handleModeSwitch failed for mode '${mode}': ${ + (e as Error)?.message ?? String(e) + }`, + ) + } } // 4) Create child as sole active (parent reference preserved for lineage) @@ -3706,11 +3737,14 @@ export class ClineProvider // Without this, the child's fire-and-forget startTask() races with step 5, // and the last writer to globalState overwrites the other's changes— // causing the parent's delegation fields to be lost. - const child = await this.createTask(message, undefined, parent as any, { - initialTodos, - initialStatus: "active", - startTask: false, - }) + const child = await this.createTask( + message, + undefined, + parent as any, + { initialTodos, initialStatus: "active", startTask: false }, + {}, + delegatedRunOverrides, + ) // 5) Persist parent delegation metadata BEFORE the child starts writing. // atomicReadAndUpdate reads from the in-memory cache and writes back within a diff --git a/src/core/webview/__tests__/ClineProvider.run-overrides.spec.ts b/src/core/webview/__tests__/ClineProvider.run-overrides.spec.ts index c7d0773870..6595884b3a 100644 --- a/src/core/webview/__tests__/ClineProvider.run-overrides.spec.ts +++ b/src/core/webview/__tests__/ClineProvider.run-overrides.spec.ts @@ -9,6 +9,7 @@ describe("ClineProvider run overrides", () => { return { providerSettingsManager: { getProfile: vi.fn().mockResolvedValue(profile), + getModeConfigId: vi.fn().mockResolvedValue(undefined), activateProfile: vi.fn(), saveConfig: vi.fn(), }, @@ -103,4 +104,18 @@ describe("ClineProvider run overrides", () => { ), ).rejects.toThrow("Persistent configuration and run overrides cannot be combined") }) + + it("uses a mode-associated profile in memory without activating it", async () => { + const host = createResolverHost({ apiProvider: "openrouter", openRouterModelId: "mode-model" }) + host.providerSettingsManager.getModeConfigId.mockResolvedValue("mode-profile") + const result = await ClineProvider.prototype["resolveRunOverrides"].call( + host as unknown as ClineProvider, + { mode: "code" }, + { apiProvider: "anthropic" }, + ) + + expect(host.providerSettingsManager.getProfile).toHaveBeenCalledWith({ id: "mode-profile" }) + expect(result.apiConfiguration).toMatchObject({ apiProvider: "openrouter", openRouterModelId: "mode-model" }) + expect(host.providerSettingsManager.activateProfile).not.toHaveBeenCalled() + }) }) diff --git a/src/extension/api.ts b/src/extension/api.ts index 6a38f75d76..3e9257a5d8 100644 --- a/src/extension/api.ts +++ b/src/extension/api.ts @@ -21,6 +21,7 @@ import { type HeadlessTaskReference, type HeadlessTaskResult, type RunOverrides, + HeadlessApiError, RooCodeEventName, TaskCommandName, isSecretStateKey, @@ -533,6 +534,22 @@ export class API extends EventEmitter implements RooCodeAPI { await this.sidebarProvider.postMessageToWebview({ type: "invoke", invoke: "sendMessage", text, images }) } + public async submitHeadlessTaskInput({ + taskId, + text, + images, + }: { + taskId: string + text?: string + images?: string[] + }): Promise { + const task = this.sidebarProvider.getTaskById(taskId) + if (!task || this.sidebarProvider.getCurrentTask()?.taskId !== taskId) { + throw new HeadlessApiError("invalid_session", `Task ${taskId} is not the active session`, "configuration") + } + await task.submitUserMessage(text ?? "", images) + } + public deleteQueuedMessage(messageId: string) { const currentTask = this.sidebarProvider.getCurrentTask() From d0eb1120610738320f945aeae34351561efce7ea Mon Sep 17 00:00:00 2001 From: Toray Altas Date: Wed, 5 Aug 2026 14:02:12 -0400 Subject: [PATCH 4/8] fix(cli): remove unused host dependency --- apps/zoo/package.json | 1 - pnpm-lock.yaml | 3 --- 2 files changed, 4 deletions(-) diff --git a/apps/zoo/package.json b/apps/zoo/package.json index bdec28a356..b6fb82ce6b 100644 --- a/apps/zoo/package.json +++ b/apps/zoo/package.json @@ -18,7 +18,6 @@ "clean": "rimraf dist .turbo" }, "dependencies": { - "@roo-code/zoo-host": "workspace:^", "@roo-code/zoo-protocol": "workspace:^", "commander": "^12.1.0" }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 4d53e1cee9..15c5765782 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -197,9 +197,6 @@ importers: apps/zoo: dependencies: - '@roo-code/zoo-host': - specifier: workspace:^ - version: link:../../packages/zoo-host '@roo-code/zoo-protocol': specifier: workspace:^ version: link:../../packages/zoo-protocol From 77db96a6dd5e7a8d3b0525556790831cffbf967b Mon Sep 17 00:00:00 2001 From: Toray Altas Date: Wed, 5 Aug 2026 14:48:25 -0400 Subject: [PATCH 5/8] test(cli): align coverage with run isolation --- src/__tests__/provider-delegation.spec.ts | 17 ++++++++++++----- src/core/auto-approval/__tests__/dcg.spec.ts | 4 ++-- .../tools/__tests__/runSlashCommandTool.spec.ts | 11 +++++------ src/core/tools/__tests__/switchModeTool.spec.ts | 2 ++ 4 files changed, 21 insertions(+), 13 deletions(-) diff --git a/src/__tests__/provider-delegation.spec.ts b/src/__tests__/provider-delegation.spec.ts index dba788acd2..cf2cff139a 100644 --- a/src/__tests__/provider-delegation.spec.ts +++ b/src/__tests__/provider-delegation.spec.ts @@ -81,11 +81,18 @@ describe("ClineProvider.delegateParentAndOpenChild()", () => { expect(removeClineFromStack).toHaveBeenCalledTimes(1) // Child task created with startTask: false and initialStatus: "active" - expect(createTask).toHaveBeenCalledWith("Do something", undefined, parentTask, { - initialTodos: [], - initialStatus: "active", - startTask: false, - }) + expect(createTask).toHaveBeenCalledWith( + "Do something", + undefined, + parentTask, + { + initialTodos: [], + initialStatus: "active", + startTask: false, + }, + {}, + undefined, + ) // Delegation metadata written via atomicReadAndUpdate with correct taskId expect(taskHistoryStore.atomicReadAndUpdate).toHaveBeenCalledTimes(1) diff --git a/src/core/auto-approval/__tests__/dcg.spec.ts b/src/core/auto-approval/__tests__/dcg.spec.ts index fb4f0ac9a7..f83e1ac3ea 100644 --- a/src/core/auto-approval/__tests__/dcg.spec.ts +++ b/src/core/auto-approval/__tests__/dcg.spec.ts @@ -19,9 +19,9 @@ describe("Destructive Command Guard auto-approval precedence", () => { mcpServers: [], } - it("auto-approves commands allowed by DCG without consulting Zoo's deny list", async () => { + it("enforces Zoo's deny list before DCG approval", async () => { expect(await checkAutoApproval({ state: baseState, ask: "command", text: "rm file" })).toEqual({ - decision: "approve", + decision: "deny", }) }) diff --git a/src/core/tools/__tests__/runSlashCommandTool.spec.ts b/src/core/tools/__tests__/runSlashCommandTool.spec.ts index e3d135b45f..4e700657e4 100644 --- a/src/core/tools/__tests__/runSlashCommandTool.spec.ts +++ b/src/core/tools/__tests__/runSlashCommandTool.spec.ts @@ -21,6 +21,7 @@ describe("runSlashCommandTool", () => { mockTask = { consecutiveMistakeCount: 0, recordToolError: vi.fn(), + switchMode: vi.fn(), sayAndCreateMissingParamError: vi.fn().mockResolvedValue("Missing parameter error"), ask: vi.fn().mockResolvedValue({}), cwd: "/test/project", @@ -437,7 +438,7 @@ Deploy application to production`, }) it("should switch mode when mode is specified in command", async () => { - const mockHandleModeSwitch = vi.fn() + const mockSwitchMode = vi.fn() const block: ToolUse<"run_slash_command"> = { type: "tool_use" as const, name: "run_slash_command" as const, @@ -464,14 +465,14 @@ Deploy application to production`, }, customModes: undefined, }), - handleModeSwitch: mockHandleModeSwitch, }) + mockTask.switchMode = mockSwitchMode vi.mocked(getCommand).mockResolvedValue(mockCommand) await runSlashCommandTool.handle(mockTask as Task, block, mockCallbacks) - expect(mockHandleModeSwitch).toHaveBeenCalledWith("debug") + expect(mockSwitchMode).toHaveBeenCalledWith("debug") expect(mockCallbacks.pushToolResult).toHaveBeenCalledWith( `Command: /debug-app Description: Debug the application @@ -485,7 +486,6 @@ Start debugging the application`, }) it("should not switch mode when mode is not specified in command", async () => { - const mockHandleModeSwitch = vi.fn() const block: ToolUse<"run_slash_command"> = { type: "tool_use" as const, name: "run_slash_command" as const, @@ -511,14 +511,13 @@ Start debugging the application`, }, customModes: undefined, }), - handleModeSwitch: mockHandleModeSwitch, }) vi.mocked(getCommand).mockResolvedValue(mockCommand) await runSlashCommandTool.handle(mockTask as Task, block, mockCallbacks) - expect(mockHandleModeSwitch).not.toHaveBeenCalled() + expect(mockTask.switchMode).not.toHaveBeenCalled() }) it("should include mode in askApproval message when mode is specified", async () => { diff --git a/src/core/tools/__tests__/switchModeTool.spec.ts b/src/core/tools/__tests__/switchModeTool.spec.ts index a82429ac7c..b17390fddf 100644 --- a/src/core/tools/__tests__/switchModeTool.spec.ts +++ b/src/core/tools/__tests__/switchModeTool.spec.ts @@ -44,6 +44,8 @@ describe("SwitchModeTool", () => { consecutiveMistakeCount: 0, recordToolError: vi.fn(), didToolFailInCurrentTurn: false, + getTaskMode: vi.fn(async () => (await mockTask.providerRef.deref()?.getState())?.mode ?? "code"), + switchMode: mockHandleModeSwitch, sayAndCreateMissingParamError: vi.fn().mockResolvedValue("Missing parameter error"), ask: vi.fn().mockResolvedValue({}), providerRef: { From c389921e858e8edcf26f8654be3716e50f63fdb6 Mon Sep 17 00:00:00 2001 From: Toray Altas Date: Wed, 5 Aug 2026 15:29:16 -0400 Subject: [PATCH 6/8] test(cli): cover automation run boundaries --- .../ask-clear-approval-buttons.spec.ts | 55 +++++++++++++++++++ src/extension/__tests__/api-headless.spec.ts | 33 ++++++++++- 2 files changed, 87 insertions(+), 1 deletion(-) diff --git a/src/core/task/__tests__/ask-clear-approval-buttons.spec.ts b/src/core/task/__tests__/ask-clear-approval-buttons.spec.ts index 8a79fe4b08..efb863d524 100644 --- a/src/core/task/__tests__/ask-clear-approval-buttons.spec.ts +++ b/src/core/task/__tests__/ask-clear-approval-buttons.spec.ts @@ -8,6 +8,8 @@ import { Task } from "../Task" type ProviderStub = { getState: () => Promise postMessageToWebview: ReturnType + handleModeSwitch?: ReturnType + resolveTaskRunOverrides?: ReturnType } function buildTask(provider: ProviderStub | undefined) { @@ -35,6 +37,59 @@ async function attachQueue(task: Task) { } describe("Task.ask auto-approval stamping", () => { + it("switches persistent and isolated task modes through their respective paths", async () => { + const handleModeSwitch = vi.fn().mockResolvedValue(undefined) + const resolveTaskRunOverrides = vi.fn().mockResolvedValue({ + apiConfiguration: { apiProvider: "openrouter", openRouterModelId: "new-model" }, + profile: "ci", + }) + const provider = { + postMessageToWebview: vi.fn(), + getState: async () => ({ mode: "code" }), + handleModeSwitch, + resolveTaskRunOverrides, + } + const task = buildTask(provider) + Object.defineProperties(task, { + taskId: { value: "task-1" }, + isolateRunConfiguration: { value: false, configurable: true }, + runOverrides: { value: { provider: "openrouter", model: "old-model", approval: "safe" } }, + }) + + await task.switchMode("debug") + expect(handleModeSwitch).toHaveBeenCalledWith("debug") + + Object.defineProperty(task, "isolateRunConfiguration", { value: true }) + task["apiConfiguration"] = { apiProvider: "openrouter", openRouterModelId: "old-model" } + task["updateApiConfiguration"] = vi.fn() + await task.switchMode("architect") + + expect(resolveTaskRunOverrides).toHaveBeenCalledWith( + { provider: "openrouter", model: "old-model", approval: "safe", mode: "architect" }, + { apiProvider: "openrouter", openRouterModelId: "old-model" }, + ) + expect(task["updateApiConfiguration"]).toHaveBeenCalledWith({ + apiProvider: "openrouter", + openRouterModelId: "new-model", + }) + expect(task.taskMode).toBe("architect") + }) + + it("returns an independent delegated override set", () => { + const task = buildTask(undefined) + Object.defineProperty(task, "runOverrides", { + value: { provider: "openrouter", model: "model-1", mode: "code", approval: "safe" }, + }) + + const delegated = task.getDelegatedRunOverrides("debug") + expect(delegated).toEqual({ provider: "openrouter", model: "model-1", mode: "debug", approval: "safe" }) + if (delegated) delegated.model = "changed" + expect(task.getDelegatedRunOverrides("debug")?.model).toBe("model-1") + + const plainTask = buildTask(undefined) + expect(plainTask.getDelegatedRunOverrides("debug")).toBeUndefined() + }) + it.each([ ["interactive", { autoApprovalEnabled: false }], ["safe", { autoApprovalEnabled: true, alwaysAllowReadOnly: true, alwaysAllowExecute: false }], diff --git a/src/extension/__tests__/api-headless.spec.ts b/src/extension/__tests__/api-headless.spec.ts index f4bad60fd2..91a522258d 100644 --- a/src/extension/__tests__/api-headless.spec.ts +++ b/src/extension/__tests__/api-headless.spec.ts @@ -22,6 +22,7 @@ type FakeTask = EventEmitter & { parentTaskId?: string pendingAskId?: number respondToAsk: ReturnType + submitUserMessage: ReturnType } function createTask(taskId: string, options: { rootTaskId?: string; parentTaskId?: string } = {}): FakeTask { @@ -29,6 +30,7 @@ function createTask(taskId: string, options: { rootTaskId?: string; parentTaskId taskId, ...options, respondToAsk: vi.fn().mockReturnValue(true), + submitUserMessage: vi.fn().mockResolvedValue(undefined), }) } @@ -40,6 +42,7 @@ describe("API headless facade", () => { let rootTask: FakeTask let currentTask: FakeTask | undefined let createTaskMock: ReturnType + let getByWorkspace: ReturnType beforeEach(() => { providerEvents = new EventEmitter() @@ -50,6 +53,7 @@ describe("API headless facade", () => { providerEvents.emit(RooCodeEventName.TaskCreated, rootTask) return rootTask }) + getByWorkspace = vi.fn().mockReturnValue([]) const fakeProvider = { context: {}, @@ -62,7 +66,7 @@ describe("API headless facade", () => { getCurrentTask: vi.fn(() => currentTask), cancelTask: vi.fn().mockResolvedValue(undefined), dispose: vi.fn().mockResolvedValue(undefined), - taskHistoryStore: { get: (taskId: string) => history.get(taskId) }, + taskHistoryStore: { get: (taskId: string) => history.get(taskId), getByWorkspace }, } // ClineProvider is concrete and has private state; this precise fake exercises only the API boundary above. provider = fakeProvider as unknown as ClineProvider @@ -208,6 +212,33 @@ describe("API headless facade", () => { ) }) + it("settles needs-input, clones workspace history, and routes active input", async () => { + const historyItems = [{ id: "root-1", task: "task" }] + getByWorkspace.mockReturnValue(historyItems) + await api.startHeadlessTask({ text: "input" }) + + await api.submitHeadlessTaskInput({ taskId: "root-1", text: "answer", images: ["image"] }) + expect(rootTask.submitUserMessage).toHaveBeenCalledWith("answer", ["image"]) + await api.settleHeadlessNeedsInput({ rootTaskId: "root-1", taskId: "root-1", content: "Need input" }) + await expect(api.waitForHeadlessTaskResult("root-1")).resolves.toMatchObject({ + outcome: "needs_input", + resumable: true, + content: "Need input", + }) + + const listed = await api.listHeadlessTaskHistory("/workspace") + expect(getByWorkspace).toHaveBeenCalledWith("/workspace") + expect(listed).toEqual(historyItems) + expect(listed).not.toBe(historyItems) + }) + + it("rejects headless input outside the active session", async () => { + await expect(api.submitHeadlessTaskInput({ taskId: "missing", text: "answer" })).rejects.toMatchObject({ + code: "invalid_session", + kind: "configuration", + }) + }) + it("reports completion that was not persisted as a terminal failure", async () => { const failureListener = vi.fn() api.on(RooCodeEventName.HeadlessTerminalFailure, failureListener) From e0dc711b7538c868c4ba2a4a67ec1a8e8df0b8e9 Mon Sep 17 00:00:00 2001 From: Toray Altas Date: Wed, 5 Aug 2026 15:40:38 -0400 Subject: [PATCH 7/8] test(cli): cover missing override profiles --- .../ClineProvider.run-overrides.spec.ts | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/src/core/webview/__tests__/ClineProvider.run-overrides.spec.ts b/src/core/webview/__tests__/ClineProvider.run-overrides.spec.ts index 6595884b3a..31979a0145 100644 --- a/src/core/webview/__tests__/ClineProvider.run-overrides.spec.ts +++ b/src/core/webview/__tests__/ClineProvider.run-overrides.spec.ts @@ -118,4 +118,20 @@ describe("ClineProvider run overrides", () => { expect(result.apiConfiguration).toMatchObject({ apiProvider: "openrouter", openRouterModelId: "mode-model" }) expect(host.providerSettingsManager.activateProfile).not.toHaveBeenCalled() }) + + it("reports a missing profile through the public task resolver", async () => { + const host = createResolverHost() + host.providerSettingsManager.getProfile.mockRejectedValueOnce(new Error("not found")) + const provider = Object.assign(host, { + resolveRunOverrides: ClineProvider.prototype["resolveRunOverrides"], + }) + + await expect( + ClineProvider.prototype.resolveTaskRunOverrides.call( + provider as unknown as ClineProvider, + { profile: "missing", mode: "code" }, + { apiProvider: "anthropic" }, + ), + ).rejects.toMatchObject({ code: "invalid_profile", kind: "configuration" }) + }) }) From e122a3282e7e64cb0c6d8c293b817fef7253c948 Mon Sep 17 00:00:00 2001 From: Toray Altas Date: Wed, 5 Aug 2026 16:06:44 -0400 Subject: [PATCH 8/8] test(cli): close automation coverage branches --- src/core/auto-approval/__tests__/dcg.spec.ts | 8 ++++++++ src/core/task/__tests__/Task.spec.ts | 15 +++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/src/core/auto-approval/__tests__/dcg.spec.ts b/src/core/auto-approval/__tests__/dcg.spec.ts index f83e1ac3ea..11804835a0 100644 --- a/src/core/auto-approval/__tests__/dcg.spec.ts +++ b/src/core/auto-approval/__tests__/dcg.spec.ts @@ -43,6 +43,14 @@ describe("Destructive Command Guard auto-approval precedence", () => { }) }) + it("uses empty command lists when DCG state omits them", async () => { + const { allowedCommands: _allowedCommands, deniedCommands: _deniedCommands, ...state } = baseState + + expect(await checkAutoApproval({ state, ask: "command", text: "unlisted-command" })).toEqual({ + decision: "approve", + }) + }) + it("does not auto-approve via DCG when execute auto-approval is off", async () => { const state = { ...baseState, alwaysAllowExecute: false } diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index b8573b1d16..8d9c0c93ab 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -400,6 +400,21 @@ describe("Cline", () => { expect(providerOn).not.toHaveBeenCalledWith(RooCodeEventName.ProviderProfileChanged, expect.any(Function)) }) + it("clones isolated run overrides at construction", () => { + const runOverrides = { provider: "openrouter", model: "model-1", mode: "code", approval: "safe" } as const + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "isolated task", + startTask: false, + runMode: "code", + isolateRunConfiguration: true, + runOverrides, + }) + + expect(task.getDelegatedRunOverrides("debug")).toEqual({ ...runOverrides, mode: "debug" }) + }) + describe("empty-response retries", () => { function stream(chunks: ApiStreamChunk[]): AsyncGenerator { return (async function* () {