diff --git a/skills/subagents/SKILL.md b/skills/subagents/SKILL.md index 5ef4f5bb7..e486bd60b 100644 --- a/skills/subagents/SKILL.md +++ b/skills/subagents/SKILL.md @@ -60,6 +60,14 @@ devspace agents continue "" devspace agents wait ``` +Stop an active turn when its work is no longer needed: + +```bash +devspace agents stop +``` + +Stopping is scoped to the current project and preserves the durable agent record as `stopped`. + ## Good uses - Review a change for correctness, security, or missing tests. diff --git a/src/cli.ts b/src/cli.ts index e02d4f5b1..47c398264 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -497,6 +497,7 @@ function printHelp(): void { " devspace agents continue [--model ] [--effort ] ", " devspace agents show [--json]", " devspace agents wait ... [--timeout ] [--json]", + " devspace agents stop [--json]", " devspace agents daemon ", " devspace -v, --version Print the installed version", "", @@ -544,6 +545,9 @@ async function runAgentsCommand(args: string[]): Promise { case "wait": await runAgentWorkflowCommand(json, () => runAgentsWait(commandArgs, json)); return; + case "stop": + await runAgentWorkflowCommand(json, () => runAgentsStop(commandArgs, json)); + return; case "targets": await runAgentWorkflowCommand(json, () => runAgentsTargets(commandArgs, json)); return; @@ -665,6 +669,19 @@ async function runAgentsWait(args: string[], json: boolean): Promise { printAgentXml(results.map(formatAgentObservation).join("\n")); } +async function runAgentsStop(args: string[], json: boolean): Promise { + const [id, ...extra] = args; + if (!id || extra.length > 0) throw new Error("Usage: devspace agents stop [--json]"); + const config = loadConfig(); + const client = createLocalAgentClient(config); + const scope = resolveCliWorkspaceContext(config.allowedRoots); + const record = presentAgentWorkflowResult(await client.stopAgent(id, scope), json); + if (!record) return; + const observation = presentAgentObservation(record); + if (json) printJson(observation); + else printAgentXml(formatAgentObservation(observation)); +} + function parseAgentsWaitArgs(args: string[]): { ids: string[]; timeoutMs?: number } { const ids: string[] = []; let timeoutMs: number | undefined; @@ -808,6 +825,7 @@ function printAgentsHelp(): void { " devspace agents continue [--model ] [--effort ] [--json] ", " devspace agents show [--json]", " devspace agents wait ... [--timeout ] [--json]", + " devspace agents stop [--json]", " devspace agents targets [--json]", " devspace agents daemon [--json]", ].join("\n"), diff --git a/src/local-agent-acp.test.ts b/src/local-agent-acp.test.ts index bb9998d15..2d6fa3fe5 100644 --- a/src/local-agent-acp.test.ts +++ b/src/local-agent-acp.test.ts @@ -16,6 +16,7 @@ const requests: Array<{ method: string; params?: unknown }> = []; const queues = new Map(); const connection = { agent: { + async notify(_method: string, _params?: unknown): Promise {}, async request(method: string, params?: unknown): Promise { requests.push({ method, params }); const input = params as { sessionId?: string } | undefined; @@ -203,6 +204,7 @@ const overlappingPrompt = new Promise((resolvePrompt) => { releaseOverlapp const promptEntered = new Promise((resolveEntered) => { markPromptEntered = resolveEntered; }); const overlapConnection = { agent: { + async notify(_method: string, _params?: unknown): Promise {}, async request(method: string, params?: unknown): Promise { const input = params as { sessionId?: string } | undefined; if (method === "session/new") { @@ -356,6 +358,7 @@ const grokQueues = new Map(); const grokCompletionRegistry = new GrokPromptCompletionRegistry(); const grokConnection = { agent: { + async notify(_method: string, _params?: unknown): Promise {}, async request(method: string, params?: unknown): Promise { grokRequests.push({ method, params }); const input = params as { sessionId?: string; _meta?: { promptId?: string } } | undefined; @@ -421,6 +424,7 @@ await grokRuntime.close(); const grokConfigurationConnection = { agent: { + async notify(_method: string, _params?: unknown): Promise {}, async request(method: string): Promise { if (method === "session/new") { return { diff --git a/src/local-agent-acp.ts b/src/local-agent-acp.ts index 88ab8c0ec..3736a6c58 100644 --- a/src/local-agent-acp.ts +++ b/src/local-agent-acp.ts @@ -8,6 +8,7 @@ import { captureAgentProviderResult, isProgrammerDefect, } from "./local-agent-errors.js"; +import { bindLocalAgentAbort, localAgentCancelledError } from "./local-agent-cancellation.js"; import { terminateProcessTree } from "./process-platform.js"; import { DEVSPACE_VERSION } from "./version.js"; import { @@ -49,6 +50,7 @@ const ACP_COMMANDS: Record = { interface AcpConnectionLike { agent: { request(method: string, params?: unknown): Promise; + notify(method: string, params?: unknown): Promise; }; close(error?: unknown): void; closed: Promise; @@ -142,6 +144,9 @@ export class AcpRuntime implements LocalAgentRuntime { throw new TypeError(`${this.provider} ACP session ${sessionId} already has an active turn.`); } this.activeSessions.add(sessionId); + const removeAbort = bindLocalAgentAbort(input.signal, () => ( + this.connection.agent.notify("session/cancel", { sessionId }) + )); const queue = this.queues.get(sessionId) ?? { values: [] }; this.queues.set(sessionId, queue); const promptId = this.provider === "grok" ? this.nextPromptId() : undefined; @@ -166,9 +171,16 @@ export class AcpRuntime implements LocalAgentRuntime { prompt: [{ type: "text", text: input.prompt }], ...(promptId ? { _meta: { promptId, requestId: promptId } } : {}), }); - const response = completion - ? await Promise.race([standardResponse, completion]) - : await standardResponse; + let response: unknown; + try { + response = completion + ? await Promise.race([standardResponse, completion]) + : await standardResponse; + } catch (cause) { + if (input.signal?.aborted) throw localAgentCancelledError(this.provider, "run", cause); + throw cause; + } + if (input.signal?.aborted) throw localAgentCancelledError(this.provider, "run"); if (completion && isGrokPromptCompletion(response)) { await yieldToAcpQueue(); } else if (promptId) { @@ -193,6 +205,7 @@ export class AcpRuntime implements LocalAgentRuntime { items: updates, }; } finally { + removeAbort(); if (promptId) this.grokCompletionRegistry?.remove(sessionId, promptId); this.activeSessions.delete(sessionId); } @@ -416,6 +429,13 @@ export class AcpLocalAgentDriver implements LocalAgentDriver { authority: "write_mode", idleTimeoutMs: 5 * 60_000, } as const; + readonly capabilities = { + sessions: { resume: true, close: true }, + turns: { interrupt: true }, + configuration: { modelOverride: true, effortOverride: true }, + permissions: { enforcement: "native" }, + mcp: { supported: true }, + } as const; private commandResolved = false; private resolvedCommand?: string; diff --git a/src/local-agent-cancellation.ts b/src/local-agent-cancellation.ts new file mode 100644 index 000000000..e8d47a9d7 --- /dev/null +++ b/src/local-agent-cancellation.ts @@ -0,0 +1,28 @@ +import { AgentProviderCancelledError } from "./local-agent-errors.js"; +import type { LocalAgentDriverKind } from "./local-agent-provider.js"; + +export function localAgentCancelledError( + provider: LocalAgentDriverKind, + operation: string, + cause?: unknown, +): AgentProviderCancelledError { + return new AgentProviderCancelledError({ + code: "PROVIDER_CANCELLED", + provider, + operation, + retryable: false, + ...(cause === undefined ? {} : { cause }), + message: `${provider} agent turn was stopped.`, + }); +} + +export function bindLocalAgentAbort( + signal: AbortSignal | undefined, + interrupt: () => void | Promise, +): () => void { + if (!signal) return () => undefined; + const onAbort = () => { void Promise.resolve(interrupt()).catch(() => undefined); }; + if (signal.aborted) onAbort(); + else signal.addEventListener("abort", onAbort, { once: true }); + return () => signal.removeEventListener("abort", onAbort); +} diff --git a/src/local-agent-claude.test.ts b/src/local-agent-claude.test.ts index bd19fbc6d..2a470b8c4 100644 --- a/src/local-agent-claude.test.ts +++ b/src/local-agent-claude.test.ts @@ -41,6 +41,8 @@ class FakeClaudeQuery implements ClaudeQueryLike, AsyncIterator { this.closeCount += 1; } + async interrupt(): Promise {} + async setPermissionMode(mode: string): Promise { this.permissionModes.push(mode); } @@ -207,6 +209,7 @@ const brokenStreamQuery: ClaudeQueryLike = { }; }, close() {}, + async interrupt() {}, async setPermissionMode() {}, async applyFlagSettings() {}, }; diff --git a/src/local-agent-claude.ts b/src/local-agent-claude.ts index 65a85df1d..d7e46ad90 100644 --- a/src/local-agent-claude.ts +++ b/src/local-agent-claude.ts @@ -5,6 +5,7 @@ import { captureAgentProviderResult, isProgrammerDefect, } from "./local-agent-errors.js"; +import { bindLocalAgentAbort, localAgentCancelledError } from "./local-agent-cancellation.js"; import type { LocalAgentDriverKind } from "./local-agent-provider.js"; import type { LocalAgentDriver, @@ -28,6 +29,7 @@ const CLAUDE_WORKSPACE_ALLOWED_TOOLS = [ export interface ClaudeQueryLike extends AsyncIterable { close(): void; + interrupt(): Promise; setPermissionMode(mode: ClaudePermissionMode): Promise; applyFlagSettings(settings: Record): Promise; setModel?(model?: string): Promise; @@ -106,6 +108,8 @@ export class ClaudeQueryRuntime implements LocalAgentRuntime { message: "Claude runtime is not running.", }); } + const removeAbort = bindLocalAgentAbort(input.signal, () => this.query.interrupt()); + try { if (this.providerSessionId) await callbacks?.onSessionId?.(this.providerSessionId); const flagSettings = claudeAuthoritySettings(input.workspaceRoot, input.writeMode); if (input.effort) { @@ -129,6 +133,7 @@ export class ClaudeQueryRuntime implements LocalAgentRuntime { try { next = await this.iterator.next(); } catch (error) { + if (input.signal?.aborted) throw localAgentCancelledError("claude", "run", error); this.alive = false; if (isProgrammerDefect(error)) throw error; throw new AgentProviderUnavailableError({ @@ -161,6 +166,7 @@ export class ClaudeQueryRuntime implements LocalAgentRuntime { } } if (record?.type !== "result") continue; + if (input.signal?.aborted) throw localAgentCancelledError("claude", "run"); const resultError = claudeResultError(record); if (resultError) { @@ -190,6 +196,9 @@ export class ClaudeQueryRuntime implements LocalAgentRuntime { items, }; } + } finally { + removeAbort(); + } }, }); } @@ -219,6 +228,13 @@ export class ClaudeLocalAgentDriver implements LocalAgentDriver { authority: "full_access_boundary", idleTimeoutMs: 3 * 60_000, } as const; + readonly capabilities = { + sessions: { resume: true, close: false }, + turns: { interrupt: true }, + configuration: { modelOverride: true, effortOverride: true }, + permissions: { enforcement: "native" }, + mcp: { supported: true }, + } as const; constructor( private readonly factory: ClaudeQueryFactory = defaultClaudeQueryFactory, diff --git a/src/local-agent-client.ts b/src/local-agent-client.ts index 5aa5bce35..74d735d6b 100644 --- a/src/local-agent-client.ts +++ b/src/local-agent-client.ts @@ -49,6 +49,7 @@ import type { AgentListError, AgentLookupError, AgentStartError, + AgentStopError, AgentWaitError, LocalAgentWaitResult, RunOverrides, @@ -160,6 +161,14 @@ export class LocalAgentClient { return decodeRequestResult(result, "agent.wait", decodeAgentWaitResults); } + async stopAgent( + agentId: string, + scope: LocalAgentWorkspaceScope, + ): Promise> { + const result = await this.request("agent.stop", { id: agentId, scope }); + return decodeRequestResult(result, "agent.stop", decodeAgentRecord); + } + async status(): Promise> { const result = await this.requestExisting("daemon.status", {}); return decodeRequestResult(result, "daemon.status", decodeDaemonStatus); @@ -695,6 +704,11 @@ function isRequestError( case "agent.get": case "agent.wait": return category === "target" || category === "scope" || category === "store"; + case "agent.stop": + return category === "target" + || category === "scope" + || category === "conflict" + || category === "store"; case "agent.list": return category === "scope" || category === "store"; case "hello": diff --git a/src/local-agent-codex.ts b/src/local-agent-codex.ts index e619a1635..134385bca 100644 --- a/src/local-agent-codex.ts +++ b/src/local-agent-codex.ts @@ -8,6 +8,7 @@ import { AgentProviderUnavailableError, captureAgentProviderResult, } from "./local-agent-errors.js"; +import { bindLocalAgentAbort, localAgentCancelledError } from "./local-agent-cancellation.js"; import { removeDevspaceNodeModulesBinFromPath } from "./local-agent-path.js"; import { terminateProcessTree } from "./process-platform.js"; import { DEVSPACE_VERSION } from "./version.js"; @@ -147,7 +148,14 @@ export class CodexAppServerRuntime implements LocalAgentRuntime { } await callbacks?.onSessionId?.(threadId); - const completed = await this.rpc.runTurn(threadId, turnParams(input, threadId)); + let completed: CodexTurnResult; + try { + completed = await this.rpc.runTurn(threadId, turnParams(input, threadId), input.signal); + } catch (cause) { + if (input.signal?.aborted) throw localAgentCancelledError(this.provider, "run", cause); + throw cause; + } + if (input.signal?.aborted) throw localAgentCancelledError(this.provider, "run"); const parsed = parseCompletedTurn(completed.event.params, completed.items); if (parsed.failure) { throw new AgentProviderExecutionError({ @@ -232,6 +240,13 @@ export class CodexLocalAgentDriver implements LocalAgentDriver { readonly provider = "codex" as const; readonly providerInstanceId = "codex"; readonly runtimePolicy = { scope: "instance", idleTimeoutMs: 5 * 60_000 } as const; + readonly capabilities = { + sessions: { resume: true, close: false }, + turns: { interrupt: true }, + configuration: { modelOverride: true, effortOverride: true }, + permissions: { enforcement: "native" }, + mcp: { supported: true }, + } as const; private commandResolved = false; private resolvedCommand?: ResolvedCodexCommand; @@ -354,7 +369,7 @@ class CodexAppServerRpc { this.write({ method, ...(params === undefined ? {} : { params }) }); } - async runTurn(threadId: string, params: unknown): Promise { + async runTurn(threadId: string, params: unknown, signal?: AbortSignal): Promise { if (this.fatalError) throw this.fatalError; if (this.turns.has(threadId)) throw new Error(`Codex thread ${threadId} already has an active turn.`); let resolveTurn!: (result: CodexTurnResult) => void; @@ -370,12 +385,20 @@ class CodexAppServerRpc { reject: rejectTurn, }; this.turns.set(threadId, turn); + const removeAbort = bindLocalAgentAbort(signal, async () => { + if (!turn.turnId) return; + await this.request("turn/interrupt", { threadId, turnId: turn.turnId }); + }); try { const response = await this.request("turn/start", params); turn.turnId = readString(asRecord(response)?.turn, "id"); + if (signal?.aborted && turn.turnId) { + await this.request("turn/interrupt", { threadId, turnId: turn.turnId }); + } if (turn.completed) return { event: turn.completed, items: turn.items }; return await completion; } finally { + removeAbort(); if (this.turns.get(threadId) === turn) this.turns.delete(threadId); } } diff --git a/src/local-agent-daemon-lifecycle.ts b/src/local-agent-daemon-lifecycle.ts index 82573c779..fd0405e50 100644 --- a/src/local-agent-daemon-lifecycle.ts +++ b/src/local-agent-daemon-lifecycle.ts @@ -12,7 +12,7 @@ import { } from "node:fs"; import { join, resolve } from "node:path"; -export const LOCAL_AGENT_DAEMON_PROTOCOL_VERSION = 6; +export const LOCAL_AGENT_DAEMON_PROTOCOL_VERSION = 7; export const LOCAL_AGENT_DAEMON_SOCKET_NAME = "agentd.sock"; export const LOCAL_AGENT_DAEMON_PID_NAME = "agentd.pid"; export const LOCAL_AGENT_DAEMON_LOCK_NAME = "agentd.lock"; diff --git a/src/local-agent-daemon-protocol.test.ts b/src/local-agent-daemon-protocol.test.ts index 7b7a92488..50641ed9b 100644 --- a/src/local-agent-daemon-protocol.test.ts +++ b/src/local-agent-daemon-protocol.test.ts @@ -194,6 +194,20 @@ const waitRequest = decodeLocalAgentDaemonRequest({ }); assert.equal(waitRequest.method, "agent.wait"); if (waitRequest.method !== "agent.wait") throw new Error("expected agent.wait request"); + +const stopRequest = decodeLocalAgentDaemonRequest({ + requestId: "stop", + protocolVersion: LOCAL_AGENT_DAEMON_PROTOCOL_VERSION, + authToken: "secret", + method: "agent.stop", + params: { + id: "agt_stop", + scope: { workspaceId: "ws_test", workspaceRoot: "/tmp/project" }, + }, +}); +assert.equal(stopRequest.method, "agent.stop"); +if (stopRequest.method !== "agent.stop") throw new Error("expected agent.stop request"); +assert.equal(stopRequest.params.id, "agt_stop"); assert.deepEqual(waitRequest.params.ids, ["agt_one", "agt_two"]); assert.equal(waitRequest.params.timeoutMs, 5_000); diff --git a/src/local-agent-daemon-protocol.ts b/src/local-agent-daemon-protocol.ts index a9f2b2fc1..37f72b613 100644 --- a/src/local-agent-daemon-protocol.ts +++ b/src/local-agent-daemon-protocol.ts @@ -19,6 +19,7 @@ export type LocalAgentDaemonMethod = | "agent.get" | "agent.list" | "agent.wait" + | "agent.stop" | "daemon.status" | "daemon.stop" | "daemon.logs"; @@ -34,6 +35,7 @@ export type LocalAgentDaemonRequest = scope: LocalAgentWorkspaceScope; timeoutMs?: number; }> + | AgentDaemonRequestBase<"agent.stop", { id: string; scope: LocalAgentWorkspaceScope }> | AgentDaemonRequestBase<"daemon.status", Record> | AgentDaemonRequestBase<"daemon.stop", { ifIdle?: boolean }> | AgentDaemonRequestBase<"daemon.logs", { lines?: number }>; @@ -169,6 +171,17 @@ export function decodeLocalAgentDaemonRequest(value: unknown): LocalAgentDaemonR method, params: decodeWaitParams(params), } as LocalAgentDaemonRequest; + case "agent.stop": + return { + requestId, + protocolVersion, + authToken, + method, + params: { + id: requiredString(asRecord(params)?.id, "id"), + scope: decodeWorkspaceScope(asRecord(params)?.scope), + }, + } as LocalAgentDaemonRequest; case "daemon.logs": return { requestId, diff --git a/src/local-agent-daemon.test.ts b/src/local-agent-daemon.test.ts index 8985bdc70..5cf7d4dcd 100644 --- a/src/local-agent-daemon.test.ts +++ b/src/local-agent-daemon.test.ts @@ -96,6 +96,11 @@ class FakeManager implements LocalAgentDaemonManager { return Result.ok(agentIds.map((id) => ({ id, status: "running" as const }))); } + async stop(_agentId: string, _scope: unknown) { + this.activeTurnCount = 0; + return Result.ok({ ...record, status: "stopped" } as LocalAgentRecord); + } + async evictIdle(): Promise {} async close(): Promise { @@ -174,6 +179,7 @@ try { assert.deepEqual(unwrap(await client.wait([record.id], recordScope, 0)), [ { id: record.id, status: "running" }, ]); + assert.equal(unwrap(await client.stopAgent(record.id, recordScope)).status, "stopped"); assert.equal(unwrap(await client.status()).state, "ready"); unwrap(await client.stop()); diff --git a/src/local-agent-daemon.ts b/src/local-agent-daemon.ts index aac96e9d9..3f929e25a 100644 --- a/src/local-agent-daemon.ts +++ b/src/local-agent-daemon.ts @@ -37,6 +37,7 @@ import type { AgentListError, AgentLookupError, AgentStartError, + AgentStopError, AgentWaitError, LocalAgentWaitResult, RunOverrides, @@ -61,6 +62,7 @@ export interface LocalAgentDaemonManager { timeoutMs?: number, signal?: AbortSignal, ): Promise>; + stop(agentId: string, scope: LocalAgentWorkspaceScope): Promise>; evictIdle(now?: number): Promise; close(): Promise; readonly activeTurnCount: number; @@ -339,6 +341,11 @@ export class LocalAgentDaemon { request.params.timeoutMs, signal, )); + case "agent.stop": + return unwrapManagerResult(await this.manager.stop( + request.params.id, + request.params.scope, + )); case "daemon.status": return this.status(); case "daemon.stop": diff --git a/src/local-agent-manager.test.ts b/src/local-agent-manager.test.ts index 90e15214f..5d2ea5dd7 100644 --- a/src/local-agent-manager.test.ts +++ b/src/local-agent-manager.test.ts @@ -5,6 +5,7 @@ import { join } from "node:path"; import { tmpdir } from "node:os"; import { LocalAgentManager } from "./local-agent-manager.js"; import { + AgentProviderCancelledError, AgentProviderExecutionError, type AgentProviderError, } from "./local-agent-errors.js"; @@ -16,6 +17,14 @@ import type { LocalAgentRuntime, LocalAgentRuntimeContext, } from "./local-agent-runtime.js"; + +const TEST_CAPABILITIES = { + sessions: { resume: true, close: false }, + turns: { interrupt: true }, + configuration: { modelOverride: true, effortOverride: true }, + permissions: { enforcement: "native" }, + mcp: { supported: false }, +} as const; import { LocalAgentRuntimePool } from "./local-agent-runtime-pool.js"; import { LocalAgentStore } from "./local-agent-store.js"; import type { SubagentsConfig } from "./local-agent-config.js"; @@ -68,6 +77,20 @@ class FakeRuntime implements LocalAgentRuntime { if (input.prompt.includes("hold")) { await new Promise((resolve) => { this.releaseHold = resolve; }); } + if (input.prompt.includes("cancel-me")) { + await new Promise((resolve) => { + const done = () => resolve(); + if (input.signal?.aborted) done(); + else input.signal?.addEventListener("abort", done, { once: true }); + }); + return Result.err(new AgentProviderCancelledError({ + code: "PROVIDER_CANCELLED", + provider: "codex", + operation: "run", + retryable: false, + message: "codex agent turn was stopped.", + })); + } return Result.ok({ provider: this.provider, providerSessionId: "thread_test", @@ -100,6 +123,7 @@ const driver: LocalAgentDriver = { providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "agent" }, + capabilities: TEST_CAPABILITIES, createRuntime: async (context) => { const runtime = new FakeRuntime(); runtimes.set(context.agentId, runtime); @@ -178,6 +202,18 @@ const unknown = await manager.start({ assert.equal(unknown.isErr(), true); if (unknown.isErr()) assert.equal(unknown.error.code, "UNKNOWN_TARGET"); +const cancellable = unwrap(await manager.start({ + target: "codex", + prompt: "cancel-me", + workspaceId: scope.workspaceId, + workspaceRoot: root, +})); +await waitFor(() => getRecord(cancellable.id).status === "running"); +const stopped = unwrap(await manager.stop(cancellable.id, scope)); +assert.equal(stopped.status, "stopped"); +assert.equal(store.getLatestTurn(cancellable.id)?.status, "stopped"); +assert.equal(store.getLatestTurn(cancellable.id)?.errorCode, "PROVIDER_CANCELLED"); + const namedInstance = unwrap(await manager.start({ target: "codex-work", prompt: "inspect named instance", @@ -293,7 +329,7 @@ const second = unwrap(await manager.start({ })); await waitFor(() => getRecord(second.id).status === "idle"); assert.notEqual(first.id, second.id); -assert.equal(runtimes.size, 3, "different agents receive independent logical runtimes"); +assert.equal(runtimes.size, 4, "different agents receive independent logical runtimes"); const failed = unwrap(await manager.start({ target: "reviewer", diff --git a/src/local-agent-manager.ts b/src/local-agent-manager.ts index f5a17986e..b6dd2fdd8 100644 --- a/src/local-agent-manager.ts +++ b/src/local-agent-manager.ts @@ -2,6 +2,7 @@ import { resolve } from "node:path"; import { Result, type Result as BetterResult } from "better-result"; import { AgentConflictError, + AgentProviderCancelledError, AgentScopeError, AgentStoreError, AgentTargetError, @@ -73,6 +74,7 @@ export type AgentContinueError = AgentStartError; export type AgentLookupError = AgentTargetError | AgentScopeError | AgentStoreError; export type AgentListError = AgentScopeError | AgentStoreError; export type AgentWaitError = AgentLookupError; +export type AgentStopError = AgentLookupError | AgentConflictError; export type LocalAgentWaitResult = | { id: string; status: "running"; wait?: "timeout" } @@ -83,6 +85,7 @@ export type LocalAgentWaitResult = interface ActiveLocalAgentTurn { turnId: number; completion: Promise; + abortController: AbortController; } /** @@ -260,6 +263,38 @@ export class LocalAgentManager { return Result.ok(results); } + async stop( + agentId: string, + scope: LocalAgentWorkspaceScope, + ): Promise> { + const record = this.get(agentId, scope); + if (record.isErr()) return record; + if (record.value.status === "stopped") return Result.ok(record.value); + const active = this.activeTurns.get(agentId); + if (!active) { + return Result.err(new AgentConflictError({ + code: "AGENT_CONFLICT", + agentId, + operation: "stop", + retryable: false, + message: `Agent ${agentId} does not have a running turn.`, + })); + } + const driver = this.drivers.get(record.value.providerInstanceId); + if (!driver?.capabilities.turns.interrupt) { + return Result.err(new AgentConflictError({ + code: "AGENT_CONFLICT", + agentId, + operation: "stop", + retryable: false, + message: `Agent ${agentId} provider does not support stopping an active turn.`, + })); + } + active.abortController.abort(); + await active.completion; + return this.get(agentId, scope); + } + async close(): Promise { if (this.closePromise) return this.closePromise; this.accepting = false; @@ -315,10 +350,22 @@ export class LocalAgentManager { if (begun.isErr()) return begun; // Defer invocation until after the tracking entry is visible. This keeps // cleanup correct even if runTurn later gains a synchronous completion path. + const abortController = new AbortController(); const turn = Promise.resolve().then(() => ( - this.runTurn(begun.value.agent, begun.value.turn.id, prompt, overrides, workspaceId) + this.runTurn( + begun.value.agent, + begun.value.turn.id, + prompt, + overrides, + workspaceId, + abortController.signal, + ) )); - this.activeTurns.set(record.id, { turnId: begun.value.turn.id, completion: turn }); + this.activeTurns.set(record.id, { + turnId: begun.value.turn.id, + completion: turn, + abortController, + }); void turn.catch(() => undefined); return Result.ok(begun.value.agent); } @@ -329,6 +376,7 @@ export class LocalAgentManager { prompt: string, overrides: RunOverrides, workspaceId?: string, + signal?: AbortSignal, ): Promise { const startedAt = Date.now(); this.log("info", "agent_run_started", { @@ -387,7 +435,7 @@ export class LocalAgentManager { if (updated.isErr()) throw updated.error; }, }; - const result = await this.pool.run(driver.value, context, input.value, callbacks); + const result = await this.pool.run(driver.value, context, { ...input.value, signal }, callbacks); if (result.isErr()) { this.persistRunError(record, turnId, result.error, startedAt); return; @@ -442,13 +490,14 @@ export class LocalAgentManager { error: LocalAgentError, startedAt: number, ): void { + const stopped = AgentProviderCancelledError.is(error); const persisted = this.store.finishTurnResult(record.id, turnId, { - status: "failed", + status: stopped ? "stopped" : "failed", error: error.message, errorCode: error.code, errorRetryable: error.retryable, }); - this.log("error", "agent_run_failed", { + this.log(stopped ? "info" : "error", stopped ? "agent_run_stopped" : "agent_run_failed", { providerInstanceId: record.providerInstanceId, driver: record.driver, agentId: record.id, diff --git a/src/local-agent-opencode-v2.ts b/src/local-agent-opencode-v2.ts index 5c0fc1ff3..604a44191 100644 --- a/src/local-agent-opencode-v2.ts +++ b/src/local-agent-opencode-v2.ts @@ -12,6 +12,7 @@ import { AgentProviderUnavailableError, captureAgentProviderResult, } from "./local-agent-errors.js"; +import { bindLocalAgentAbort, localAgentCancelledError } from "./local-agent-cancellation.js"; import type { LocalAgentRunCallbacks, LocalAgentRunInput, @@ -151,6 +152,10 @@ export class OpencodeV2Runtime implements LocalAgentRuntime { const controller = new AbortController(); this.promptControllers.add(controller); let timedOut = false; + const removeAbort = bindLocalAgentAbort(input.signal, async () => { + await this.client.session.interrupt({ sessionID: sessionId }); + controller.abort(); + }); const timer = setTimeout(() => { timedOut = true; controller.abort(); @@ -168,6 +173,7 @@ export class OpencodeV2Runtime implements LocalAgentRuntime { type: "assistant", }, { signal: controller.signal }); } catch (error) { + if (input.signal?.aborted) throw localAgentCancelledError("opencode", "prompt", error); if (!timedOut) throw error; throw new AgentProviderProtocolError({ code: "PROVIDER_PROTOCOL_ERROR", @@ -179,6 +185,7 @@ export class OpencodeV2Runtime implements LocalAgentRuntime { }); } finally { clearTimeout(timer); + removeAbort(); this.promptControllers.delete(controller); } } diff --git a/src/local-agent-opencode.ts b/src/local-agent-opencode.ts index cb028fd33..7461d3e78 100644 --- a/src/local-agent-opencode.ts +++ b/src/local-agent-opencode.ts @@ -8,6 +8,7 @@ import { AgentProviderUnavailableError, captureAgentProviderResult, } from "./local-agent-errors.js"; +import { bindLocalAgentAbort, localAgentCancelledError } from "./local-agent-cancellation.js"; import type { LocalAgentDriver, LocalAgentRunCallbacks, @@ -125,6 +126,13 @@ export class OpencodeRuntime implements LocalAgentRuntime { const controller = new AbortController(); this.promptControllers.add(controller); let timedOut = false; + const removeAbort = bindLocalAgentAbort(input.signal, async () => { + await this.client.session.abort({ + sessionID: sessionId, + directory: input.workspaceRoot, + }, { throwOnError: false }); + controller.abort(); + }); const timer = setTimeout(() => { timedOut = true; controller.abort(); @@ -132,6 +140,7 @@ export class OpencodeRuntime implements LocalAgentRuntime { try { return await promptOpencodeSession(this.client, sessionId, input, controller.signal); } catch (error) { + if (input.signal?.aborted) throw localAgentCancelledError("opencode", "prompt", error); if (!timedOut) throw error; throw new AgentProviderProtocolError({ code: "PROVIDER_PROTOCOL_ERROR", @@ -143,6 +152,7 @@ export class OpencodeRuntime implements LocalAgentRuntime { }); } finally { clearTimeout(timer); + removeAbort(); this.promptControllers.delete(controller); } } @@ -152,6 +162,13 @@ export class OpencodeLocalAgentDriver implements LocalAgentDriver { readonly provider = "opencode" as const; readonly providerInstanceId = "opencode"; readonly runtimePolicy = { scope: "instance", idleTimeoutMs: 5 * 60_000 } as const; + readonly capabilities = { + sessions: { resume: true, close: false }, + turns: { interrupt: true }, + configuration: { modelOverride: true, effortOverride: true }, + permissions: { enforcement: "native" }, + mcp: { supported: true }, + } as const; private readonly factory: OpencodeFactory; private readonly v2Factory: OpencodeV2Factory; private readonly env: NodeJS.ProcessEnv; diff --git a/src/local-agent-pi.test.ts b/src/local-agent-pi.test.ts index 6c1471cbb..28b945eab 100644 --- a/src/local-agent-pi.test.ts +++ b/src/local-agent-pi.test.ts @@ -28,6 +28,8 @@ class FakePiSession implements PiSessionLike { for (const listener of this.listeners) listener({ type: "agent_end" } as AgentSessionEvent); } + async abort(): Promise {} + subscribe(listener: AgentSessionEventListener): () => void { this.listeners.add(listener); return () => this.listeners.delete(listener); diff --git a/src/local-agent-pi.ts b/src/local-agent-pi.ts index 453168e7d..4d2a5fd0a 100644 --- a/src/local-agent-pi.ts +++ b/src/local-agent-pi.ts @@ -6,6 +6,7 @@ import { AgentProviderUnavailableError, captureAgentProviderResult, } from "./local-agent-errors.js"; +import { bindLocalAgentAbort, localAgentCancelledError } from "./local-agent-cancellation.js"; import type { LocalAgentDriver, LocalAgentRunCallbacks, @@ -38,6 +39,7 @@ export type PiSessionLike = Pick< | "setModel" | "setThinkingLevel" | "dispose" + | "abort" >; export type PiSessionFactory = ( @@ -78,6 +80,8 @@ export class PiSessionRuntime implements LocalAgentRuntime { message: "Pi runtime is not running.", }); } + const removeAbort = bindLocalAgentAbort(input.signal, () => this.session.abort()); + try { await callbacks?.onSessionId?.(this.session.sessionId); await this.applyOverrides(input); this.events = []; @@ -85,6 +89,10 @@ export class PiSessionRuntime implements LocalAgentRuntime { this.collectingEvents = true; try { await this.session.prompt(input.prompt); + if (input.signal?.aborted) throw localAgentCancelledError("pi", "run"); + } catch (cause) { + if (input.signal?.aborted) throw localAgentCancelledError("pi", "run", cause); + throw cause; } finally { this.collectingEvents = false; } @@ -116,6 +124,9 @@ export class PiSessionRuntime implements LocalAgentRuntime { finalResponse, items: [...this.events, ...currentMessages], }; + } finally { + removeAbort(); + } }, }); } @@ -166,6 +177,13 @@ export class PiLocalAgentDriver implements LocalAgentDriver { readonly provider = "pi" as const; readonly providerInstanceId = "pi"; readonly runtimePolicy = { scope: "agent", idleTimeoutMs: 3 * 60_000 } as const; + readonly capabilities = { + sessions: { resume: true, close: false }, + turns: { interrupt: true }, + configuration: { modelOverride: true, effortOverride: true }, + permissions: { enforcement: "client-boundary" }, + mcp: { supported: false }, + } as const; constructor( private readonly factory: PiSessionFactory = defaultPiSessionFactory, diff --git a/src/local-agent-provider-registry.ts b/src/local-agent-provider-registry.ts index de944d72f..60c70ffbb 100644 --- a/src/local-agent-provider-registry.ts +++ b/src/local-agent-provider-registry.ts @@ -51,6 +51,7 @@ export class LocalAgentProviderRegistry { class ProviderInstanceDriver implements LocalAgentDriver { readonly provider: LocalAgentDriverKind; readonly runtimePolicy: LocalAgentDriver["runtimePolicy"]; + readonly capabilities: LocalAgentDriver["capabilities"]; constructor( readonly providerInstanceId: LocalAgentProviderInstanceId, @@ -58,6 +59,7 @@ class ProviderInstanceDriver implements LocalAgentDriver { ) { this.provider = driver.provider; this.runtimePolicy = driver.runtimePolicy; + this.capabilities = driver.capabilities; } createRuntime(context: Parameters[0]) { diff --git a/src/local-agent-runtime.test.ts b/src/local-agent-runtime.test.ts index 1aaead71f..23d8393da 100644 --- a/src/local-agent-runtime.test.ts +++ b/src/local-agent-runtime.test.ts @@ -15,6 +15,14 @@ import type { LocalAgentRuntimeContext, } from "./local-agent-runtime.js"; +const TEST_CAPABILITIES = { + sessions: { resume: true, close: false }, + turns: { interrupt: true }, + configuration: { modelOverride: true, effortOverride: true }, + permissions: { enforcement: "native" }, + mcp: { supported: false }, +} as const; + const context: LocalAgentRuntimeContext = { agentId: "agt_test", providerInstanceId: "codex", @@ -96,6 +104,7 @@ const driver: LocalAgentDriver = { providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "instance", idleTimeoutMs: Number.POSITIVE_INFINITY }, + capabilities: TEST_CAPABILITIES, createRuntime: async () => { createCount += 1; await Promise.resolve(); @@ -136,6 +145,7 @@ const sessionDriver: LocalAgentDriver = { providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "instance", idleTimeoutMs: Number.POSITIVE_INFINITY }, + capabilities: TEST_CAPABILITIES, createRuntime: async () => Result.ok(sessionRuntime), }; await sessionPool.run(sessionDriver, context, input); @@ -164,6 +174,7 @@ const shutdownReleaseDriver: LocalAgentDriver = { providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "instance", idleTimeoutMs: Number.POSITIVE_INFINITY }, + capabilities: TEST_CAPABILITIES, createRuntime: async () => Result.ok(shutdownReleaseRuntime), }; await shutdownReleasePool.run(shutdownReleaseDriver, context, input); @@ -213,6 +224,7 @@ const cleanupDriver: LocalAgentDriver = { providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "instance" }, + capabilities: TEST_CAPABILITIES, createRuntime: async () => Result.ok(cleanupRuntime), }; const cleanupFailure = await cleanupPool.run(cleanupDriver, context, input); @@ -230,6 +242,7 @@ if (cleanupFailure.isErr()) assert.equal(cleanupFailure.error.message, "provider providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "instance" }, + capabilities: TEST_CAPABILITIES, createRuntime: async () => Result.ok(attempts++ === 0 ? deadRuntime : replacementRuntime), }; @@ -255,6 +268,7 @@ if (cleanupFailure.isErr()) assert.equal(cleanupFailure.error.message, "provider providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "instance" }, + capabilities: TEST_CAPABILITIES, createRuntime: async () => Result.ok(completedTurnRuntime), }; @@ -272,6 +286,7 @@ const raceDriver: LocalAgentDriver = { providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "instance" }, + capabilities: TEST_CAPABILITIES, createRuntime: () => creating, }; const pendingRun = racePool.run(raceDriver, context, input); @@ -304,6 +319,7 @@ if (afterClose.isErr()) { providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "instance" }, + capabilities: TEST_CAPABILITIES, async createRuntime() { createAttempts += 1; if (createAttempts === 1) { @@ -358,6 +374,7 @@ if (afterClose.isErr()) { providerInstanceId: "codex", provider: "codex", runtimePolicy: { scope: "instance", idleTimeoutMs: Number.POSITIVE_INFINITY }, + capabilities: TEST_CAPABILITIES, createRuntime: async () => Result.ok(releaseRuntime), }; @@ -384,6 +401,7 @@ if (afterClose.isErr()) { providerInstanceId, provider: "codex", runtimePolicy, + capabilities: TEST_CAPABILITIES, createRuntime: async () => { creates += 1; return Result.ok(new FakeRuntime()); diff --git a/src/local-agent-runtime.ts b/src/local-agent-runtime.ts index eb3335189..e73e3e7d6 100644 --- a/src/local-agent-runtime.ts +++ b/src/local-agent-runtime.ts @@ -16,6 +16,7 @@ export interface LocalAgentRunInput { effort?: string; modelOverrideRequested?: boolean; effortOverrideRequested?: boolean; + signal?: AbortSignal; } export interface LocalAgentRunResult { @@ -56,6 +57,14 @@ export interface LocalAgentRuntimePolicy { sessionIdleTimeoutMs?: number; } +export interface LocalAgentCapabilities { + sessions: { resume: boolean; close: boolean }; + turns: { interrupt: boolean }; + configuration: { modelOverride: boolean; effortOverride: boolean }; + permissions: { enforcement: "native" | "client-boundary" | "unsupported" }; + mcp: { supported: boolean }; +} + /** * A runtime is deliberately disposable. Nothing from this interface is * persisted; the provider session ID in LocalAgentStore is the continuation @@ -76,5 +85,6 @@ export interface LocalAgentDriver { readonly providerInstanceId: LocalAgentProviderInstanceId; readonly provider: LocalAgentDriverKind; readonly runtimePolicy: LocalAgentRuntimePolicy; + readonly capabilities: LocalAgentCapabilities; createRuntime(context: LocalAgentRuntimeContext): Promise>; }