From 9e1f8da11cdc1075a3a2162a0c62ca270dd07e4f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 8 Oct 2026 09:59:03 -0400 Subject: [PATCH] Q55: the operator can stop in-flight runs; pause stops them too Co-Authored-By: Claude Opus 5.5 --- src/main/app/ipc-handlers.ts | 10 +- src/main/services/agent-goal-tools.ts | 2 + src/main/services/agent-run-loop.ts | 19 ++- src/main/services/composio/composio-client.ts | 10 ++ src/main/services/heartbeat-scheduler.ts | 4 +- src/main/services/provider-http.ts | 15 ++- src/main/services/run-control.ts | 54 +++++++++ src/main/services/task-execution-service.ts | 21 +++- src/main/services/workspace-command.ts | 2 + src/preload/index.ts | 1 + src/renderer/src/i18n/zh.ts | 3 + src/renderer/src/pages/HomePage.tsx | 1 + src/renderer/src/pages/RunsPage.tsx | 18 ++- src/renderer/src/pages/work/task-helpers.ts | 1 + src/shared/commands/organization.ts | 4 + src/shared/desktop-api.ts | 2 + src/shared/ipc.ts | 1 + tests/integration/run-stop.test.ts | 108 ++++++++++++++++++ 18 files changed, 259 insertions(+), 17 deletions(-) create mode 100644 src/main/services/run-control.ts create mode 100644 tests/integration/run-stop.test.ts diff --git a/src/main/app/ipc-handlers.ts b/src/main/app/ipc-handlers.ts index a7033520..179b40a3 100644 --- a/src/main/app/ipc-handlers.ts +++ b/src/main/app/ipc-handlers.ts @@ -33,6 +33,7 @@ import { importStagedSkills, stageBuiltinSkillsLibrary } from "../services/built import { coalesce } from "./coalesce"; import { normalizeExternalBrowserUrl } from "./external-url"; import { resumeGoal } from "../services/goal-review-service"; +import { stopAllRuns, stopRun } from "../services/run-control"; export interface IpcDeps { connection: DatabaseConnection; @@ -147,7 +148,14 @@ export function registerIpcHandlers(deps: IpcDeps): void { ipcMain.handle(IPC_CHANNELS.createAgent, (_event, input) => mutate(() => organization.createAgent(input))); ipcMain.handle(IPC_CHANNELS.updateAgent, (_event, input) => mutate(() => organization.updateAgent(input))); ipcMain.handle(IPC_CHANNELS.updateEmployeeProfile, (_event, input) => mutate(() => organization.updateEmployeeProfile(input))); - ipcMain.handle(IPC_CHANNELS.setAutonomousPaused, (_event, input) => mutate(() => organization.setAutonomousPaused(Boolean(input.paused)))); + ipcMain.handle(IPC_CHANNELS.setAutonomousPaused, (_event, input) => mutate(() => { + organization.setAutonomousPaused(Boolean(input.paused)); + // Pausing also stops the sessions already running, not just future scheduling. + if (input.paused) stopAllRuns(); + })); + ipcMain.handle(IPC_CHANNELS.stopRun, (_event, input) => mutate(() => { + if (!stopRun(input.runId)) throw new Error("That run is not running anymore."); + })); ipcMain.handle(IPC_CHANNELS.setIntegrationKey, (_event, input) => mutate(() => integrationKeys.write(input.name, input.value))); ipcMain.handle(IPC_CHANNELS.clearIntegrationKey, (_event, input) => mutate(() => integrationKeys.clear(input.name))); ipcMain.handle(IPC_CHANNELS.createTask, (_event, input) => mutate(() => organization.createTask(input))); diff --git a/src/main/services/agent-goal-tools.ts b/src/main/services/agent-goal-tools.ts index ca4c38a7..6ee02394 100644 --- a/src/main/services/agent-goal-tools.ts +++ b/src/main/services/agent-goal-tools.ts @@ -6,6 +6,7 @@ import type { AgentToolContext, AgentToolDeps } from "./agent-tools"; import type { JsonRecord } from "./provider-http"; import type { ProviderService } from "./provider-service"; import type { AgentTurnMessage, ToolCall, ToolDefinition } from "./provider-tool-turn"; +import { throwIfRunStopped } from "./run-control"; const AUDIT_MAX_ITERATIONS = 6; const STATUS_TASK_LINES = 8; @@ -83,6 +84,7 @@ export async function completeGoalTool(args: GoalCompletionArgs): Promise { + throwIfRunStopped(); + throw error; }); + throwIfRunStopped(); lastRequestId = turn.requestId; input.onModelTurn?.(); @@ -96,18 +102,19 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn messages.push({ role: "assistant", text: turn.text, toolCalls: turn.toolCalls }); const results: Array<{ toolCallId: string; toolName: string; output: string }> = []; for (const call of turn.toolCalls) { + throwIfRunStopped(); toolCallCount += 1; let output: string; if (isComposioTool(call.name)) { - output = await executeComposioTool(deps.tools.db, deps.providers.integrationKeys, input.orgId, input.toolContext.agent.id, call); + output = await untilStopped(executeComposioTool(deps.tools.db, deps.providers.integrationKeys, input.orgId, input.toolContext.agent.id, call)); } else if (isWorkspaceTool(call.name)) { - output = await executeWorkspaceTool(input.toolContext, call); + output = await untilStopped(executeWorkspaceTool(input.toolContext, call)); } else if (isWebTool(call.name)) { - output = await executeWebTool(call); + output = await untilStopped(executeWebTool(call)); } else if (call.name === "complete_goal") { // Runs an independent audit turn against the provider, so it needs // the loop's provider access rather than the sync dispatcher. - output = await completeGoalTool({ + output = await untilStopped(completeGoalTool({ deps: deps.tools, ctx: input.toolContext, providers: deps.providers, @@ -115,7 +122,7 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn workspaceId: input.workspaceId, inputBudgetTokens: input.inputBudgetTokens, call, - }); + })); } else { output = dispatchAgentTool(deps.tools, input.toolContext, call, { announceTaskAssignments: shouldAnnounceTaskAssignment(call, turn.toolCalls), diff --git a/src/main/services/composio/composio-client.ts b/src/main/services/composio/composio-client.ts index 0d190042..00656efc 100644 --- a/src/main/services/composio/composio-client.ts +++ b/src/main/services/composio/composio-client.ts @@ -1,4 +1,5 @@ import { net } from "electron"; +import { currentRunSignal, RUN_STOPPED_MESSAGE } from "../run-control"; const BASE_URL = "https://backend.composio.dev/api/v3"; @@ -153,6 +154,15 @@ export class ComposioClient { ): Promise> { return new Promise((resolve, reject) => { const request = net.request({ method, url }); + const runSignal = currentRunSignal(); + if (runSignal?.aborted) { + reject(new Error(RUN_STOPPED_MESSAGE)); + return; + } + runSignal?.addEventListener("abort", () => { + request.abort(); + reject(new Error(RUN_STOPPED_MESSAGE)); + }, { once: true }); request.setHeader("x-api-key", this.apiKey); request.setHeader("Accept", "application/json"); if (body) request.setHeader("Content-Type", "application/json"); diff --git a/src/main/services/heartbeat-scheduler.ts b/src/main/services/heartbeat-scheduler.ts index 9b650975..5574ff24 100644 --- a/src/main/services/heartbeat-scheduler.ts +++ b/src/main/services/heartbeat-scheduler.ts @@ -55,7 +55,7 @@ export class HeartbeatScheduler { and not exists (select 1 from tasks c where c.parent_task_id = t.id and c.status <> 'done') and (t.origin is not null or t.goal_id is null or exists (select 1 from goals g where g.id = t.goal_id and g.status = 'active')) and (a.heartbeat_interval_sec <= 0 or not exists ( - select 1 from task_runs hb where hb.agent_id = a.id and hb.trigger = 'heartbeat' and hb.state <> 'superseded' + select 1 from task_runs hb where hb.agent_id = a.id and hb.trigger = 'heartbeat' and hb.state not in ('superseded', 'stopped') and hb.created_at > strftime('%Y-%m-%dT%H:%M:%fZ', ?, '-' || a.heartbeat_interval_sec || ' seconds') )) order by case t.priority when 'high' then 0 when 'normal' then 1 else 2 end, t.created_at @@ -64,7 +64,7 @@ export class HeartbeatScheduler { const cooling = new Set(this.cooldowns.cooledProviderIds()); const attemptMessages = this.db.prepare(` select r.error_message from task_runs r join tasks t on t.id = r.task_id - where r.task_id = ? and r.state <> 'superseded' and r.rowid > t.run_budget_floor + where r.task_id = ? and r.state not in ('superseded', 'stopped') and r.rowid > t.run_budget_floor `); const created: string[] = []; const pacedAgents = new Set(); diff --git a/src/main/services/provider-http.ts b/src/main/services/provider-http.ts index d0f3c314..86fb5044 100644 --- a/src/main/services/provider-http.ts +++ b/src/main/services/provider-http.ts @@ -1,5 +1,6 @@ import type { ProviderObservedRateLimitState } from "../../shared/provider-runtime-domain"; import { parseProviderRateLimitHeaders } from "./provider-rate-limit-registry"; +import { currentRunSignal, throwIfRunStopped } from "./run-control"; export type JsonRecord = Record; @@ -69,7 +70,7 @@ export async function postEventStream( body: JsonRecord, timeouts: ProviderRequestTimeouts = {}, ): Promise { - const controller = new AbortController(); + const controller = linkedController(); const response = await withDeadline(controller, timeouts.headersTimeoutMs ?? HEADERS_TIMEOUT_MS, () => fetch(endpoint, { method: "POST", headers: { @@ -92,7 +93,7 @@ export async function postEventStream( } async function requestJson(endpoint: string, init: RequestInit, timeouts: ProviderRequestTimeouts = {}): Promise { - const controller = new AbortController(); + const controller = linkedController(); const response = await withDeadline(controller, timeouts.headersTimeoutMs ?? HEADERS_TIMEOUT_MS, () => fetch(endpoint, { ...init, signal: controller.signal })); const rateLimitState = parseProviderRateLimitHeaders(response.headers); const raw = await withDeadline(controller, timeouts.bodyTimeoutMs ?? BODY_TIMEOUT_MS, () => response.text()); @@ -110,11 +111,21 @@ async function requestJson(endpoint: string, init: RequestInit, timeouts: Provid * then stalls the body must not hang the caller, so the body read gets its * own (longer) deadline after the headers arrive. */ +/** A request controller that also aborts when the operator stops the run it belongs to. */ +function linkedController(): AbortController { + const controller = new AbortController(); + const runSignal = currentRunSignal(); + if (runSignal?.aborted) controller.abort(); + else runSignal?.addEventListener("abort", () => controller.abort(), { once: true }); + return controller; +} + async function withDeadline(controller: AbortController, timeoutMs: number, operation: () => Promise): Promise { const timer = setTimeout(() => controller.abort(), timeoutMs); try { return await operation(); } catch (error) { + throwIfRunStopped(); if (controller.signal.aborted) throw new ProviderHttpError("Provider request timed out.", null); throw error; } finally { diff --git a/src/main/services/run-control.ts b/src/main/services/run-control.ts new file mode 100644 index 00000000..20686724 --- /dev/null +++ b/src/main/services/run-control.ts @@ -0,0 +1,54 @@ +import { AsyncLocalStorage } from "node:async_hooks"; + +/** The error message a run ends with when the operator stops it. */ +export const RUN_STOPPED_MESSAGE = "Stopped by the operator."; + +const runSignals = new AsyncLocalStorage(); +const controllers = new Map(); + +/** Runs a session under a stop signal that every provider request and tool inside it observes. */ +export async function withRunControl(runId: string, operation: () => Promise): Promise { + const controller = new AbortController(); + controllers.set(runId, controller); + try { + return await runSignals.run(controller.signal, operation); + } finally { + controllers.delete(runId); + } +} + +/** Stops one in-flight run; false when it is not running in this process. */ +export function stopRun(runId: string): boolean { + const controller = controllers.get(runId); + if (!controller) return false; + console.info(`[run-control] stopping run ${runId}`); + controller.abort(); + return true; +} + +export function stopAllRuns(): number { + const ids = [...controllers.keys()]; + for (const id of ids) stopRun(id); + return ids.length; +} + +/** The stop signal of the run this code executes in, if any. */ +export function currentRunSignal(): AbortSignal | undefined { + return runSignals.getStore(); +} + +export function throwIfRunStopped(): void { + if (currentRunSignal()?.aborted) throw new Error(RUN_STOPPED_MESSAGE); +} + +/** Settles with the operation, or rejects as soon as the run is stopped; work without its own abort is abandoned. */ +export function untilStopped(operation: Promise): Promise { + const signal = currentRunSignal(); + if (!signal) return operation; + if (signal.aborted) return Promise.reject(new Error(RUN_STOPPED_MESSAGE)); + return new Promise((resolve, reject) => { + const onAbort = () => reject(new Error(RUN_STOPPED_MESSAGE)); + signal.addEventListener("abort", onAbort, { once: true }); + operation.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort)); + }); +} diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index e2f64333..c75b529a 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -13,6 +13,7 @@ import { runAgentLoop, runTools } from "./agent-run-loop"; import type { RunModel } from "./skill-runtime"; import { combineMidRunPolls, createGoalChangePoller } from "./goal-change-poller"; import { recordRunAction, runJournalLines } from "./run-journal"; +import { currentRunSignal, RUN_STOPPED_MESSAGE, withRunControl } from "./run-control"; import type { AgentToolContext } from "./agent-tools"; import { buildAgentInstructions, @@ -46,6 +47,7 @@ const TASK_RUN_STATE = { COMPLETED: "completed", FAILED: "failed", SUPERSEDED: "superseded", + STOPPED: "stopped", } as const; @@ -97,7 +99,7 @@ export class TaskExecutionService { const workspace = loadWorkspace(this.db, task.workspace_id); const isolation = createIsolationContext(this.db, input.orgId, task, agent, provider, workspace); const runId = this.createRun(input.orgId, task.id, agent.id, provider.id, isolation); - return this.executeAgentRun({ orgId: input.orgId, runId, task, agent, provider, workspace, isolation }); + return withRunControl(runId, () => this.executeAgentRun({ orgId: input.orgId, runId, task, agent, provider, workspace, isolation })); } /** Executes a claimed run after re-checking that the task still matches it; a mismatch supersedes the run. */ @@ -155,7 +157,7 @@ export class TaskExecutionService { if (task && task.origin === null) recordRunMemory(this.db, { orgId, runId, projectId: task.project_id, taskTitle: task.title, outcome: "failed", text: message }); throw error; } - return this.executeAgentRun(prepared); + return withRunControl(runId, () => this.executeAgentRun(prepared)); } private async executeAgentRun(args: { @@ -280,6 +282,11 @@ export class TaskExecutionService { return runId; } catch (error) { const message = error instanceof Error ? error.message : String(error); + // A stop wins over whatever error the interrupted step raised. + if (currentRunSignal()?.aborted) { + this.markStopped(orgId, runId, providerRequestId, { taskId: task.id, agentId: agent.id, providerId: provider.id }); + throw error; + } this.failRun(orgId, runId, providerRequestId, message); if (isTransientProviderFailure(message)) this.cooldowns.park(provider, message); else if (task.origin === null) recordRunMemory(this.db, { orgId, runId, projectId: task.project_id, taskTitle: task.title, outcome: "failed", text: message }); @@ -415,6 +422,16 @@ export class TaskExecutionService { this.organization.updateTaskStatus(orgId, run.task_id, TASK_STATUS.DONE); } + /** The operator stopped the run: not a failure, no retry budget spent, no memory recorded. */ + private markStopped(orgId: string, runId: string, providerRequestId: string | null, payload: { taskId: string; agentId: string; providerId: string }): void { + this.db.prepare(` + update task_runs set state = ?, provider_request_id = coalesce(?, provider_request_id), completed_at = ?, error_message = ? + where id = ? and org_id = ? + `).run(TASK_RUN_STATE.STOPPED, providerRequestId, new Date().toISOString(), RUN_STOPPED_MESSAGE, runId, orgId); + this.closeChatWakeTask(orgId, runId); + this.audit.record({ orgId, entityType: "task_run", entityId: runId, actorType: "employee", actorId: null, action: "stopped", payload }); + } + /** Closes a queued run that no longer matches its task; it is not a failure and spends no retry budget. */ private supersedeRun(orgId: string, runId: string, run: RunRow, reason: string): string { const message = `Superseded: ${reason}`; diff --git a/src/main/services/workspace-command.ts b/src/main/services/workspace-command.ts index 41a90db0..af83179b 100644 --- a/src/main/services/workspace-command.ts +++ b/src/main/services/workspace-command.ts @@ -1,4 +1,5 @@ import { spawn } from "node:child_process"; +import { currentRunSignal } from "./run-control"; const COMMAND_TIMEOUT_MS = 120_000; const KEEP_BYTES = 8_000; @@ -51,6 +52,7 @@ export function runCommand(workspaceRoot: string, command: string): Promise child.kill(), { once: true }); const collect = (chunk: Buffer): void => output.push(chunk); child.stdout.on("data", collect); child.stderr.on("data", collect); diff --git a/src/preload/index.ts b/src/preload/index.ts index 91e2b7e5..234d3300 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -17,6 +17,7 @@ const api: DesktopApi = { updateAgent: (input) => invoke(IPC_CHANNELS.updateAgent, input), updateEmployeeProfile: (input) => invoke(IPC_CHANNELS.updateEmployeeProfile, input), setAutonomousPaused: (input) => invoke(IPC_CHANNELS.setAutonomousPaused, input), + stopRun: (input) => invoke(IPC_CHANNELS.stopRun, input), setIntegrationKey: (input) => invoke(IPC_CHANNELS.setIntegrationKey, input), clearIntegrationKey: (input) => invoke(IPC_CHANNELS.clearIntegrationKey, input), createTask: (input) => invoke(IPC_CHANNELS.createTask, input), diff --git a/src/renderer/src/i18n/zh.ts b/src/renderer/src/i18n/zh.ts index 28232179..8b6e97b2 100644 --- a/src/renderer/src/i18n/zh.ts +++ b/src/renderer/src/i18n/zh.ts @@ -32,6 +32,9 @@ export const zh: Record = { "superseded": "已被取代", "stalled": "已停滞", "Resume": "恢复", + "Stop": "停止", + "Stopped": "已停止", + "stopped": "已停止", "Why it did not run": "未运行的原因", "Done tasks do not run": "已完成的任务不能再运行", "A session is already queued or running": "已有会话在排队或运行中", diff --git a/src/renderer/src/pages/HomePage.tsx b/src/renderer/src/pages/HomePage.tsx index b4a0b6dc..d4a9f054 100644 --- a/src/renderer/src/pages/HomePage.tsx +++ b/src/renderer/src/pages/HomePage.tsx @@ -360,5 +360,6 @@ function runStateLabel(state: string): string { if (state === "completed") return t("Done"); if (state === "failed") return t("Failed"); if (state === "superseded") return t("Superseded"); + if (state === "stopped") return t("Stopped"); return state; } diff --git a/src/renderer/src/pages/RunsPage.tsx b/src/renderer/src/pages/RunsPage.tsx index 9419ba02..c0f404a8 100644 --- a/src/renderer/src/pages/RunsPage.tsx +++ b/src/renderer/src/pages/RunsPage.tsx @@ -1,5 +1,5 @@ import { useMemo, useState } from "react"; -import { Avatar, Pill, Segmented } from "../shared/ui"; +import { Avatar, Button, Pill, Segmented } from "../shared/ui"; import type { AgentRecord, ProviderRecord, TaskRecord, TaskRunRecord } from "../../../shared/domain"; import type { PageProps } from "./types"; import { activeOrgId, formatRelative } from "./types"; @@ -14,7 +14,7 @@ const FILTERS: Array<{ id: StatusFilter; label: string }> = [ { id: "failed", label: t("Failed") }, ]; -export function RunsPage({ snapshot }: PageProps) { +export function RunsPage({ snapshot, actions }: PageProps) { const orgId = activeOrgId(snapshot); const taskRuns = useMemo(() => snapshot.taskRuns.filter((r) => r.orgId === orgId), [snapshot.taskRuns, orgId]); const tasks = useMemo(() => snapshot.tasks.filter((t) => t.orgId === orgId), [snapshot.tasks, orgId]); @@ -89,7 +89,7 @@ export function RunsPage({ snapshot }: PageProps) {
{selected ? ( - + ) : (
{t("Select a run")} @@ -100,7 +100,8 @@ export function RunsPage({ snapshot }: PageProps) { ); } -function RunDetail({ run, tasks, agents, providers, providerRequests }: { +function RunDetail({ run, tasks, agents, providers, providerRequests, actions }: { + actions: PageProps["actions"]; run: TaskRunRecord; tasks: TaskRecord[]; agents: AgentRecord[]; @@ -115,6 +116,12 @@ function RunDetail({ run, tasks, agents, providers, providerRequests }: { const started = run.startedAt ? new Date(run.startedAt) : null; const completed = run.completedAt ? new Date(run.completedAt) : null; const durationMs = started && completed ? completed.getTime() - started.getTime() : null; + const [stopError, setStopError] = useState(null); + const stop = (runId: string) => { + setStopError(null); + actions.run(() => window.agentCompany.stopRun({ runId })) + .catch((error: unknown) => setStopError(error instanceof Error ? error.message : String(error))); + }; return (
@@ -125,7 +132,9 @@ function RunDetail({ run, tasks, agents, providers, providerRequests }: {
{run.id}
{runLabel(run.state)} + {run.state === "running" && }
+ {stopError &&
{stopError}
}
{t("Agent")}
{agent?.name ?? "—"} ({agent?.role ?? "—"})
@@ -204,6 +213,7 @@ function runLabel(state: string): string { if (state === "completed") return t("Done"); if (state === "failed") return t("Failed"); if (state === "superseded") return t("Superseded"); + if (state === "stopped") return t("Stopped"); return state; } diff --git a/src/renderer/src/pages/work/task-helpers.ts b/src/renderer/src/pages/work/task-helpers.ts index 43b678aa..21c5fd1e 100644 --- a/src/renderer/src/pages/work/task-helpers.ts +++ b/src/renderer/src/pages/work/task-helpers.ts @@ -28,6 +28,7 @@ export function prettyRunState(state: string): string { if (state === "completed") return t("completed"); if (state === "failed") return t("failed"); if (state === "superseded") return t("superseded"); + if (state === "stopped") return t("stopped"); return state.replaceAll("_", " "); } diff --git a/src/shared/commands/organization.ts b/src/shared/commands/organization.ts index 022d063e..9efbddd0 100644 --- a/src/shared/commands/organization.ts +++ b/src/shared/commands/organization.ts @@ -48,6 +48,10 @@ export interface SetAutonomousPausedInput { paused: boolean; } +export interface StopRunInput { + runId: string; +} + export interface UpdateEmployeeProfileInput { orgId: string; employeeId: string; diff --git a/src/shared/desktop-api.ts b/src/shared/desktop-api.ts index be88dfda..16538f8a 100644 --- a/src/shared/desktop-api.ts +++ b/src/shared/desktop-api.ts @@ -7,6 +7,7 @@ import type { CancelProviderResponseInput, CreateAgentInput, SetAutonomousPausedInput, + StopRunInput, SetIntegrationKeyInput, ClearIntegrationKeyInput, UpdateAgentInput, @@ -119,6 +120,7 @@ export interface DesktopApi { updateAgent(input: UpdateAgentInput): Promise>; updateEmployeeProfile(input: UpdateEmployeeProfileInput): Promise>; setAutonomousPaused(input: SetAutonomousPausedInput): Promise>; + stopRun(input: StopRunInput): Promise>; setIntegrationKey(input: SetIntegrationKeyInput): Promise>; clearIntegrationKey(input: ClearIntegrationKeyInput): Promise>; createTask(input: CreateTaskInput): Promise>; diff --git a/src/shared/ipc.ts b/src/shared/ipc.ts index 8ea0f7b5..51727091 100644 --- a/src/shared/ipc.ts +++ b/src/shared/ipc.ts @@ -9,6 +9,7 @@ export const IPC_CHANNELS = { updateAgent: "agentcompany:update-agent", updateEmployeeProfile: "agentcompany:update-employee-profile", setAutonomousPaused: "agentcompany:set-autonomous-paused", + stopRun: "agentcompany:stop-run", setIntegrationKey: "agentcompany:set-integration-key", clearIntegrationKey: "agentcompany:clear-integration-key", createTask: "agentcompany:create-task", diff --git a/tests/integration/run-stop.test.ts b/tests/integration/run-stop.test.ts new file mode 100644 index 00000000..44e31038 --- /dev/null +++ b/tests/integration/run-stop.test.ts @@ -0,0 +1,108 @@ +import { randomUUID } from "node:crypto"; +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "vitest"; +import { DatabaseConnection } from "../../src/main/database/connection"; +import { OnboardingService } from "../../src/main/services/onboarding-service"; +import { ProviderService } from "../../src/main/services/provider-service"; +import { providerAdapters } from "../../src/main/services/provider-adapters"; +import { TaskExecutionService } from "../../src/main/services/task-execution-service"; +import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; +import { postJson } from "../../src/main/services/provider-http"; +import { RUN_STOPPED_MESSAGE, stopRun, withRunControl } from "../../src/main/services/run-control"; +import type { AgentTurnResult, ProviderToolTurnInput } from "../../src/main/services/provider-tool-turn"; + +const testVault = { + encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), + decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"), +}; + +function company(turn: (input: ProviderToolTurnInput) => Promise) { + const connection = new DatabaseConnection(join(tmpdir(), `ac-stop-${randomUUID()}.db`)); + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, turn); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + const execution = new TaskExecutionService(connection.db, providers); + let orgId = ""; + connection.transaction(() => { + orgId = onboarding.bootstrap({ + companyName: "FlowPilot", companyDescription: "CSM analytics.", employeeName: "Yang", workspacePath: "", + executionMode: "ai_ceo", autoApproveHires: true, reviewDeliverables: false, + goalTitle: "Ship the MVP.", goalDescription: "A working product.", + providerKey: "openai", providerSecret: "sk-test-key", providerModel: "gpt-5.4", + }).orgId; + }); + return { connection, execution, orgId, claim: () => claimNextQueuedRun(connection)!.id }; +} + +describe("stopping runs", () => { + test("a stopped run executes none of the tools its last turn asked for and spends no retry budget", async () => { + let runId = ""; + const turn = async (): Promise => { + stopRun(runId); + return { text: "Announcing.", toolCalls: [{ id: "c1", name: "send_message", arguments: { channel: "general", body: "Should never be posted." } }], rawResponseJson: "{}" }; + }; + const connection = new DatabaseConnection(join(tmpdir(), `ac-stop-${randomUUID()}.db`)); + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, turn); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + const execution = new TaskExecutionService(connection.db, providers); + let orgId = ""; + connection.transaction(() => { + orgId = onboarding.bootstrap({ + companyName: "FlowPilot", companyDescription: "CSM analytics.", employeeName: "Yang", workspacePath: "", + executionMode: "ai_ceo", autoApproveHires: true, reviewDeliverables: false, + goalTitle: "Ship the MVP.", goalDescription: "A working product.", + providerKey: "openai", providerSecret: "sk-test-key", providerModel: "gpt-5.4", + }).orgId; + }); + runId = claimNextQueuedRun(connection)!.id; + await expect(execution.executeRun(orgId, runId)).rejects.toThrow(RUN_STOPPED_MESSAGE); + expect(connection.db.prepare("select state, error_message from task_runs where id = ?").get(runId)).toEqual({ state: "stopped", error_message: RUN_STOPPED_MESSAGE }); + expect(connection.db.prepare("select count(*) as n from chat_messages where body_md = 'Should never be posted.'").get()).toEqual({ n: 0 }); + expect(connection.db.prepare("select count(*) as n from audit_events where entity_id = ? and action = 'stopped'").get(runId)).toEqual({ n: 1 }); + expect(stopRun(runId)).toBe(false); + connection.close(); + }); + + test("a stop during the goal audit keeps the goal open, and any error after a stop still counts as a stop", async () => { + let runId = ""; + const turn = async (input: ProviderToolTurnInput): Promise => { + if (input.instructions?.startsWith("You are the independent completion auditor")) { + stopRun(runId); + return { text: "Verified. ", toolCalls: [], rawResponseJson: "{}" }; + } + return { text: "Closing.", toolCalls: [{ id: "c1", name: "complete_goal", arguments: { goal_id: goalId, evidence: "Shipped." } }], rawResponseJson: "{}" }; + }; + const { connection, execution, orgId, claim } = company(turn); + const goalId = (connection.db.prepare("select id from goals where org_id = ?").get(orgId) as { id: string }).id; + runId = claim(); + await expect(execution.executeRun(orgId, runId)).rejects.toThrow(RUN_STOPPED_MESSAGE); + expect(connection.db.prepare("select status from goals where id = ?").get(goalId)).toEqual({ status: "active" }); + expect(connection.db.prepare("select state from task_runs where id = ?").get(runId)).toEqual({ state: "stopped" }); + + // The interrupted step may raise its own error; the run is still recorded as stopped, not failed. + const failing = company(async () => { + stopRun(failingRun); + throw new Error("Provider overloaded"); + }); + const failingRun = failing.claim(); + await expect(failing.execution.executeRun(failing.orgId, failingRun)).rejects.toThrow(); + expect(failing.connection.db.prepare("select state from task_runs where id = ?").get(failingRun)).toEqual({ state: "stopped" }); + connection.close(); + failing.connection.close(); + }); + + test("stopping a run aborts its provider request in flight", async () => { + const server = createServer(() => { /* never answers */ }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const port = (server.address() as AddressInfo).port; + const started = Date.now(); + const request = withRunControl("run_http", () => postJson(`http://127.0.0.1:${port}/v1`, {}, { hello: "world" })); + setTimeout(() => stopRun("run_http"), 50); + await expect(request).rejects.toThrow(RUN_STOPPED_MESSAGE); + expect(Date.now() - started).toBeLessThan(5_000); + server.closeAllConnections(); + server.close(); + }); +});