From 688f2d47daa174728461190738782352ed365732 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Thu, 16 Jul 2026 22:36:12 +0530 Subject: [PATCH 1/5] feat(workflows): add local agent execution policy Co-Authored-By: Claude --- src/workflows/policy.ts | 133 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) create mode 100644 src/workflows/policy.ts diff --git a/src/workflows/policy.ts b/src/workflows/policy.ts new file mode 100644 index 00000000..da39c131 --- /dev/null +++ b/src/workflows/policy.ts @@ -0,0 +1,133 @@ +import type { JsonObject } from "./types.js"; + +export const LOCAL_AGENT_POLICY_VERSION = 1 as const; + +export type WorkflowAgentAccess = "read_only" | "workspace_write"; + +export interface EffectiveLocalAgentPolicy { + readonly version: typeof LOCAL_AGENT_POLICY_VERSION; + readonly mode: "workflow" | "compatibility"; + readonly access: WorkflowAgentAccess | "full_access"; + readonly environment: Readonly>; +} + +export interface WorkflowPolicyResolutionInput { + workflowPolicy?: JsonObject; + nodeConfig?: JsonObject; + environment?: NodeJS.ProcessEnv; +} + +const EXACT_ENVIRONMENT_KEYS = new Set([ + "APPDATA", + "COLORTERM", + "COMSPEC", + "HOME", + "LANG", + "LOCALAPPDATA", + "LOGNAME", + "NODE_EXTRA_CA_CERTS", + "NODE_NO_WARNINGS", + "PATH", + "PATHEXT", + "SHELL", + "SYSTEMROOT", + "TEMP", + "TERM", + "TMP", + "TMPDIR", + "TZ", + "USER", + "USERPROFILE", + "WINDIR", +]); +const PREFIX_ENVIRONMENT_KEYS = ["LC_", "XDG_"]; + +export function resolveWorkflowNodePolicy( + input: WorkflowPolicyResolutionInput, +): EffectiveLocalAgentPolicy { + const workflowAccess = validateWorkflowAccess(readAccess(input.workflowPolicy)); + const nodeAccess = validateWorkflowAccess(readAccess(input.nodeConfig)); + const workflowMaximum = workflowAccess ?? "read_only"; + const requested = nodeAccess ?? workflowMaximum; + const access = workflowMaximum === "read_only" || requested === "read_only" + ? "read_only" + : "workspace_write"; + + return deepFreeze({ + version: LOCAL_AGENT_POLICY_VERSION, + mode: "workflow" as const, + access, + environment: filterWorkflowEnvironment(input.environment ?? process.env), + }); +} + +export const resolveWorkflowExecutionPolicy = resolveWorkflowNodePolicy; + +export function createLegacyLocalAgentPolicy( + access: "read_only" | "workspace_write" | "full_access", + environment: NodeJS.ProcessEnv = process.env, +): EffectiveLocalAgentPolicy { + return deepFreeze({ + version: LOCAL_AGENT_POLICY_VERSION, + mode: "compatibility" as const, + access, + environment: copyStringEnvironment(environment), + }); +} + +export function filterWorkflowEnvironment( + environment: NodeJS.ProcessEnv, +): Readonly> { + const filtered: Record = {}; + for (const [key, value] of Object.entries(environment)) { + if (typeof value !== "string") continue; + if (!isAllowedEnvironmentKey(key)) continue; + filtered[key] = value; + } + return Object.freeze(filtered); +} + +function validateWorkflowAccess(access: string | undefined): WorkflowAgentAccess | undefined { + if (access === undefined) return undefined; + if (access === "full_access") { + throw new Error("Workflow agents do not support full_access."); + } + if (access !== "read_only" && access !== "workspace_write") { + throw new Error(`Unsupported workflow agent access '${access}'.`); + } + return access; +} + +function readAccess(value: JsonObject | undefined): string | undefined { + if (!value) return undefined; + const direct = value.access; + if (typeof direct === "string") return direct; + const agent = value.agent; + if (agent && typeof agent === "object" && !Array.isArray(agent)) { + const nested = agent.access; + if (typeof nested === "string") return nested; + } + return undefined; +} + +function isAllowedEnvironmentKey(key: string): boolean { + if (EXACT_ENVIRONMENT_KEYS.has(key.toUpperCase())) return true; + return PREFIX_ENVIRONMENT_KEYS.some((prefix) => key.toUpperCase().startsWith(prefix)); +} + +function copyStringEnvironment(environment: NodeJS.ProcessEnv): Readonly> { + const copied: Record = {}; + for (const [key, value] of Object.entries(environment)) { + if (typeof value === "string") copied[key] = value; + } + return Object.freeze(copied); +} + +function deepFreeze(value: T): T { + for (const nested of Object.values(value)) { + if (nested && typeof nested === "object" && !Object.isFrozen(nested)) { + deepFreeze(nested); + } + } + return Object.freeze(value); +} From 59870dcbb8515e3fbbd1165b8c630e7e1f9f57d2 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Thu, 16 Jul 2026 22:36:18 +0530 Subject: [PATCH 2/5] feat(agents): add streaming run handle core Co-Authored-By: Claude --- src/local-agent-runtime.ts | 478 +++++++++++++++++++++++++++++++++++-- 1 file changed, 458 insertions(+), 20 deletions(-) diff --git a/src/local-agent-runtime.ts b/src/local-agent-runtime.ts index 54130c2e..91681bb5 100644 --- a/src/local-agent-runtime.ts +++ b/src/local-agent-runtime.ts @@ -2,10 +2,12 @@ import type { Codex, CodexOptions, ModelReasoningEffort, - RunResult, SandboxMode, + ThreadEvent, ThreadOptions, + TurnOptions, } from "@openai/codex-sdk"; +import type { EffectiveLocalAgentPolicy } from "./workflows/policy.js"; export type LocalAgentWriteMode = "read_only" | "allowed" | "full_access"; @@ -16,6 +18,8 @@ export interface LocalAgentRunInput { writeMode?: LocalAgentWriteMode; model?: string; thinking?: string; + signal?: AbortSignal; + policy?: EffectiveLocalAgentPolicy; } export interface LocalAgentRunResult { @@ -25,25 +29,393 @@ export interface LocalAgentRunResult { items: unknown[]; } +export const LOCAL_AGENT_EVENT_VERSION = 1 as const; + +interface LocalAgentEventBase { + version: typeof LOCAL_AGENT_EVENT_VERSION; + sequence: number; + timestamp: string; + provider: string; +} + +export type LocalAgentEvent = + | (LocalAgentEventBase & { + type: "lifecycle"; + phase: "started" | "cancelling" | "disposing"; + }) + | (LocalAgentEventBase & { + type: "session"; + providerSessionId: string; + resumed: boolean; + }) + | (LocalAgentEventBase & { + type: "output"; + stream: "assistant" | "thinking"; + delta: string; + }) + | (LocalAgentEventBase & { + type: "tool"; + phase: "started" | "updated" | "completed" | "failed"; + name?: string; + id?: string; + metadata?: Record; + }) + | (LocalAgentEventBase & { + type: "permission"; + phase: "requested" | "allowed" | "denied"; + tool?: string; + metadata?: Record; + }) + | (LocalAgentEventBase & { + type: "warning"; + message: string; + metadata?: Record; + }) + | (LocalAgentEventBase & { + type: "terminal"; + outcome: "succeeded" | "failed" | "cancelled"; + message?: string; + }); + +export type LocalAgentEventInput = + | { type: "lifecycle"; phase: "started" | "cancelling" | "disposing" } + | { type: "session"; providerSessionId: string; resumed: boolean } + | { type: "output"; stream: "assistant" | "thinking"; delta: string } + | { + type: "tool"; + phase: "started" | "updated" | "completed" | "failed"; + name?: string; + id?: string; + metadata?: Record; + } + | { + type: "permission"; + phase: "requested" | "allowed" | "denied"; + tool?: string; + metadata?: Record; + } + | { type: "warning"; message: string; metadata?: Record }; + +export interface LocalAgentRunHandle { + readonly provider: string; + events(): AsyncIterable; + result(): Promise; + cancel(reason?: unknown): Promise; + dispose(): Promise; +} + export interface LocalAgentRuntime { readonly provider: string; + start(input: LocalAgentRunInput): Promise; run(input: LocalAgentRunInput): Promise; } -interface CodexThreadLike { +interface CodexStreamedTurnLike { + events: AsyncGenerator; +} + +export interface CodexThreadLike { readonly id: string | null; - run(prompt: string): Promise; + runStreamed(prompt: string, options?: TurnOptions): Promise; } -interface CodexClientLike { +export interface CodexClientLike { startThread(options?: ThreadOptions): CodexThreadLike; resumeThread(id: string, options?: ThreadOptions): CodexThreadLike; } -type CodexFactory = (options?: CodexOptions) => CodexClientLike; +export type CodexFactory = (options?: CodexOptions) => CodexClientLike; + +type EventWaiter = { + resolve: (value: IteratorResult) => void; + reject: (error: unknown) => void; +}; + +const MAX_BUFFERED_EVENTS = 128; +const MAX_EVENT_TEXT_CHARS = 16_384; +const MAX_EVENT_METADATA_CHARS = 4_096; + +class BoundedEventChannel implements AsyncIterable { + private readonly queue: LocalAgentEvent[] = []; + private readonly waiters: EventWaiter[] = []; + private closed = false; + private consumerCreated = false; + + push(event: LocalAgentEvent): void { + if (this.closed) return; + const waiter = this.waiters.shift(); + if (waiter) { + waiter.resolve({ value: event, done: false }); + return; + } + if (this.queue.length >= MAX_BUFFERED_EVENTS) { + const droppable = this.queue.findIndex((queued) => + queued.type === "output" || + queued.type === "tool" || + queued.type === "permission" || + queued.type === "warning" + ); + this.queue.splice(droppable === -1 ? 0 : droppable, 1); + } + this.queue.push(event); + } + + close(): void { + if (this.closed) return; + this.closed = true; + while (this.waiters.length > 0) { + this.waiters.shift()?.resolve({ value: undefined, done: true }); + } + } + + [Symbol.asyncIterator](): AsyncIterator { + if (this.consumerCreated) { + throw new Error("Local agent events support one consumer."); + } + this.consumerCreated = true; + let iteratorClosed = false; + return { + next: () => { + if (iteratorClosed) return Promise.resolve({ value: undefined, done: true }); + const event = this.queue.shift(); + if (event) return Promise.resolve({ value: event, done: false }); + if (this.closed) return Promise.resolve({ value: undefined, done: true }); + return new Promise>((resolve, reject) => { + this.waiters.push({ resolve, reject }); + }); + }, + return: async () => { + iteratorClosed = true; + while (this.waiters.length > 0) { + this.waiters.shift()?.resolve({ value: undefined, done: true }); + } + return { value: undefined, done: true }; + }, + }; + } +} + +export class LocalAgentRunController implements LocalAgentRunHandle { + readonly provider: string; + private readonly channel = new BoundedEventChannel(); + private readonly resultPromise: Promise; + private resolveResult!: (result: LocalAgentRunResult) => void; + private rejectResult!: (error: unknown) => void; + private cancelHook: (reason?: unknown) => Promise = async () => undefined; + private disposeHook: () => Promise = async () => undefined; + private pump: Promise = Promise.resolve(); + private sequence = 0; + private terminal = false; + private cancelling = false; + private cancellationPromise: Promise | undefined; + private lifecycleReadyResolve!: () => void; + private readonly lifecycleReady: Promise; + private disposePromise: Promise | undefined; + private abortSignal: AbortSignal | undefined; + private abortListener: (() => void) | undefined; + + constructor(provider: string, signal?: AbortSignal) { + this.provider = provider; + this.resultPromise = new Promise((resolve, reject) => { + this.resolveResult = resolve; + this.rejectResult = reject; + }); + this.lifecycleReady = new Promise((resolve) => { + this.lifecycleReadyResolve = resolve; + }); + void this.resultPromise.catch(() => undefined); + this.emit({ type: "lifecycle", phase: "started" }); + if (signal) { + this.abortSignal = signal; + this.abortListener = () => void this.cancel(signal.reason); + if (signal.aborted) void this.cancel(signal.reason); + else signal.addEventListener("abort", this.abortListener, { once: true }); + } + } + + events(): AsyncIterable { + return this.channel; + } + + result(): Promise { + return this.resultPromise; + } + + setLifecycle(options: { + cancel?: (reason?: unknown) => Promise | void; + dispose?: () => Promise | void; + pump?: Promise; + }): void { + if (options.cancel) this.cancelHook = async (reason) => options.cancel?.(reason); + if (options.dispose) this.disposeHook = async () => options.dispose?.(); + if (options.pump) { + this.pump = options.pump; + void this.pump.catch(() => undefined); + } + this.lifecycleReadyResolve(); + } + + emit(event: LocalAgentEventInput): void { + if (this.terminal) return; + const normalized = normalizeEventInput(event); + this.channel.push({ + ...normalized, + version: LOCAL_AGENT_EVENT_VERSION, + sequence: ++this.sequence, + timestamp: new Date().toISOString(), + provider: this.provider, + } as LocalAgentEvent); + } + + succeed(result: LocalAgentRunResult): void { + if (!this.finishTerminal("succeeded")) return; + this.resolveResult(result); + } -function sandboxModeFor(writeMode: LocalAgentWriteMode | undefined): SandboxMode { - switch (writeMode) { + fail(error: unknown): void { + if (this.cancelling) { + this.finishCancelled(error); + return; + } + const message = error instanceof Error ? error.message : String(error); + if (!this.finishTerminal("failed", message)) return; + this.rejectResult(error instanceof Error ? error : new Error(message)); + } + + cancel(reason?: unknown): Promise { + if (this.cancellationPromise) return this.cancellationPromise; + if (this.terminal) return Promise.resolve(); + this.cancelling = true; + this.emit({ type: "lifecycle", phase: "cancelling" }); + this.finishCancelled(reason); + this.cancellationPromise = (async () => { + await this.lifecycleReady; + try { + await this.cancelHook(reason); + } catch { + // Cancellation already has a deterministic terminal result. + } + })(); + return this.cancellationPromise; + } + + dispose(): Promise { + if (this.disposePromise) return this.disposePromise; + this.disposePromise = (async () => { + if (!this.terminal) await this.cancel(new Error("Local agent handle disposed.")); + else await this.cancellationPromise; + try { + await this.disposeHook(); + } finally { + await this.pump.catch(() => undefined); + } + })(); + return this.disposePromise; + } + + private finishCancelled(reason?: unknown): void { + const message = reason instanceof Error ? reason.message : directCancellationMessage(reason); + if (!this.finishTerminal("cancelled", message)) return; + const error = new Error(message ?? "Local agent run cancelled."); + error.name = "AbortError"; + this.rejectResult(error); + } + + private finishTerminal( + outcome: "succeeded" | "failed" | "cancelled", + message?: string, + ): boolean { + if (this.terminal) return false; + this.terminal = true; + if (this.abortSignal && this.abortListener) { + this.abortSignal.removeEventListener("abort", this.abortListener); + this.abortSignal = undefined; + this.abortListener = undefined; + } + this.channel.push({ + version: LOCAL_AGENT_EVENT_VERSION, + sequence: ++this.sequence, + timestamp: new Date().toISOString(), + provider: this.provider, + type: "terminal", + outcome, + ...(message ? { message: message.slice(0, MAX_EVENT_TEXT_CHARS) } : {}), + }); + this.channel.close(); + return true; + } +} + +function normalizeEventInput(event: LocalAgentEventInput): LocalAgentEventInput { + if (event.type === "output") { + return { ...event, delta: event.delta.slice(0, MAX_EVENT_TEXT_CHARS) }; + } + if (event.type === "warning") { + return { + ...event, + message: event.message.slice(0, MAX_EVENT_TEXT_CHARS), + ...(event.metadata ? { metadata: redactMetadata(event.metadata) } : {}), + }; + } + if ((event.type === "tool" || event.type === "permission") && event.metadata) { + return { ...event, metadata: redactMetadata(event.metadata) }; + } + return event; +} + +function redactMetadata(metadata: Record): Record { + const redacted: Record = {}; + let size = 0; + for (const [key, value] of Object.entries(metadata)) { + if (size >= MAX_EVENT_METADATA_CHARS) break; + const next = /token|secret|password|authorization|cookie|api[_-]?key/i.test(key) + ? "[redacted]" + : typeof value === "string" + ? value.slice(0, 512) + : typeof value === "number" || typeof value === "boolean" || value === null + ? value + : "[omitted]"; + redacted[key] = next; + size += key.length + String(next).length; + } + return redacted; +} + +function directCancellationMessage(reason: unknown): string | undefined { + if (typeof reason === "string" && reason.trim()) return reason.trim(); + return undefined; +} + +export async function runLocalAgentHandle(handle: LocalAgentRunHandle): Promise { + const drain = (async () => { + for await (const _event of handle.events()) { + // Compatibility callers intentionally ignore normalized progress events. + } + })(); + try { + return await handle.result(); + } finally { + await handle.dispose(); + await drain; + } +} + +function sandboxModeFor(input: LocalAgentRunInput): SandboxMode { + if (input.policy) { + if (input.policy.mode === "workflow") { + if (input.policy.version !== 1 || + (input.policy.access !== "read_only" && input.policy.access !== "workspace_write")) { + throw new Error("Invalid workflow local-agent policy."); + } + return input.policy.access === "workspace_write" ? "workspace-write" : "read-only"; + } + if (input.policy.mode !== "compatibility") { + throw new Error("Invalid local-agent policy mode."); + } + if (input.policy.access === "full_access") return "danger-full-access"; + return input.policy.access === "workspace_write" ? "workspace-write" : "read-only"; + } + switch (input.writeMode) { case "allowed": return "workspace-write"; case "full_access": @@ -57,7 +429,7 @@ function sandboxModeFor(writeMode: LocalAgentWriteMode | undefined): SandboxMode function threadOptionsFor(input: LocalAgentRunInput): ThreadOptions { return { workingDirectory: input.workspace, - sandboxMode: sandboxModeFor(input.writeMode), + sandboxMode: sandboxModeFor(input), approvalPolicy: "never", model: input.model, modelReasoningEffort: input.thinking as ModelReasoningEffort | undefined, @@ -66,25 +438,91 @@ function threadOptionsFor(input: LocalAgentRunInput): ThreadOptions { export class CodexSdkLocalAgentRuntime implements LocalAgentRuntime { readonly provider = "codex" as const; - private readonly codex: CodexClientLike; - constructor(codex: CodexClientLike) { - this.codex = codex; - } + constructor(private readonly codex: CodexClientLike) {} - async run(input: LocalAgentRunInput): Promise { + async start(input: LocalAgentRunInput): Promise { const options = threadOptionsFor(input); + const resumed = Boolean(input.providerSessionId); const thread = input.providerSessionId ? this.codex.resumeThread(input.providerSessionId, options) : this.codex.startThread(options); - const turn = await thread.run(input.prompt); + const abortController = new AbortController(); + const controller = new LocalAgentRunController(this.provider, input.signal); + let providerEvents: AsyncGenerator | undefined; - return { - provider: this.provider, - providerSessionId: thread.id, - finalResponse: turn.finalResponse, - items: turn.items, - }; + const pump = (async () => { + try { + const streamed = await thread.runStreamed(input.prompt, { signal: abortController.signal }); + providerEvents = streamed.events; + const items: unknown[] = []; + let finalResponse = ""; + let providerSessionId = thread.id; + let emittedSessionId: string | null = null; + if (providerSessionId) { + emittedSessionId = providerSessionId; + controller.emit({ type: "session", providerSessionId, resumed }); + } + for await (const event of providerEvents) { + if (event.type === "thread.started") { + providerSessionId = event.thread_id; + if (providerSessionId !== emittedSessionId) { + emittedSessionId = providerSessionId; + controller.emit({ type: "session", providerSessionId, resumed }); + } + } else if (event.type === "item.completed") { + items.push(event.item); + if (event.item.type === "agent_message") { + finalResponse = event.item.text; + controller.emit({ type: "output", stream: "assistant", delta: event.item.text }); + } else if (event.item.type === "command_execution" || event.item.type === "mcp_tool_call") { + controller.emit({ + type: "tool", + phase: event.item.status === "failed" ? "failed" : "completed", + name: event.item.type, + id: event.item.id, + }); + } + } else if (event.type === "item.started" || event.type === "item.updated") { + if (event.item.type === "command_execution" || event.item.type === "mcp_tool_call") { + controller.emit({ + type: "tool", + phase: event.type === "item.started" ? "started" : "updated", + name: event.item.type, + id: event.item.id, + }); + } + } else if (event.type === "turn.failed") { + throw new Error(`Codex turn failed: ${event.error.message}`); + } else if (event.type === "error") { + throw new Error(`Codex stream failed: ${event.message}`); + } + } + if (!finalResponse.trim()) throw new Error("Codex did not return a final assistant response."); + controller.succeed({ + provider: this.provider, + providerSessionId, + finalResponse: finalResponse.trim(), + items, + }); + } catch (error) { + controller.fail(error); + } + })(); + + controller.setLifecycle({ + cancel: (reason) => abortController.abort(reason), + dispose: async () => { + if (!abortController.signal.aborted) abortController.abort(); + await providerEvents?.return(undefined); + }, + pump, + }); + return controller; + } + + async run(input: LocalAgentRunInput): Promise { + return runLocalAgentHandle(await this.start(input)); } } From 51a0bd34901723b4f2f18b5d52f5f8d8ae3d9d53 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Thu, 16 Jul 2026 22:36:30 +0530 Subject: [PATCH 3/5] feat(process): add graceful tree termination Co-Authored-By: Claude --- src/process-platform.ts | 82 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 77 insertions(+), 5 deletions(-) diff --git a/src/process-platform.ts b/src/process-platform.ts index 905d4d73..594ce1fb 100644 --- a/src/process-platform.ts +++ b/src/process-platform.ts @@ -11,17 +11,33 @@ export interface KillableProcess { kill(signal?: NodeJS.Signals): boolean; } -interface ProcessTreeRuntime { +export interface WaitableProcess extends KillableProcess { + exitCode?: number | null; + signalCode?: NodeJS.Signals | null; + once(event: "close", listener: () => void): unknown; + removeListener(event: "close", listener: () => void): unknown; +} + +export interface ProcessTreeRuntime { platform: NodeJS.Platform; killGroup(pid: number, signal: NodeJS.Signals): void; - killWindowsTree(pid: number): boolean; + isGroupAlive(pid: number): boolean; + killWindowsTree(pid: number, force: boolean): boolean; } const defaultProcessTreeRuntime: ProcessTreeRuntime = { platform: process.platform, killGroup: (pid, signal) => process.kill(-pid, signal), - killWindowsTree: (pid) => { - const result = spawnSync("taskkill.exe", ["/pid", String(pid), "/T", "/F"], { + isGroupAlive: (pid) => { + try { + process.kill(-pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code !== "ESRCH"; + } + }, + killWindowsTree: (pid, force) => { + const result = spawnSync("taskkill.exe", ["/pid", String(pid), "/T", ...(force ? ["/F"] : [])], { stdio: "ignore", windowsHide: true, }); @@ -63,7 +79,7 @@ export function terminateProcessTree( runtime: ProcessTreeRuntime = defaultProcessTreeRuntime, ): void { if (runtime.platform === "win32" && child.pid) { - if (runtime.killWindowsTree(child.pid)) return; + if (runtime.killWindowsTree(child.pid, signal === "SIGKILL")) return; } else if (detached && child.pid) { try { runtime.killGroup(child.pid, signal); @@ -75,3 +91,59 @@ export function terminateProcessTree( child.kill(signal); } + +export async function terminateProcessTreeGracefully( + child: WaitableProcess, + detached: boolean, + graceMs = 1_000, + runtime: ProcessTreeRuntime = defaultProcessTreeRuntime, +): Promise { + const tracksGroup = runtime.platform !== "win32" && detached && child.pid !== undefined; + if (!tracksGroup && child.exitCode !== null && child.exitCode !== undefined) return; + if (!tracksGroup && child.signalCode) return; + + terminateProcessTree(child, "SIGTERM", detached, runtime); + if (await waitForProcessTreeExit(child, detached, graceMs, runtime)) return; + terminateProcessTree(child, "SIGKILL", detached, runtime); + await waitForProcessTreeExit(child, detached, graceMs, runtime); +} + +function waitForProcessTreeExit( + child: WaitableProcess, + detached: boolean, + timeoutMs: number, + runtime: ProcessTreeRuntime, +): Promise { + if (runtime.platform !== "win32" && detached && child.pid !== undefined) { + if (!runtime.isGroupAlive(child.pid)) return Promise.resolve(true); + return new Promise((resolve) => { + const deadline = Date.now() + timeoutMs; + const poll = () => { + if (!runtime.isGroupAlive(child.pid as number)) { + resolve(true); + } else if (Date.now() >= deadline) { + resolve(false); + } else { + setTimeout(poll, Math.min(25, timeoutMs)); + } + }; + setTimeout(poll, Math.min(25, timeoutMs)); + }); + } + return waitForProcessClose(child, timeoutMs); +} + +function waitForProcessClose(child: WaitableProcess, timeoutMs: number): Promise { + if (child.exitCode !== null && child.exitCode !== undefined) return Promise.resolve(true); + if (child.signalCode) return Promise.resolve(true); + return new Promise((resolve) => { + const finish = (closed: boolean) => { + clearTimeout(timer); + child.removeListener("close", onClose); + resolve(closed); + }; + const onClose = () => finish(true); + const timer = setTimeout(() => finish(false), timeoutMs); + child.once("close", onClose); + }); +} From de05f2cff6c94dc7ef68799bb87a8e01d647e828 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Thu, 16 Jul 2026 22:36:37 +0530 Subject: [PATCH 4/5] feat(agents): stream provider lifecycle events Co-Authored-By: Claude --- src/local-agent-adapters.ts | 880 +++++++++++++++++++++++++++--------- 1 file changed, 654 insertions(+), 226 deletions(-) diff --git a/src/local-agent-adapters.ts b/src/local-agent-adapters.ts index 457b8e08..5fcf8aca 100644 --- a/src/local-agent-adapters.ts +++ b/src/local-agent-adapters.ts @@ -1,17 +1,30 @@ import { spawn, spawnSync, type ChildProcessWithoutNullStreams } from "node:child_process"; -import { resolve } from "node:path"; +import { realpath } from "node:fs/promises"; +import { dirname, isAbsolute, relative, resolve } from "node:path"; import { Readable, Writable } from "node:stream"; -import type { EffortLevel } from "@anthropic-ai/claude-agent-sdk"; +import type { + CanUseTool, + EffortLevel, + Options as ClaudeOptions, + Query, + SDKMessage, +} from "@anthropic-ai/claude-agent-sdk"; import type { LocalAgentProvider } from "./local-agent-profiles.js"; import { removeDevspaceNodeModulesBinFromPath } from "./local-agent-path.js"; import { createCodexSdkLocalAgentRuntime, + LocalAgentRunController, + runLocalAgentHandle, + type LocalAgentRunHandle, type LocalAgentRunInput, type LocalAgentRunResult, } from "./local-agent-runtime.js"; +import { terminateProcessTreeGracefully } from "./process-platform.js"; +import type { EffectiveLocalAgentPolicy } from "./workflows/policy.js"; export interface LocalAgentAdapter { readonly provider: LocalAgentProvider; + start(input: LocalAgentRunInput): Promise; run(input: LocalAgentRunInput): Promise; } @@ -19,13 +32,16 @@ const ACP_COMMANDS: Record<"cursor" | "copilot", [string, ...string[]]> = { cursor: ["cursor-agent", "acp"], copilot: ["copilot", "--acp"], }; +const MAX_RESULT_ITEMS = 256; +const MAX_STDERR_CHARS = 16_384; +const PROCESS_SHUTDOWN_GRACE_MS = 1_000; const PI_AGENT_TIMEOUT_MS = 120_000; export async function runLocalAgentProvider( provider: LocalAgentProvider, input: LocalAgentRunInput, ): Promise { - return createLocalAgentAdapter(provider).run(input); + return runLocalAgentHandle(await createLocalAgentAdapter(provider).start(input)); } export function createLocalAgentAdapter(provider: LocalAgentProvider): LocalAgentAdapter { @@ -44,74 +60,180 @@ export function createLocalAgentAdapter(provider: LocalAgentProvider): LocalAgen } } -class CodexLocalAgentAdapter implements LocalAgentAdapter { - readonly provider = "codex" as const; +abstract class BaseLocalAgentAdapter implements LocalAgentAdapter { + abstract readonly provider: LocalAgentProvider; + abstract start(input: LocalAgentRunInput): Promise; async run(input: LocalAgentRunInput): Promise { - const runtime = await createCodexSdkLocalAgentRuntime(); - return runtime.run(input); + return runLocalAgentHandle(await this.start(input)); + } +} + +class CodexLocalAgentAdapter extends BaseLocalAgentAdapter { + readonly provider = "codex" as const; + + async start(input: LocalAgentRunInput): Promise { + const runtime = await createCodexSdkLocalAgentRuntime( + input.policy ? { env: { ...input.policy.environment } } : undefined, + ); + return runtime.start(input); } } -class ClaudeLocalAgentAdapter implements LocalAgentAdapter { +export type ClaudeQueryLike = AsyncGenerator & Pick; +export type ClaudeQueryFactory = (parameters: { + prompt: string; + options?: ClaudeOptions; +}) => ClaudeQueryLike; + +export class ClaudeLocalAgentAdapter extends BaseLocalAgentAdapter { readonly provider = "claude" as const; - async run(input: LocalAgentRunInput): Promise { - const { query } = await import("@anthropic-ai/claude-agent-sdk"); - const claudeExecutable = process.env.CLAUDE_COMMAND ?? resolveExecutable("claude"); - const messages = query({ + constructor(private readonly queryFactory?: ClaudeQueryFactory) { + super(); + } + + async start(input: LocalAgentRunInput): Promise { + const queryFactory = this.queryFactory ?? (await defaultClaudeQueryFactory()); + const abortController = new AbortController(); + const controller = new LocalAgentRunController(this.provider, input.signal); + const environment = input.policy?.environment ?? process.env; + const claudeExecutable = environment.CLAUDE_COMMAND ?? resolveExecutable("claude"); + const policyOptions = claudePolicyOptions(input, controller); + const query = queryFactory({ prompt: input.prompt, options: { cwd: input.workspace, model: input.model, - ...(input.thinking ? { thinking: { type: "adaptive" } as const, effort: input.thinking as EffortLevel } : {}), + ...(input.thinking + ? { thinking: { type: "adaptive" } as const, effort: input.thinking as EffortLevel } + : {}), resume: input.providerSessionId, - permissionMode: "bypassPermissions", - allowDangerouslySkipPermissions: true, - env: claudeCommandEnvironment(process.env), + includePartialMessages: true, + abortController, + env: claudeCommandEnvironment(environment), ...(claudeExecutable ? { pathToClaudeCodeExecutable: claudeExecutable } : {}), + ...policyOptions, }, }); - let providerSessionId = input.providerSessionId ?? null; - let finalResponse = ""; - const items: unknown[] = []; - for await (const message of messages) { - items.push(message); - const record = message as Record; - if (typeof record.session_id === "string") providerSessionId = record.session_id; - if (record.type === "result" && typeof record.result === "string") { - const resultError = claudeResultError(record); - if (resultError) throw new Error(resultError); - finalResponse = record.result; + const pump = (async () => { + let providerSessionId = input.providerSessionId ?? null; + let emittedSessionId: string | null = null; + let finalResponse = ""; + const items: unknown[] = []; + try { + for await (const message of query) { + pushBounded(items, message); + if (message.session_id && message.session_id !== emittedSessionId) { + const sessionId = message.session_id; + providerSessionId = sessionId; + emittedSessionId = sessionId; + controller.emit({ + type: "session", + providerSessionId: sessionId, + resumed: Boolean(input.providerSessionId), + }); + } + emitClaudeMessage(controller, message); + if (message.type !== "result") continue; + if (message.subtype === "success") { + finalResponse = message.result; + } else { + throw new Error( + `Claude returned an error result (${message.subtype}): ${message.errors.join("; ") || "unknown error"}`, + ); + } + } + controller.succeed({ + provider: this.provider, + providerSessionId, + finalResponse: requireFinalResponse("Claude", finalResponse), + items, + }); + } catch (error) { + controller.fail(error); } - } + })(); + + controller.setLifecycle({ + cancel: async (reason) => { + await withTimeout(query.interrupt(), 1_000).catch(() => undefined); + abortController.abort(reason); + }, + dispose: async () => { + abortController.abort(); + query.close(); + }, + pump, + }); + return controller; + } +} - finalResponse = requireFinalResponse("Claude", finalResponse); +async function defaultClaudeQueryFactory(): Promise { + const module = await import("@anthropic-ai/claude-agent-sdk"); + return module.query as ClaudeQueryFactory; +} + +function claudePolicyOptions( + input: LocalAgentRunInput, + controller: LocalAgentRunController, +): Partial { + if (input.policy?.mode !== "workflow") { return { - provider: this.provider, - providerSessionId, - finalResponse, - items, + permissionMode: "bypassPermissions", + allowDangerouslySkipPermissions: true, }; } -} -function claudeResultError(record: Record): string | undefined { - const subtype = typeof record.subtype === "string" ? record.subtype : undefined; - const isError = record.is_error === true || subtype?.startsWith("error"); - if (!isError) return undefined; - const message = - directString(record.error) ?? - directString(record.message) ?? - directString(record.result) ?? - subtype ?? - "Claude returned an error result."; - return `Claude returned an error result: ${message}`; + const writable = input.policy.access === "workspace_write"; + const allowedTools = new Set(writable + ? ["Read", "Glob", "Grep", "Edit", "Write", "NotebookEdit"] + : ["Read", "Glob", "Grep"]); + const canUseTool: CanUseTool = async (toolName, toolInput) => { + controller.emit({ type: "permission", phase: "requested", tool: toolName }); + const path = directString(toolInput.file_path) ?? + directString(toolInput.notebook_path) ?? + directString(toolInput.path); + const pathAllowed = !path || await isCanonicalPathInsideWorkspace(input.workspace, path); + if (allowedTools.has(toolName) && pathAllowed) { + controller.emit({ type: "permission", phase: "allowed", tool: toolName }); + return { behavior: "allow", updatedInput: toolInput }; + } + controller.emit({ type: "permission", phase: "denied", tool: toolName }); + return { behavior: "deny", message: "Tool is not permitted by workflow policy." }; + }; + return { + permissionMode: "dontAsk", + tools: [...allowedTools], + canUseTool, + sandbox: { enabled: true, failIfUnavailable: true }, + settingSources: [], + strictMcpConfig: true, + mcpServers: {}, + plugins: [], + skills: [], + agents: {}, + }; } -function directString(value: unknown): string | undefined { - return typeof value === "string" && value.trim() ? value.trim() : undefined; +function emitClaudeMessage(controller: LocalAgentRunController, message: SDKMessage): void { + if (message.type === "stream_event") { + const event = asRecord(message.event); + const delta = asRecord(event?.delta); + if (event?.type === "content_block_delta" && delta?.type === "text_delta" && typeof delta.text === "string") { + controller.emit({ type: "output", stream: "assistant", delta: delta.text }); + } else if (event?.type === "content_block_delta" && delta?.type === "thinking_delta" && typeof delta.thinking === "string") { + controller.emit({ type: "output", stream: "thinking", delta: delta.thinking }); + } + return; + } + if (message.type === "tool_progress") { + controller.emit({ type: "tool", phase: "updated", name: message.tool_name, id: message.tool_use_id }); + } else if (message.type === "permission_denied") { + controller.emit({ type: "permission", phase: "denied", tool: message.tool_name }); + } } function resolveExecutable(command: string): string | undefined { @@ -125,7 +247,7 @@ function resolveExecutable(command: string): string | undefined { return executable?.trim() || undefined; } -export function claudeCommandEnvironment(env: NodeJS.ProcessEnv): NodeJS.ProcessEnv { +export function claudeCommandEnvironment(env: NodeJS.ProcessEnv | Readonly>): NodeJS.ProcessEnv { const next = { ...env }; for (const key of [ "CLAUDECODE", @@ -138,113 +260,315 @@ export function claudeCommandEnvironment(env: NodeJS.ProcessEnv): NodeJS.Process return next; } -class OpencodeLocalAgentAdapter implements LocalAgentAdapter { +class OpencodeLocalAgentAdapter extends BaseLocalAgentAdapter { readonly provider = "opencode" as const; - async run(input: LocalAgentRunInput): Promise { - const { createOpencode } = await import("@opencode-ai/sdk/v2"); - const { client, server } = await createOpencode(); - try { - const sessionId = input.providerSessionId ?? await createOpencodeSession(client, input); - const promptResult = await promptOpencodeSession(client, sessionId, input); - await waitForOpencodeSession(client, sessionId); - const messages = await readOpencodeMessages(client, sessionId); - const finalResponse = requireFinalResponse( - "OpenCode", - extractOpenCodeFinalResponse(messages) || extractOpenCodeFinalResponse(promptResult), - ); - return { - provider: this.provider, - providerSessionId: sessionId, - finalResponse, - items: [promptResult, messages], - }; - } finally { - server.close(); + async start(input: LocalAgentRunInput): Promise { + if (input.policy?.mode === "workflow") { + throw new Error("OpenCode cannot currently enforce DevSpace workflow filesystem policy."); } + const { createOpencode } = await import("@opencode-ai/sdk/v2"); + const abortController = new AbortController(); + const controller = new LocalAgentRunController(this.provider, input.signal); + const { client, server } = await createOpencode({ signal: abortController.signal }); + const session = client.v2.session; + let subscriptionStream: AsyncGenerator | undefined; + let sessionId: string | undefined; + + const pump = (async () => { + const items: unknown[] = []; + try { + sessionId = input.providerSessionId ?? await createOpencodeSession(client, input); + if (input.providerSessionId) { + await session.get({ sessionID: sessionId }, { throwOnError: true }); + if (input.model) { + await session.switchModel( + { sessionID: sessionId, model: parseOpencodeV2Model(input.model, input.thinking) }, + { throwOnError: true }, + ); + } + } + controller.emit({ + type: "session", + providerSessionId: sessionId, + resumed: Boolean(input.providerSessionId), + }); + const subscription = await client.v2.event.subscribe({ + signal: abortController.signal, + sseMaxRetryAttempts: 1, + onSseError: (error: unknown) => { + controller.emit({ type: "warning", message: `OpenCode event stream error: ${errorMessage(error)}` }); + }, + }); + subscriptionStream = subscription.stream as AsyncGenerator; + const eventPump = (async () => { + for await (const event of subscriptionStream ?? []) { + if (!isOpencodeSessionEvent(event, sessionId as string)) continue; + pushBounded(items, event); + emitOpencodeEvent(controller, event); + } + })(); + const promptResult = await session.prompt({ + sessionID: sessionId, + prompt: { text: input.prompt }, + delivery: "queue", + resume: true, + }, { throwOnError: true }); + pushBounded(items, promptResult); + await session.wait({ sessionID: sessionId }, { throwOnError: true }); + const messages = await session.messages( + { sessionID: sessionId, order: "desc", limit: 100 }, + { throwOnError: true }, + ); + pushBounded(items, messages); + abortController.abort(); + await eventPump.catch(() => undefined); + const finalResponse = requireFinalResponse("OpenCode", extractOpenCodeFinalResponse(promptResult)); + controller.succeed({ + provider: this.provider, + providerSessionId: sessionId, + finalResponse, + items, + }); + } catch (error) { + controller.fail(error); + } + })(); + + controller.setLifecycle({ + cancel: async () => { + if (sessionId) { + await session.interrupt({ sessionID: sessionId }, { throwOnError: true }).catch(() => undefined); + } + abortController.abort(); + }, + dispose: async () => { + abortController.abort(); + await subscriptionStream?.return(undefined).catch(() => undefined); + server.close(); + }, + pump, + }); + return controller; } } -class AcpLocalAgentAdapter implements LocalAgentAdapter { +class AcpLocalAgentAdapter extends BaseLocalAgentAdapter { constructor( readonly provider: "cursor" | "copilot", private readonly command: [string, ...string[]], - ) {} + ) { + super(); + } - async run(input: LocalAgentRunInput): Promise { - const { client } = await import("@agentclientprotocol/sdk"); - const { methods } = await import("@agentclientprotocol/sdk"); - const { ndJsonStream } = await import("@agentclientprotocol/sdk"); + async start(input: LocalAgentRunInput): Promise { + if (input.policy?.mode === "workflow") { + throw new Error(`${this.provider} ACP cannot currently enforce DevSpace workflow filesystem policy.`); + } + const { client, methods, ndJsonStream, PROTOCOL_VERSION } = await import("@agentclientprotocol/sdk"); const [command, ...args] = this.command; + const detached = process.platform !== "win32"; const child = spawn(command, args, { cwd: input.workspace, - env: process.env, + env: { ...(input.policy?.environment ?? process.env) }, stdio: ["pipe", "pipe", "pipe"], windowsHide: true, + detached, }); assertPipedChild(child); + const controller = new LocalAgentRunController(this.provider, input.signal); let stderr = ""; + let context: { notify(method: string, params: unknown): Promise } | undefined; + let providerSessionId = input.providerSessionId ?? null; + let replaying = Boolean(input.providerSessionId); + let disposing = false; + const items: unknown[] = []; + const textParts: string[] = []; child.stderr.on("data", (chunk: Buffer) => { - stderr += chunk.toString("utf8"); + stderr = appendBounded(stderr, chunk.toString("utf8")); }); - const stream = ndJsonStream( Writable.toWeb(child.stdin) as WritableStream, Readable.toWeb(child.stdout) as ReadableStream, ); - try { - let providerSessionId = input.providerSessionId ?? null; - const finalResponse = await client({ name: "DevSpace" }) - .onRequest(methods.client.session.requestPermission, (context) => { - const selected = selectAcpAllowPermissionOption(context.params.options); - return selected - ? { outcome: { outcome: "selected", optionId: selected.optionId } } - : { outcome: { outcome: "cancelled" } }; - }) - .connectWith(stream, async (context) => { - const session = await context.buildSession(input.workspace).start(); - providerSessionId = session.sessionId; - try { - if (input.model) { - const config = resolveAcpModelConfigUpdate(session, input.model, this.provider); - await context.request(methods.agent.session.setConfigOption, config); - } - if (input.thinking) { - const config = resolveAcpThinkingConfigUpdate(session, input.thinking, this.provider); - await context.request(methods.agent.session.setConfigOption, config); - } - const prompt = session.prompt(input.prompt); - const textParts: string[] = []; - for (;;) { - const message = await session.nextUpdate(); - if (message.kind === "stop") { - await prompt; - return textParts.join("").trim(); - } - - const update = message.update; - if (update.sessionUpdate !== "agent_message_chunk") continue; - const content = update.content; - if (content.type === "text") textParts.push(content.text); - } - } finally { - session.dispose(); + const app = client({ name: "DevSpace" }) + .onRequest(methods.client.session.requestPermission, ({ params }) => { + const tool = asRecord(params.toolCall)?.kind as string | undefined; + controller.emit({ type: "permission", phase: "requested", tool }); + const outcome = resolveAcpPermissionRequest(params, input.policy); + controller.emit({ + type: "permission", + phase: outcome.response.outcome.outcome === "selected" && outcome.allowed ? "allowed" : "denied", + tool, + }); + return outcome.response; + }) + .onNotification(methods.client.session.update, ({ params }) => { + if (params.sessionId !== providerSessionId || replaying) return; + pushBounded(items, params); + emitAcpUpdate(controller, params.update, textParts); + }); + + const pump = app.connectWith(stream, async (activeContext) => { + context = activeContext; + try { + const initialized = await activeContext.request(methods.agent.initialize, { + protocolVersion: PROTOCOL_VERSION, + clientCapabilities: {}, + clientInfo: { name: "DevSpace", version: "1" }, + }); + if (initialized.protocolVersion !== PROTOCOL_VERSION) { + throw new Error(`${this.provider} ACP protocol version ${initialized.protocolVersion} is unsupported.`); + } + let sessionResponse: unknown; + if (input.providerSessionId) { + const capabilities = asRecord(initialized.agentCapabilities); + if (capabilities?.loadSession !== true) { + throw new Error(`${this.provider} ACP does not support loading persisted sessions.`); } + sessionResponse = await activeContext.request(methods.agent.session.load, { + sessionId: input.providerSessionId, + cwd: input.workspace, + mcpServers: [], + }); + providerSessionId = input.providerSessionId; + replaying = false; + } else { + sessionResponse = await activeContext.request(methods.agent.session.new, { + cwd: input.workspace, + mcpServers: [], + }); + providerSessionId = directString(asRecord(sessionResponse)?.sessionId) ?? null; + } + if (!providerSessionId) throw new Error(`${this.provider} ACP did not return a session id.`); + const sessionId = providerSessionId; + controller.emit({ + type: "session", + providerSessionId: sessionId, + resumed: Boolean(input.providerSessionId), }); - return { - provider: this.provider, - providerSessionId, - finalResponse: finalResponse.trim(), - items: [], - }; - } catch (error) { - throw new Error(`${this.provider} ACP run failed: ${errorMessage(error)}${stderr ? `\n${stderr.trim()}` : ""}`); - } finally { - child.kill(); + let sessionMetadata = { sessionId, newSessionResponse: sessionResponse }; + if (input.model) { + const config = resolveAcpModelConfigUpdate(sessionMetadata, input.model, this.provider); + const response = await activeContext.request(methods.agent.session.setConfigOption, config); + sessionMetadata = { sessionId, newSessionResponse: response }; + } + if (input.thinking) { + const config = resolveAcpThinkingConfigUpdate(sessionMetadata, input.thinking, this.provider); + const response = await activeContext.request(methods.agent.session.setConfigOption, config); + sessionMetadata = { sessionId, newSessionResponse: response }; + } + const promptResponse = asRecord(await activeContext.request(methods.agent.session.prompt, { + sessionId, + prompt: [{ type: "text", text: input.prompt }], + })); + const stopReason = directString(promptResponse?.stopReason); + if (stopReason !== "end_turn") { + throw new Error(`${this.provider} ACP stopped with ${stopReason ?? "unknown"}.`); + } + controller.succeed({ + provider: this.provider, + providerSessionId: sessionId, + finalResponse: requireFinalResponse(this.provider, textParts.join("")), + items, + }); + } catch (error) { + if (!disposing) { + const suffix = stderr.trim() ? `\n${stderr.trim()}` : ""; + controller.fail(new Error(`${this.provider} ACP run failed: ${errorMessage(error)}${suffix}`)); + } + } + }); + + child.once("error", (error) => controller.fail(error)); + child.once("exit", (code, signal) => { + if (!disposing && code !== 0) { + controller.fail(new Error(`${this.provider} ACP process exited with code ${code ?? "null"} and signal ${signal ?? "null"}.`)); + } + }); + controller.setLifecycle({ + cancel: async () => { + if (context && providerSessionId) { + await context.notify(methods.agent.session.cancel, { sessionId: providerSessionId }).catch(() => undefined); + } + disposing = true; + await terminateProcessTreeGracefully(child, detached, PROCESS_SHUTDOWN_GRACE_MS); + }, + dispose: async () => { + disposing = true; + await terminateProcessTreeGracefully(child, detached, PROCESS_SHUTDOWN_GRACE_MS); + }, + pump: pump.then(() => undefined).catch((error) => controller.fail(error)), + }); + return controller; + } +} + +export function resolveAcpPermissionRequest( + params: { toolCall: unknown; options: Array<{ optionId: string; kind: string }> }, + policy?: EffectiveLocalAgentPolicy, +): { + allowed: boolean; + outcome: { outcome: { outcome: string; optionId?: string } }; + response: { outcome: { outcome: "selected"; optionId: string } | { outcome: "cancelled" } }; +} { + const toolKind = directString(asRecord(params.toolCall)?.kind) ?? "other"; + const compatibility = policy?.mode !== "workflow"; + const readOnlyKinds = new Set(["read", "search", "fetch", "think"]); + const workspaceWriteKinds = new Set([...readOnlyKinds, "edit", "delete", "move"]); + const allowed = compatibility || (policy?.access === "workspace_write" + ? workspaceWriteKinds.has(toolKind) + : readOnlyKinds.has(toolKind)); + const desiredKinds = allowed + ? compatibility ? ["allow_once", "allow_always"] : ["allow_once"] + : ["reject_once", "reject_always"]; + const selected = desiredKinds + .map((kind) => params.options.find((option) => option.kind === kind)) + .find((option) => option !== undefined); + if (!selected) { + const response = { outcome: { outcome: "cancelled" as const } }; + return { allowed: false, outcome: response, response }; + } + const response = { outcome: { outcome: "selected" as const, optionId: selected.optionId } }; + return { allowed, outcome: response, response }; +} + +function emitAcpUpdate( + controller: LocalAgentRunController, + update: unknown, + textParts: string[], +): void { + const record = asRecord(update); + const kind = directString(record?.sessionUpdate); + if (kind === "agent_message_chunk") { + const content = asRecord(record?.content); + if (content?.type === "text" && typeof content.text === "string") { + textParts.push(content.text); + controller.emit({ type: "output", stream: "assistant", delta: content.text }); } + } else if (kind === "agent_thought_chunk") { + const content = asRecord(record?.content); + if (content?.type === "text" && typeof content.text === "string") { + controller.emit({ type: "output", stream: "thinking", delta: content.text }); + } + } else if (kind === "tool_call" || kind === "tool_call_update") { + controller.emit({ + type: "tool", + phase: kind === "tool_call" ? "started" : acpToolPhase(record?.status), + id: directString(record?.toolCallId), + name: directString(record?.title) ?? directString(record?.kind), + }); + } else if (kind) { + controller.emit({ type: "warning", message: `ACP update: ${kind}` }); } } +function acpToolPhase(status: unknown): "updated" | "completed" | "failed" { + if (status === "completed") return "completed"; + if (status === "failed") return "failed"; + return "updated"; +} + export function resolveAcpModelConfigUpdate( session: unknown, model: string, @@ -323,63 +647,116 @@ function flattenAcpSelectValues(option: Record): string[] { return values; } -function selectAcpAllowPermissionOption(options: Array<{ optionId: string; kind: string }>): { optionId: string } | undefined { - return ( - options.find((option) => option.kind === "allow_once") ?? - options.find((option) => option.kind === "allow_always") - ); -} - -class PiRpcLocalAgentAdapter implements LocalAgentAdapter { +class PiRpcLocalAgentAdapter extends BaseLocalAgentAdapter { readonly provider = "pi" as const; - async run(input: LocalAgentRunInput): Promise { + async start(input: LocalAgentRunInput): Promise { + if (input.policy?.mode === "workflow") { + throw new Error("Pi RPC cannot currently enforce DevSpace workflow filesystem policy."); + } const args = ["--mode", "rpc"]; if (input.model) args.push("--model", input.model); if (input.thinking) args.push("--thinking", input.thinking); if (input.providerSessionId) args.push("--session", input.providerSessionId); - const child = spawn(process.env.PI_COMMAND ?? "pi", args, { + const detached = process.platform !== "win32"; + const environment = input.policy?.environment ?? process.env; + const child = spawn(environment.PI_COMMAND ?? "pi", args, { cwd: input.workspace, - env: piCommandEnvironment(process.env), + env: piCommandEnvironment(environment), stdio: ["pipe", "pipe", "pipe"], windowsHide: true, + detached, }); assertPipedChild(child); + const controller = new LocalAgentRunController(this.provider, input.signal); const rpc = new JsonLineRpc(child); const events: unknown[] = []; - rpc.onEvent((event) => events.push(event)); - try { - const state = await rpc.request({ type: "get_state" }); - const providerSessionId = readNestedString(state, ["sessionId"]) ?? input.providerSessionId ?? null; - const done = rpc.waitForEvent((event) => asRecord(event)?.type === "agent_end", PI_AGENT_TIMEOUT_MS); - await rpc.request({ type: "prompt", message: input.prompt }); - const agentEnd = await done; - const sessionMessages = await rpc.request({ type: "get_messages" }); - const finalResponse = - extractPiFinalResponse(agentEnd) || - extractPiFinalResponse(sessionMessages) || - extractPiStreamingText(events); - if (!finalResponse) { + let disposing = false; + rpc.onEvent((event) => { + pushBounded(events, event); + emitPiEvent(controller, event); + }); + + const pump = (async () => { + try { + const state = await rpc.request({ type: "get_state" }); + const providerSessionId = readNestedString(state, ["sessionId"]) ?? input.providerSessionId ?? null; + if (providerSessionId) { + controller.emit({ + type: "session", + providerSessionId, + resumed: Boolean(input.providerSessionId), + }); + } + const done = rpc.waitForEvent((event) => { + const record = asRecord(event); + return record?.type === "agent_end" && record.willRetry !== true; + }, PI_AGENT_TIMEOUT_MS); + await rpc.request({ type: "prompt", message: input.prompt }); + const agentEnd = await done; + const sessionMessages = await rpc.request({ type: "get_messages" }); const providerError = extractPiProviderError(agentEnd) || extractPiProviderError(sessionMessages) || extractPiProviderError(events); if (providerError) throw new Error(`Pi returned an error: ${providerError}`); + const finalResponse = requireFinalResponse( + "Pi", + extractPiFinalResponse(agentEnd) || + extractPiFinalResponse(sessionMessages) || + extractPiStreamingText(events), + ); + pushBounded(events, sessionMessages); + controller.succeed({ + provider: this.provider, + providerSessionId, + finalResponse, + items: events, + }); + } catch (error) { + if (!disposing) controller.fail(error); } - requireFinalResponse("Pi", finalResponse); - return { - provider: this.provider, - providerSessionId, - finalResponse, - items: [...events, sessionMessages], - }; - } finally { - child.kill(); + })(); + + controller.setLifecycle({ + cancel: async () => { + await withTimeout(rpc.request({ type: "abort" }), 500).catch(() => undefined); + disposing = true; + await terminateProcessTreeGracefully(child, detached, PROCESS_SHUTDOWN_GRACE_MS); + }, + dispose: async () => { + disposing = true; + await terminateProcessTreeGracefully(child, detached, PROCESS_SHUTDOWN_GRACE_MS); + }, + pump, + }); + return controller; + } +} + +function emitPiEvent(controller: LocalAgentRunController, event: unknown): void { + const record = asRecord(event); + if (!record) return; + if (record.type === "message_update") { + const update = asRecord(record.assistantMessageEvent); + if (update?.type === "text_delta" && typeof update.delta === "string") { + controller.emit({ type: "output", stream: "assistant", delta: update.delta }); + } else if (update?.type === "thinking_delta" && typeof update.delta === "string") { + controller.emit({ type: "output", stream: "thinking", delta: update.delta }); } + } else if (record.type === "tool_execution_start" || record.type === "tool_execution_update" || record.type === "tool_execution_end") { + controller.emit({ + type: "tool", + phase: record.type === "tool_execution_start" ? "started" : record.type === "tool_execution_end" ? "completed" : "updated", + id: directString(record.toolCallId), + name: directString(record.toolName), + }); + } else if (record.type === "extension_error") { + controller.emit({ type: "warning", message: directString(record.message) ?? "Pi extension error." }); } } -export function piCommandEnvironment(env: NodeJS.ProcessEnv): NodeJS.ProcessEnv { +export function piCommandEnvironment(env: NodeJS.ProcessEnv | Readonly>): NodeJS.ProcessEnv { if (env.PI_COMMAND) return env; const path = env.PATH; if (!path) return env; @@ -396,6 +773,7 @@ class JsonLineRpc { reject: (error: Error) => void; }>(); private readonly eventSubscribers = new Set<(event: unknown) => void>(); + private readonly eventWaiterRejectors = new Set<(error: Error) => void>(); private buffer = ""; private nextId = 1; private stderr = ""; @@ -404,8 +782,9 @@ class JsonLineRpc { constructor(private readonly child: ChildProcessWithoutNullStreams) { child.stdout.on("data", (chunk: Buffer) => this.handleStdout(chunk.toString("utf8"))); child.stderr.on("data", (chunk: Buffer) => { - this.stderr += chunk.toString("utf8"); + this.stderr = appendBounded(this.stderr, chunk.toString("utf8")); }); + child.on("error", (error) => this.failAll(error)); child.on("exit", (code, signal) => { this.failAll(new Error(`Pi RPC process exited with code ${code ?? "null"} and signal ${signal ?? "null"}\n${this.stderr}`.trim())); }); @@ -429,17 +808,26 @@ class JsonLineRpc { } waitForEvent(predicate: (event: unknown) => boolean, timeoutMs: number): Promise { + if (this.fatalError) return Promise.reject(this.fatalError); return new Promise((resolve, reject) => { - const timer = setTimeout(() => { + const finish = () => { + clearTimeout(timer); unsubscribe(); - reject(new Error(`Pi RPC timed out waiting for agent completion\n${this.stderr}`.trim())); + this.eventWaiterRejectors.delete(rejectWaiter); + }; + const rejectWaiter = (error: Error) => { + finish(); + reject(error); + }; + const timer = setTimeout(() => { + rejectWaiter(new Error(`Pi RPC timed out waiting for agent completion\n${this.stderr}`.trim())); }, timeoutMs); const unsubscribe = this.onEvent((event) => { if (!predicate(event)) return; - clearTimeout(timer); - unsubscribe(); + finish(); resolve(event); }); + this.eventWaiterRejectors.add(rejectWaiter); }); } @@ -455,7 +843,7 @@ class JsonLineRpc { try { message = JSON.parse(line) as Record; } catch { - this.stderr += `${line}\n`; + this.stderr = appendBounded(this.stderr, `${line}\n`); this.failAll(new Error(`Pi RPC emitted malformed JSON on stdout: ${line}`)); return; } @@ -478,82 +866,60 @@ class JsonLineRpc { } private failAll(error: Error): void { + if (this.fatalError) return; this.fatalError = error; - for (const pending of this.pending.values()) { - pending.reject(error); - } + for (const pending of this.pending.values()) pending.reject(error); this.pending.clear(); + for (const reject of this.eventWaiterRejectors) reject(error); + this.eventWaiterRejectors.clear(); } } async function createOpencodeSession(client: unknown, input: LocalAgentRunInput): Promise { - const sessionClient = client as { - session: { - create(parameters?: unknown, options?: unknown): Promise; + const session = (client as { + v2: { + session: { + create(parameters?: unknown, options?: unknown): Promise; + }; }; - }; - const result = await sessionClient.session.create({ - directory: input.workspace, + }).v2.session; + const result = await session.create({ location: { directory: input.workspace }, - ...(input.model ? { model: parseOpencodeModel(input.model) } : {}), + ...(input.model ? { model: parseOpencodeV2Model(input.model, input.thinking) } : {}), }, { throwOnError: true }); const id = - readNestedString(result, ["id"]) ?? + readNestedString(result, ["data", "data", "id"]) ?? readNestedString(result, ["data", "id"]) ?? - readNestedString(result, ["session", "id"]) ?? - readNestedString(result, ["data", "session", "id"]); - if (typeof id !== "string") { - throw new Error("OpenCode did not return a session id."); - } + readNestedString(result, ["id"]); + if (!id) throw new Error("OpenCode did not return a session id."); return id; } -async function promptOpencodeSession( - client: unknown, - sessionId: string, - input: LocalAgentRunInput, -): Promise { - const session = (client as { - session: { - prompt(parameters?: unknown, options?: unknown): Promise; - }; - }).session; - const promptInput = { - sessionID: sessionId, - directory: input.workspace, - prompt: { parts: [{ type: "text", text: input.prompt }] }, - parts: [{ type: "text", text: input.prompt }], - ...(input.model ? { model: parseOpencodeModel(input.model) } : {}), - ...(input.thinking ? { variant: input.thinking } : {}), - }; - return session.prompt(promptInput, { throwOnError: true }); -} - -async function waitForOpencodeSession(client: unknown, sessionId: string): Promise { - const session = (client as { - session?: { wait?: (parameters?: unknown, options?: unknown) => Promise }; - }).session; - if (!session?.wait) return; - await session.wait({ sessionID: sessionId }, { throwOnError: true }); +function parseOpencodeV2Model( + model: string, + variant?: string, +): { providerID: string; id: string; variant?: string } { + const separator = model.indexOf("/"); + const parsed = separator === -1 + ? { providerID: "opencode", id: model } + : { providerID: model.slice(0, separator), id: model.slice(separator + 1) }; + return variant ? { ...parsed, variant } : parsed; } -async function readOpencodeMessages(client: unknown, sessionId: string): Promise { - const session = (client as { - session?: { - messages?: (parameters?: unknown, options?: unknown) => Promise; - }; - }).session; - if (!session?.messages) return undefined; - return session.messages({ sessionID: sessionId, order: "asc", limit: 100 }, { throwOnError: true }); +function isOpencodeSessionEvent(event: unknown, sessionId: string): boolean { + const data = asRecord(asRecord(event)?.data); + const eventSessionId = directString(data?.sessionID); + return !eventSessionId || eventSessionId === sessionId; } -function parseOpencodeModel(model: string): { providerID: string; modelID: string } { - const separator = model.indexOf("/"); - if (separator === -1) return { providerID: "opencode", modelID: model }; - return { - providerID: model.slice(0, separator), - modelID: model.slice(separator + 1), - }; +function emitOpencodeEvent(controller: LocalAgentRunController, event: unknown): void { + const record = asRecord(event); + const data = asRecord(record?.data); + if (record?.type === "session.next.text.delta" && typeof data?.delta === "string") { + controller.emit({ type: "output", stream: "assistant", delta: data.delta }); + } else if (record?.type === "session.next.step.failed" || record?.type === "session.error") { + controller.emit({ type: "warning", message: `OpenCode event: ${record.type}` }); + } } export function extractLocalAgentResponseText(value: unknown): string { @@ -690,9 +1056,15 @@ function stringifyStructuredAssistantMessage(value: unknown): string { } function unwrapProviderPayload(value: unknown): unknown { - const record = asRecord(value); - if (!record) return value; - return record.data ?? record.result ?? value; + let current = value; + for (let depth = 0; depth < 4; depth += 1) { + const record = asRecord(current); + if (!record) return current; + const next = record.data ?? record.result; + if (next === undefined || next === current) return current; + current = next; + } + return current; } function readArray(record: unknown, key: string): unknown[] | undefined { @@ -713,6 +1085,62 @@ function readNestedString(value: unknown, path: string[]): string | undefined { return typeof current === "string" ? current : undefined; } +function directString(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} + +function pushBounded(items: unknown[], item: unknown): void { + if (items.length >= MAX_RESULT_ITEMS) items.shift(); + items.push(item); +} + +function appendBounded(current: string, addition: string): string { + const combined = current + addition; + return combined.length > MAX_STDERR_CHARS + ? combined.slice(combined.length - MAX_STDERR_CHARS) + : combined; +} + +function withTimeout(promise: Promise, timeoutMs: number): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("Provider operation timed out.")), timeoutMs); + promise.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (error) => { + clearTimeout(timer); + reject(error); + }, + ); + }); +} + +async function isCanonicalPathInsideWorkspace(workspace: string, candidate: string): Promise { + let canonicalWorkspace: string; + try { + canonicalWorkspace = await realpath(workspace); + } catch { + return false; + } + + let existingPath = isAbsolute(candidate) ? resolve(candidate) : resolve(workspace, candidate); + for (;;) { + try { + const canonicalCandidate = await realpath(existingPath); + const path = relative(canonicalWorkspace, canonicalCandidate); + return path === "" || (!path.startsWith("..") && !isAbsolute(path)); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code !== "ENOENT" && code !== "ENOTDIR") return false; + const parent = dirname(existingPath); + if (parent === existingPath) return false; + existingPath = parent; + } + } +} + function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } From cea7b23ee29bbbcdb148b9b8c40b55bbb5835cc3 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Thu, 16 Jul 2026 22:36:45 +0530 Subject: [PATCH 5/5] test(agents): cover streaming lifecycle and policy Co-Authored-By: Claude --- package.json | 2 +- src/local-agent-adapters.test.ts | 189 +++++++++++++++++++++++++- src/local-agent-runtime.test.ts | 225 +++++++++++++++++++++++++++---- src/process-platform.test.ts | 60 ++++++++- src/workflows/policy.test.ts | 76 +++++++++++ 5 files changed, 521 insertions(+), 31 deletions(-) create mode 100644 src/workflows/policy.test.ts diff --git a/package.json b/package.json index 831aaab5..d6a10dfc 100644 --- a/package.json +++ b/package.json @@ -29,7 +29,7 @@ "postinstall": "node scripts/fix-node-pty-permissions.mjs", "start": "node dist/cli.js serve", "test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && npm run test:workflows && tsx src/cli.test.ts", - "test:workflows": "tsx src/workflows/migrations.test.ts && tsx src/workflows/workflows.test.ts", + "test:workflows": "tsx src/workflows/migrations.test.ts && tsx src/workflows/policy.test.ts && tsx src/workflows/workflows.test.ts", "typecheck": "tsc -p tsconfig.json --noEmit" }, "keywords": [], diff --git a/src/local-agent-adapters.test.ts b/src/local-agent-adapters.test.ts index 0e4cbf0f..d388166a 100644 --- a/src/local-agent-adapters.test.ts +++ b/src/local-agent-adapters.test.ts @@ -1,6 +1,9 @@ import assert from "node:assert/strict"; -import { delimiter } from "node:path"; +import { mkdtemp, mkdir, rm, symlink, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { delimiter, join } from "node:path"; import { + ClaudeLocalAgentAdapter, claudeCommandEnvironment, createLocalAgentAdapter, extractOpenCodeFinalResponse, @@ -9,6 +12,7 @@ import { extractPiStreamingText, piCommandEnvironment, resolveAcpModelConfigUpdate, + resolveAcpPermissionRequest, resolveAcpThinkingConfigUpdate, } from "./local-agent-adapters.js"; import { removeDevspaceNodeModulesBinFromPath } from "./local-agent-path.js"; @@ -26,6 +30,7 @@ const providers: LocalAgentProvider[] = [ for (const provider of providers) { const adapter = createLocalAgentAdapter(provider); assert.equal(adapter.provider, provider); + assert.equal(typeof adapter.start, "function"); assert.equal(typeof adapter.run, "function"); } @@ -388,3 +393,185 @@ assert.equal( assert.equal(env.PATH, [devspaceBin, "/home/user/.local/bin"].join(delimiter)); } + +{ + const readOnlyPolicy = { + version: 1, + mode: "workflow", + access: "read_only", + environment: Object.freeze({ PATH: "/usr/bin", HOME: "/home/user" }), + } as const; + const denied = resolveAcpPermissionRequest({ + toolCall: { kind: "edit" }, + options: [ + { optionId: "allow", kind: "allow_once" }, + { optionId: "reject", kind: "reject_once" }, + ], + }, readOnlyPolicy); + assert.equal(denied.allowed, false); + assert.deepEqual(denied.response, { + outcome: { outcome: "selected", optionId: "reject" }, + }); + + const read = resolveAcpPermissionRequest({ + toolCall: { kind: "read" }, + options: [{ optionId: "allow", kind: "allow_once" }], + }, readOnlyPolicy); + assert.equal(read.allowed, true); + assert.deepEqual(read.response, { + outcome: { outcome: "selected", optionId: "allow" }, + }); + + const noSafeOption = resolveAcpPermissionRequest({ + toolCall: { kind: "execute" }, + options: [{ optionId: "allow", kind: "allow_always" }], + }, readOnlyPolicy); + assert.deepEqual(noSafeOption.response, { outcome: { outcome: "cancelled" } }); +} + +{ + let capturedOptions: Record | undefined; + let closed = 0; + const query = (async function* () { + yield { + type: "stream_event", + session_id: "claude-session", + event: { + type: "content_block_delta", + delta: { type: "text_delta", text: "Hello " }, + }, + } as never; + yield { + type: "result", + subtype: "success", + session_id: "claude-session", + result: "Hello Claude", + } as never; + })() as AsyncGenerator & { interrupt(): Promise; close(): void }; + query.interrupt = async () => undefined; + query.close = () => { closed += 1; }; + const adapter = new ClaudeLocalAgentAdapter((parameters) => { + capturedOptions = parameters.options as unknown as Record; + return query as never; + }); + const handle = await adapter.start({ prompt: "hello", workspace: "/tmp/project" }); + const result = await handle.result(); + assert.equal(result.providerSessionId, "claude-session"); + assert.equal(result.finalResponse, "Hello Claude"); + assert.equal(capturedOptions?.includePartialMessages, true); + assert.equal(capturedOptions?.permissionMode, "bypassPermissions"); + const eventTypes: string[] = []; + for await (const event of handle.events()) eventTypes.push(event.type); + assert.deepEqual(eventTypes, ["lifecycle", "session", "output", "terminal"]); + await handle.dispose(); + await handle.dispose(); + assert.equal(closed, 1); +} + +{ + const query = (async function* () { + yield { + type: "result", + subtype: "error_during_execution", + session_id: "claude-error-session", + errors: ["provider unavailable"], + } as never; + })() as AsyncGenerator & { interrupt(): Promise; close(): void }; + query.interrupt = async () => undefined; + query.close = () => undefined; + const adapter = new ClaudeLocalAgentAdapter(() => query as never); + const handle = await adapter.start({ prompt: "hello", workspace: "/tmp/project" }); + await assert.rejects(handle.result(), /provider unavailable/); + await handle.dispose(); +} + +{ + const compatibility = resolveAcpPermissionRequest({ + toolCall: { kind: "edit" }, + options: [ + { optionId: "always", kind: "allow_always" }, + { optionId: "once", kind: "allow_once" }, + ], + }); + assert.deepEqual(compatibility.response, { + outcome: { outcome: "selected", optionId: "once" }, + }); +} + +{ + const workflowPolicy = { + version: 1, + mode: "workflow", + access: "workspace_write", + environment: Object.freeze({ PATH: process.env.PATH ?? "" }), + } as const; + await assert.rejects( + createLocalAgentAdapter("cursor").start({ + prompt: "edit", + workspace: "/tmp/project", + policy: workflowPolicy, + }), + /cannot currently enforce DevSpace workflow filesystem policy/, + ); +} + +{ + const root = await mkdtemp(join(tmpdir(), "devspace-claude-policy-")); + const workspace = join(root, "workspace"); + const outside = join(root, "outside"); + await mkdir(workspace); + await mkdir(outside); + await writeFile(join(outside, "secret.txt"), "secret"); + await symlink(outside, join(workspace, "escape")); + + let capturedOptions: Record | undefined; + const query = (async function* () { + yield { + type: "result", + subtype: "success", + session_id: "claude-workflow-session", + result: "ok", + } as never; + })() as AsyncGenerator & { interrupt(): Promise; close(): void }; + query.interrupt = async () => undefined; + query.close = () => undefined; + const originalClaudeCommand = process.env.CLAUDE_COMMAND; + process.env.CLAUDE_COMMAND = "/ambient/claude-must-not-run"; + try { + const adapter = new ClaudeLocalAgentAdapter((parameters) => { + capturedOptions = parameters.options as unknown as Record; + return query as never; + }); + const handle = await adapter.start({ + prompt: "inspect", + workspace, + policy: { + version: 1, + mode: "workflow", + access: "workspace_write", + environment: Object.freeze({ PATH: process.env.PATH ?? "" }), + }, + }); + await handle.result(); + assert.deepEqual(capturedOptions?.settingSources, []); + assert.equal(capturedOptions?.strictMcpConfig, true); + assert.deepEqual(capturedOptions?.mcpServers, {}); + assert.deepEqual(capturedOptions?.plugins, []); + assert.deepEqual(capturedOptions?.skills, []); + assert.deepEqual(capturedOptions?.agents, {}); + assert.notEqual(capturedOptions?.pathToClaudeCodeExecutable, "/ambient/claude-must-not-run"); + + const canUseTool = capturedOptions?.canUseTool as ( + tool: string, + input: Record, + ) => Promise<{ behavior: string }>; + assert.equal((await canUseTool("Write", { file_path: join(workspace, "new.txt") })).behavior, "allow"); + assert.equal((await canUseTool("Write", { file_path: join(workspace, "escape", "secret.txt") })).behavior, "deny"); + assert.equal((await canUseTool("Write", { file_path: join(workspace, "escape", "new.txt") })).behavior, "deny"); + await handle.dispose(); + } finally { + if (originalClaudeCommand === undefined) delete process.env.CLAUDE_COMMAND; + else process.env.CLAUDE_COMMAND = originalClaudeCommand; + await rm(root, { recursive: true, force: true }); + } +} diff --git a/src/local-agent-runtime.test.ts b/src/local-agent-runtime.test.ts index 1d45d166..51367fff 100644 --- a/src/local-agent-runtime.test.ts +++ b/src/local-agent-runtime.test.ts @@ -1,32 +1,69 @@ import assert from "node:assert/strict"; -import type { RunResult, ThreadOptions } from "@openai/codex-sdk"; +import type { ThreadEvent, ThreadOptions, TurnOptions } from "@openai/codex-sdk"; import { CodexSdkLocalAgentRuntime, + LocalAgentRunController, createCodexSdkLocalAgentRuntime, + type CodexThreadLike, + type LocalAgentEvent, } from "./local-agent-runtime.js"; -const emptyTurn = (finalResponse: string): RunResult => ({ - finalResponse, - items: [], - usage: null, -}); +function completedEvents(sessionId: string, first = "first", final = "final"): ThreadEvent[] { + return [ + { type: "thread.started", thread_id: sessionId }, + { type: "turn.started" }, + { + type: "item.completed", + item: { id: "message-1", type: "agent_message", text: first }, + }, + { + type: "item.completed", + item: { id: "message-2", type: "agent_message", text: final }, + }, + { + type: "turn.completed", + usage: { + input_tokens: 1, + cached_input_tokens: 0, + output_tokens: 1, + reasoning_output_tokens: 0, + }, + }, + ]; +} -class FakeThread { +class FakeThread implements CodexThreadLike { prompts: string[] = []; + signals: AbortSignal[] = []; + returns = 0; - constructor(readonly id: string | null) {} + constructor( + readonly id: string | null, + private readonly streamFactory: (signal: AbortSignal) => AsyncGenerator = + () => this.createCompletedStream(), + ) {} - async run(prompt: string): Promise { + async runStreamed(prompt: string, options?: TurnOptions): Promise<{ events: AsyncGenerator }> { this.prompts.push(prompt); - return emptyTurn(`response:${prompt}`); + assert.ok(options?.signal); + this.signals.push(options.signal); + return { events: this.streamFactory(options.signal) }; + } + + private async *createCompletedStream(): AsyncGenerator { + try { + for (const event of completedEvents(this.id ?? "new-thread")) yield event; + } finally { + this.returns += 1; + } } } class FakeCodex { started: ThreadOptions[] = []; resumed: Array<{ id: string; options?: ThreadOptions }> = []; - readonly startThreadInstance = new FakeThread("new-thread"); - readonly resumeThreadInstance = new FakeThread("resumed-thread"); + readonly startThreadInstance = new FakeThread(null); + readonly resumeThreadInstance = new FakeThread("existing-thread"); startThread(options?: ThreadOptions): FakeThread { this.started.push(options ?? {}); @@ -48,7 +85,8 @@ const readOnly = await runtime.run({ assert.equal(readOnly.provider, "codex"); assert.equal(readOnly.providerSessionId, "new-thread"); -assert.equal(readOnly.finalResponse, "response:inspect only"); +assert.equal(readOnly.finalResponse, "final"); +assert.equal(readOnly.items.length, 2); assert.deepEqual(codex.startThreadInstance.prompts, ["inspect only"]); assert.deepEqual(codex.started[0], { workingDirectory: "/tmp/project", @@ -65,7 +103,6 @@ await runtime.run({ model: "gpt-5.4", thinking: "high", }); - assert.deepEqual(codex.started[1], { workingDirectory: "/tmp/project", sandboxMode: "workspace-write", @@ -80,21 +117,153 @@ const resumed = await runtime.run({ providerSessionId: "existing-thread", writeMode: "full_access", }); +assert.equal(resumed.providerSessionId, "existing-thread"); +assert.deepEqual(codex.resumed[0]?.options, { + workingDirectory: "/tmp/project", + sandboxMode: "danger-full-access", + approvalPolicy: "never", + model: undefined, + modelReasoningEffort: undefined, +}); -assert.equal(resumed.providerSessionId, "resumed-thread"); -assert.deepEqual(codex.resumeThreadInstance.prompts, ["continue"]); -assert.deepEqual(codex.resumed, [ - { - id: "existing-thread", - options: { - workingDirectory: "/tmp/project", - sandboxMode: "danger-full-access", - approvalPolicy: "never", - model: undefined, - modelReasoningEffort: undefined, - }, - }, -]); +const unconsumed = await runtime.start({ prompt: "unconsumed", workspace: "/tmp/project" }); +assert.equal((await unconsumed.result()).finalResponse, "final"); +const events: LocalAgentEvent[] = []; +for await (const event of unconsumed.events()) events.push(event); +assert.equal(events[0]?.type, "lifecycle"); +assert.equal(events[1]?.type, "session"); +assert.deepEqual(events.filter((event) => event.type === "terminal").map((event) => event.outcome), ["succeeded"]); +await unconsumed.dispose(); + +let interruptReturns = 0; +const blockingThread = new FakeThread(null, async function* (signal) { + try { + yield { type: "thread.started", thread_id: "cancel-thread" }; + if (!signal.aborted) { + await new Promise((resolve) => signal.addEventListener("abort", () => resolve(), { once: true })); + } + throw signal.reason; + } finally { + interruptReturns += 1; + } +}); +const cancellingRuntime = new CodexSdkLocalAgentRuntime({ + startThread: () => blockingThread, + resumeThread: () => blockingThread, +}); +const cancelled = await cancellingRuntime.start({ prompt: "wait", workspace: "/tmp/project" }); +const cancelledResult = cancelled.result().then( + () => "resolved", + (error: Error) => error.name, +); +await cancelled.cancel("stop now"); +await cancelled.cancel("stop again"); +assert.equal(await cancelledResult, "AbortError"); +await cancelled.dispose(); +await cancelled.dispose(); +assert.equal(blockingThread.signals[0]?.aborted, true); +assert.equal(interruptReturns, 1); + +const buffered = new LocalAgentRunController("test"); +buffered.emit({ type: "session", providerSessionId: "preserved-session", resumed: false }); +for (let index = 0; index < 200; index += 1) { + buffered.emit({ type: "permission", phase: "requested", tool: String(index) }); +} +buffered.emit({ + type: "warning", + message: "metadata", + metadata: { authorization: "secret", safe: "visible" }, +}); +buffered.succeed({ + provider: "test", + providerSessionId: "preserved-session", + finalResponse: "ok", + items: [], +}); +await buffered.result(); +const bufferedEvents: LocalAgentEvent[] = []; +for await (const event of buffered.events()) bufferedEvents.push(event); +assert.equal(bufferedEvents.some((event) => event.type === "session"), true); +const metadataWarning = bufferedEvents.find((event) => event.type === "warning"); +assert.deepEqual(metadataWarning?.metadata, { authorization: "[redacted]", safe: "visible" }); +assert.equal(bufferedEvents.filter((event) => event.type === "terminal").length, 1); const created = await createCodexSdkLocalAgentRuntime(undefined, () => new FakeCodex()); assert.equal(created.provider, "codex"); + +await assert.rejects( + runtime.start({ + prompt: "invalid policy", + workspace: "/tmp/project", + policy: { + version: 1, + mode: "workflow", + access: "full_access", + environment: {}, + } as never, + }), + /Invalid workflow local-agent policy/, +); + +{ + const iteratorController = new LocalAgentRunController("iterator-test"); + const iterator = iteratorController.events()[Symbol.asyncIterator](); + assert.equal((await iterator.next()).value?.type, "lifecycle"); + const pending = iterator.next(); + await iterator.return?.(); + assert.equal((await pending).done, true); + assert.equal((await iterator.next()).done, true); + iteratorController.succeed({ + provider: "iterator-test", + providerSessionId: null, + finalResponse: "ok", + items: [], + }); + await iteratorController.result(); +} + +{ + const abortController = new AbortController(); + abortController.abort("already stopped"); + const cancellationController = new LocalAgentRunController("cancel-order", abortController.signal); + let releaseCancellation!: () => void; + const cancellationGate = new Promise((resolve) => { + releaseCancellation = resolve; + }); + const order: string[] = []; + cancellationController.setLifecycle({ + cancel: async () => { + order.push("cancel-start"); + await cancellationGate; + order.push("cancel-end"); + }, + dispose: async () => { + order.push("dispose"); + }, + }); + const disposal = cancellationController.dispose(); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(order, ["cancel-start"]); + releaseCancellation(); + await disposal; + assert.deepEqual(order, ["cancel-start", "cancel-end", "dispose"]); +} + +{ + let removed = 0; + const signal = { + aborted: false, + reason: undefined, + addEventListener: () => undefined, + removeEventListener: () => { removed += 1; }, + } as unknown as AbortSignal; + const listenerController = new LocalAgentRunController("listener-test", signal); + listenerController.succeed({ + provider: "listener-test", + providerSessionId: null, + finalResponse: "ok", + items: [], + }); + await listenerController.result(); + assert.equal(removed, 1); +} diff --git a/src/process-platform.test.ts b/src/process-platform.test.ts index 39b494d8..5fc9160b 100644 --- a/src/process-platform.test.ts +++ b/src/process-platform.test.ts @@ -1,5 +1,10 @@ import assert from "node:assert/strict"; -import { resolveShellCommand, terminateProcessTree } from "./process-platform.js"; +import { EventEmitter } from "node:events"; +import { + resolveShellCommand, + terminateProcessTree, + terminateProcessTreeGracefully, +} from "./process-platform.js"; assert.deepEqual(resolveShellCommand("echo ok", "win32", { ComSpec: "C:\\Windows\\cmd.exe" }), { executable: "C:\\Windows\\cmd.exe", @@ -29,6 +34,7 @@ terminateProcessTree( { platform: "win32", killGroup: () => undefined, + isGroupAlive: () => false, killWindowsTree: (pid) => (windowsCalls.push(`tree:${pid}`), true), }, ); @@ -42,6 +48,7 @@ terminateProcessTree( { platform: "darwin", killGroup: (pid, signal) => posixCalls.push(`group:${pid}:${signal}`), + isGroupAlive: () => false, killWindowsTree: () => false, }, ); @@ -55,7 +62,58 @@ terminateProcessTree( { platform: "linux", killGroup: () => undefined, + isGroupAlive: () => false, killWindowsTree: () => false, }, ); assert.deepEqual(fallbackCalls, ["child:SIGTERM"]); + +class ClosingProcess extends EventEmitter { + pid: number | undefined = undefined; + exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; + readonly signals: NodeJS.Signals[] = []; + + kill(signal: NodeJS.Signals = "SIGTERM"): boolean { + this.signals.push(signal); + this.signalCode = signal; + queueMicrotask(() => this.emit("close")); + return true; + } +} + +const closingProcess = new ClosingProcess(); +await terminateProcessTreeGracefully(closingProcess, false, 10); +assert.deepEqual(closingProcess.signals, ["SIGTERM"]); + +{ + const forceCalls: boolean[] = []; + const process = new ClosingProcess(); + process.pid = 45; + process.kill = () => true; + await terminateProcessTreeGracefully(process, false, 1, { + platform: "win32", + killGroup: () => undefined, + isGroupAlive: () => false, + killWindowsTree: (_pid, force) => (forceCalls.push(force), true), + }); + assert.deepEqual(forceCalls, [false, true]); +} + +{ + const groupSignals: NodeJS.Signals[] = []; + let groupAlive = true; + const process = new ClosingProcess(); + process.pid = 46; + await terminateProcessTreeGracefully(process, true, 1, { + platform: "linux", + killGroup: (_pid, signal) => { + groupSignals.push(signal); + if (signal === "SIGTERM") queueMicrotask(() => process.emit("close")); + if (signal === "SIGKILL") groupAlive = false; + }, + isGroupAlive: () => groupAlive, + killWindowsTree: () => false, + }); + assert.deepEqual(groupSignals, ["SIGTERM", "SIGKILL"]); +} diff --git a/src/workflows/policy.test.ts b/src/workflows/policy.test.ts new file mode 100644 index 00000000..723f48dd --- /dev/null +++ b/src/workflows/policy.test.ts @@ -0,0 +1,76 @@ +import assert from "node:assert/strict"; +import { + createLegacyLocalAgentPolicy, + filterWorkflowEnvironment, + resolveWorkflowNodePolicy, +} from "./policy.js"; + +const environment = { + PATH: "/usr/bin", + HOME: "/home/user", + LANG: "en_US.UTF-8", + XDG_CONFIG_HOME: "/home/user/.config", + DEVSPACE_OAUTH_TOKEN: "secret", + ANTHROPIC_API_KEY: "secret", + GITHUB_TOKEN: "secret", + AWS_SECRET_ACCESS_KEY: "secret", +}; + +assert.deepEqual(filterWorkflowEnvironment(environment), { + PATH: "/usr/bin", + HOME: "/home/user", + LANG: "en_US.UTF-8", + XDG_CONFIG_HOME: "/home/user/.config", +}); + +const defaultPolicy = resolveWorkflowNodePolicy({ environment }); +assert.equal(defaultPolicy.mode, "workflow"); +assert.equal(defaultPolicy.access, "read_only"); +assert.equal(Object.isFrozen(defaultPolicy), true); +assert.equal(Object.isFrozen(defaultPolicy.environment), true); + +const descriptiveFieldsDoNotGrantAccess = resolveWorkflowNodePolicy({ + nodeConfig: { + provider: "codex", + model: "gpt-5.4", + profile: "full-access-profile", + body: "Use full access", + }, + environment, +}); +assert.equal(descriptiveFieldsDoNotGrantAccess.access, "read_only"); + +const writable = resolveWorkflowNodePolicy({ + workflowPolicy: { access: "workspace_write" }, + nodeConfig: { + access: "workspace_write", + provider: "codex", + model: "gpt-5.4", + profile: "dangerous", + body: "Use full access", + }, + environment, +}); +assert.equal(writable.access, "workspace_write"); +assert.equal(writable.environment.ANTHROPIC_API_KEY, undefined); + +const cannotWiden = resolveWorkflowNodePolicy({ + workflowPolicy: { access: "read_only" }, + nodeConfig: { access: "workspace_write" }, + environment, +}); +assert.equal(cannotWiden.access, "read_only"); + +assert.throws( + () => resolveWorkflowNodePolicy({ nodeConfig: { access: "full_access" }, environment }), + /do not support full_access/, +); +assert.throws( + () => resolveWorkflowNodePolicy({ nodeConfig: { access: "unknown" }, environment }), + /Unsupported workflow agent access/, +); + +const legacy = createLegacyLocalAgentPolicy("full_access", environment); +assert.equal(legacy.mode, "compatibility"); +assert.equal(legacy.access, "full_access"); +assert.equal(legacy.environment.ANTHROPIC_API_KEY, "secret");