Q55: the operator can stop in-flight runs; pause stops them too

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
ZacharyZhang-NYandClaude Opus 5.5 committed 2026-10-08 09:59:03 -04:00
1 parent e809cfefea
commit 9e1f8da11c
18 files changed
+259 -17

No files matched your search

+9 -1
View File
@@ -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)));
+2
View File
@@ -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<string
if (!verdict.approved) {
return `The independent audit rejected closing the goal: ${verdict.reason} The goal stays active — close that gap or create the tasks that will.`;
}
throwIfRunStopped();
deps.db.prepare("update goals set status = 'achieved', stalled_reason = null, updated_at = ? where id = ? and org_id = ?")
.run(new Date().toISOString(), goal.id, ctx.orgId);
new AuditService(deps.db).record({
+13 -6
View File
@@ -8,6 +8,7 @@ import type { AgentTurnMessage, ToolCall, ToolDefinition } from "./provider-tool
import { isComposioTool, executeComposioTool } from "./composio/composio-tool-bridge";
import { capToolOutput, compactToolHistory } from "./run-context-compaction";
import { fitContextBudget } from "./context-budget";
import { throwIfRunStopped, untilStopped } from "./run-control";
const MAX_ITERATIONS = 12;
@@ -67,14 +68,19 @@ export async function runAgentLoop(deps: AgentRunLoopDeps, input: AgentRunLoopIn
console.info(`[run-loop] iteration ${iteration}: elided ${compaction.elided} earlier tool result(s)`);
}
fitContextBudget(input.instructions, messages, allTools, input.inputBudgetTokens, iteration);
const turn = await deps.providers.runToolTurn({
throwIfRunStopped();
const turn = await untilStopped(deps.providers.runToolTurn({
orgId: input.orgId,
providerId: input.providerId,
workspaceId: input.workspaceId,
instructions: input.instructions,
messages,
tools: allTools,
})).catch((error: unknown) => {
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),
@@ -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<Record<string, unknown>> {
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");
+2 -2
View File
@@ -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<string>();
+13 -2
View File
@@ -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<string, unknown>;
@@ -69,7 +70,7 @@ export async function postEventStream(
body: JsonRecord,
timeouts: ProviderRequestTimeouts = {},
): Promise<ProviderEventStreamResponse> {
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<ProviderJsonResponse> {
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<T>(controller: AbortController, timeoutMs: number, operation: () => Promise<T>): Promise<T> {
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 {
+54
View File
@@ -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<AbortSignal>();
const controllers = new Map<string, AbortController>();
/** Runs a session under a stop signal that every provider request and tool inside it observes. */
export async function withRunControl<T>(runId: string, operation: () => Promise<T>): Promise<T> {
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<T>(operation: Promise<T>): Promise<T> {
const signal = currentRunSignal();
if (!signal) return operation;
if (signal.aborted) return Promise.reject(new Error(RUN_STOPPED_MESSAGE));
return new Promise<T>((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));
});
}
+19 -2
View File
@@ -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}`;
+2
View File
@@ -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<stri
killedByTimeout = true;
child.kill();
}, COMMAND_TIMEOUT_MS);
currentRunSignal()?.addEventListener("abort", () => child.kill(), { once: true });
const collect = (chunk: Buffer): void => output.push(chunk);
child.stdout.on("data", collect);
child.stderr.on("data", collect);
+1
View File
@@ -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),
+3
View File
@@ -32,6 +32,9 @@ export const zh: Record<string, string> = {
"superseded": "已被取代",
"stalled": "已停滞",
"Resume": "恢复",
"Stop": "停止",
"Stopped": "已停止",
"stopped": "已停止",
"Why it did not run": "未运行的原因",
"Done tasks do not run": "已完成的任务不能再运行",
"A session is already queued or running": "已有会话在排队或运行中",
+1
View File
@@ -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;
}
+14 -4
View File
@@ -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) {
</aside>
<div style={{ flex: 1, minWidth: 0, overflow: "auto", padding: 24 }}>
{selected ? (
<RunDetail run={selected} tasks={tasks} agents={agents} providers={providers} providerRequests={providerRequests} />
<RunDetail run={selected} tasks={tasks} agents={agents} providers={providers} providerRequests={providerRequests} actions={actions} />
) : (
<div style={{ display: "grid", placeItems: "center", height: "100%", color: "var(--text-3)", fontSize: 14 }}>
{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<string | null>(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 (
<div>
@@ -125,7 +132,9 @@ function RunDetail({ run, tasks, agents, providers, providerRequests }: {
<div style={{ fontSize: 12, color: "var(--text-3)" }}>{run.id}</div>
</div>
<Pill tone={runTone(run.state)}>{runLabel(run.state)}</Pill>
{run.state === "running" && <Button size="small" onClick={() => stop(run.id)}>{t("Stop")}</Button>}
</div>
{stopError && <div className="form-error" role="alert">{stopError}</div>}
<div style={{ display: "grid", gridTemplateColumns: "120px 1fr", gap: "8px 16px", fontSize: 14, marginBottom: 20 }}>
<MetaLabel>{t("Agent")}</MetaLabel><div>{agent?.name ?? "—"} ({agent?.role ?? "—"})</div>
@@ -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;
}
@@ -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("_", " ");
}
+4
View File
@@ -48,6 +48,10 @@ export interface SetAutonomousPausedInput {
paused: boolean;
}
export interface StopRunInput {
runId: string;
}
export interface UpdateEmployeeProfileInput {
orgId: string;
employeeId: string;
+2
View File
@@ -7,6 +7,7 @@ import type {
CancelProviderResponseInput,
CreateAgentInput,
SetAutonomousPausedInput,
StopRunInput,
SetIntegrationKeyInput,
ClearIntegrationKeyInput,
UpdateAgentInput,
@@ -119,6 +120,7 @@ export interface DesktopApi {
updateAgent(input: UpdateAgentInput): Promise<DesktopResult<AppSnapshot>>;
updateEmployeeProfile(input: UpdateEmployeeProfileInput): Promise<DesktopResult<AppSnapshot>>;
setAutonomousPaused(input: SetAutonomousPausedInput): Promise<DesktopResult<AppSnapshot>>;
stopRun(input: StopRunInput): Promise<DesktopResult<AppSnapshot>>;
setIntegrationKey(input: SetIntegrationKeyInput): Promise<DesktopResult<AppSnapshot>>;
clearIntegrationKey(input: ClearIntegrationKeyInput): Promise<DesktopResult<AppSnapshot>>;
createTask(input: CreateTaskInput): Promise<DesktopResult<AppSnapshot>>;
+1
View File
@@ -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",