From a075f931d5c4ab7726b0b0b40f2c6bfbc03d2e91 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Fri, 9 Oct 2026 02:11:11 -0400 Subject: [PATCH] Q157: a run can be cancelled before it starts, and a failed or stopped one retried or resumed Co-Authored-By: Claude Opus 5.5 --- src/main/app/ipc-contracts/organization.ts | 1 + src/main/app/ipc-handlers.ts | 2 + src/main/services/chat-wake-close.ts | 14 +++++++ src/main/services/run-cancel.ts | 15 +++++++ src/main/services/task-execution-service.ts | 18 ++------- src/preload/index.ts | 1 + src/renderer/src/i18n/zh.ts | 4 ++ src/renderer/src/pages/RunsPage.tsx | 30 ++++++++++---- src/shared/desktop-api.ts | 2 + src/shared/ipc.ts | 1 + tests/e2e/ceo-chat.spec.ts | 3 ++ tests/integration/run-cancel.test.ts | 45 +++++++++++++++++++++ 12 files changed, 114 insertions(+), 22 deletions(-) create mode 100644 src/main/services/chat-wake-close.ts create mode 100644 src/main/services/run-cancel.ts create mode 100644 tests/integration/run-cancel.test.ts diff --git a/src/main/app/ipc-contracts/organization.ts b/src/main/app/ipc-contracts/organization.ts index a007876e..68a89eda 100644 --- a/src/main/app/ipc-contracts/organization.ts +++ b/src/main/app/ipc-contracts/organization.ts @@ -47,6 +47,7 @@ export const organizationContracts = { deleteTask: z.strictObject({ orgId: id, taskId: id }) satisfies z.ZodType, setTaskBlocked: z.strictObject({ orgId: id, taskId: id, reason: z.string().nullable() }) satisfies z.ZodType, taskDeleteImpact: z.strictObject({ orgId: id, taskId: id }), + cancelQueuedRun: z.strictObject({ orgId: id, runId: id }), updateTaskStatus: z.strictObject({ orgId: id, taskId: id, status: z.enum(TASK_STATUS) }) satisfies z.ZodType, runTask: z.strictObject({ orgId: id, taskId: id }) satisfies z.ZodType, createGoal: z.strictObject({ orgId: id, title: z.string(), description: z.string(), ownerAgentId: nullableId }) satisfies z.ZodType, diff --git a/src/main/app/ipc-handlers.ts b/src/main/app/ipc-handlers.ts index 859ae91f..bdf1b935 100644 --- a/src/main/app/ipc-handlers.ts +++ b/src/main/app/ipc-handlers.ts @@ -37,6 +37,7 @@ import { coalesce } from "./coalesce"; import { normalizeExternalBrowserUrl } from "./external-url"; import { resumeGoal } from "../services/goal-review-service"; import { stopAllRuns, stopRun } from "../services/run-control"; +import { cancelQueuedRun } from "../services/run-cancel"; export interface IpcDeps { connection: DatabaseConnection; @@ -166,6 +167,7 @@ export function registerIpcHandlers(deps: IpcDeps): void { handle("updateTask", (input) => mutate(() => organization.updateTask(input))); handle("setTaskBlocked", (input) => mutate(() => organization.setTaskBlocked(input))); handle("taskDeleteImpact", (input) => ok(organization.taskDeleteImpact(input.orgId, input.taskId))); + handle("cancelQueuedRun", (input) => mutate(() => cancelQueuedRun(deps.connection.db, input.orgId, input.runId))); handle("deleteTask", (input) => mutate(() => organization.deleteTask(input))); handle("updateTaskStatus", (input) => mutate(() => organization.updateTaskStatus(input.orgId, input.taskId, input.status))); handle("runTask", async (input) => { diff --git a/src/main/services/chat-wake-close.ts b/src/main/services/chat-wake-close.ts new file mode 100644 index 00000000..6174a15a --- /dev/null +++ b/src/main/services/chat-wake-close.ts @@ -0,0 +1,14 @@ +import type { DatabaseSync } from "node:sqlite"; +import { TASK_STATUS } from "../../shared/domain"; +import { OrganizationService } from "./organization-service"; + +/** A chat-wake task only carries its reply run, so the run ending closes it; left open it would strand forever. */ +export function closeChatWakeTask(db: DatabaseSync, orgId: string, runId: string): void { + const run = db.prepare("select r.task_id, t.origin from task_runs r join tasks t on t.id = r.task_id where r.id = ? and r.org_id = ?") + .get(runId, orgId) as { task_id: string; origin: string | null } | undefined; + if (!run || run.origin !== "chat_wake") return; + const task = db.prepare("select status from tasks where id = ? and org_id = ?") + .get(run.task_id, orgId) as { status: string } | undefined; + if (!task || (task.status !== TASK_STATUS.BACKLOG && task.status !== TASK_STATUS.IN_PROGRESS)) return; + new OrganizationService(db).updateTaskStatus(orgId, run.task_id, TASK_STATUS.DONE); +} diff --git a/src/main/services/run-cancel.ts b/src/main/services/run-cancel.ts new file mode 100644 index 00000000..12df29d7 --- /dev/null +++ b/src/main/services/run-cancel.ts @@ -0,0 +1,15 @@ +import type { DatabaseSync } from "node:sqlite"; +import { AuditService } from "./audit-service"; +import { closeChatWakeTask } from "./chat-wake-close"; + +/** Cancels a run that has not started: not a failure, no retry budget spent; a chat reply it carried closes. */ +export function cancelQueuedRun(db: DatabaseSync, orgId: string, runId: string): void { + const result = db.prepare(` + update task_runs set state = 'stopped', completed_at = ?, error_message = 'Cancelled before it started.' + where id = ? and org_id = ? and state = 'queued' + `).run(new Date().toISOString(), runId, orgId); + if (result.changes === 0) throw new Error("That run is no longer queued."); + console.info(`[task-run] run ${runId} cancelled before it started`); + closeChatWakeTask(db, orgId, runId); + new AuditService(db).record({ orgId, entityType: "task_run", entityId: runId, actorType: "employee", actorId: null, action: "cancelled", payload: {} }); +} diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 2817f09b..f34dd24c 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -1,3 +1,4 @@ +import { closeChatWakeTask } from "./chat-wake-close"; import type { DatabaseSync } from "node:sqlite"; import type { RunTaskInput } from "../../shared/commands"; import { TASK_STATUS, WRITING_MODE } from "../../shared/domain"; @@ -436,18 +437,7 @@ export class TaskExecutionService { set state = ?, provider_request_id = coalesce(?, provider_request_id), completed_at = ?, error_message = null where id = ? and org_id = ? `).run(TASK_RUN_STATE.COMPLETED, providerRequestId, new Date().toISOString(), runId, orgId); - if (!checkpoint) this.closeChatWakeTask(orgId, runId); - } - - /** A chat-wake task only carries its reply run, so the run completing closes it; left open it would strand forever. */ - private closeChatWakeTask(orgId: string, runId: string): void { - const run = this.db.prepare("select r.task_id, t.origin from task_runs r join tasks t on t.id = r.task_id where r.id = ? and r.org_id = ?") - .get(runId, orgId) as { task_id: string; origin: string | null } | undefined; - if (!run || run.origin !== "chat_wake") return; - const task = this.db.prepare("select status from tasks where id = ? and org_id = ?") - .get(run.task_id, orgId) as { status: string } | undefined; - if (!task || (task.status !== TASK_STATUS.BACKLOG && task.status !== TASK_STATUS.IN_PROGRESS)) return; - this.organization.updateTaskStatus(orgId, run.task_id, TASK_STATUS.DONE); + if (!checkpoint) closeChatWakeTask(this.db, orgId, runId); } /** The operator stopped the run: not a failure, no retry budget spent, no memory recorded. */ @@ -456,7 +446,7 @@ export class TaskExecutionService { 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(), reason, runId, orgId); - this.closeChatWakeTask(orgId, runId); + closeChatWakeTask(this.db, orgId, runId); this.audit.record({ orgId, entityType: "task_run", entityId: runId, actorType: "system", actorId: null, action: "stopped", payload: { ...payload, reason } }); } @@ -466,7 +456,7 @@ export class TaskExecutionService { console.info(`[task-run] run ${runId} ${message}`); this.db.prepare("update task_runs set state = ?, completed_at = ?, error_message = ? where id = ? and org_id = ?") .run(TASK_RUN_STATE.SUPERSEDED, new Date().toISOString(), message, runId, orgId); - this.closeChatWakeTask(orgId, runId); + closeChatWakeTask(this.db, orgId, runId); this.audit.record({ orgId, entityType: "task_run", diff --git a/src/preload/index.ts b/src/preload/index.ts index 85bc68b8..5479571f 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -92,6 +92,7 @@ const api: DesktopApi = { chatSpaceHistory: (input) => invoke(IPC_CHANNELS.chatSpaceHistory, input), setTaskBlocked: (input) => invoke(IPC_CHANNELS.setTaskBlocked, input), taskDeleteImpact: (input) => invoke(IPC_CHANNELS.taskDeleteImpact, input), + cancelQueuedRun: (input) => invoke(IPC_CHANNELS.cancelQueuedRun, input), revokeConnector: (input) => invoke(IPC_CHANNELS.revokeConnector, input), refreshComposioAccounts: (input) => invoke(IPC_CHANNELS.refreshComposioAccounts, input), search: (input) => invoke(IPC_CHANNELS.search, input), diff --git a/src/renderer/src/i18n/zh.ts b/src/renderer/src/i18n/zh.ts index 3f32d9c8..c9e54b1a 100644 --- a/src/renderer/src/i18n/zh.ts +++ b/src/renderer/src/i18n/zh.ts @@ -51,6 +51,10 @@ export const zh: Record = { "No notes": "无备注", "Showing {shown} of {total}": "显示 {total} 项中的 {shown} 项", "Show earlier decisions": "显示更早的决定", + "Cancel run": "取消运行", + "Running…": "运行中…", + "A cancelled session can start again on the next heartbeat; block the task to keep it stopped.": "取消的会话可能在下一次心跳时再次开始；如需保持停止，请阻塞该任务。", + "A new session on the task picks up from what earlier sessions did and where they stopped.": "任务的新会话会接着之前会话做过的工作和停下的位置继续。", "Edit the task": "编辑任务", "Changing the task replaces this request with a fresh one.": "修改任务会用新的申请替换这一项。", "Delete task": "删除任务", diff --git a/src/renderer/src/pages/RunsPage.tsx b/src/renderer/src/pages/RunsPage.tsx index 2d924975..25587fa6 100644 --- a/src/renderer/src/pages/RunsPage.tsx +++ b/src/renderer/src/pages/RunsPage.tsx @@ -90,7 +90,7 @@ export function RunsPage({ snapshot, actions, navigate, focus }: PageProps) {
{selected ? ( - navigate("work", { kind: "task", id: taskId })} /> + navigate("work", { kind: "task", id: taskId })} /> ) : (
{t("Select a run")} @@ -118,12 +118,19 @@ function RunDetail({ run, tasks, agents, providers, providerRequests, actions, o 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))); + // Stop a running session, cancel one not yet started, or start a new session that picks up from the journal. + const [controlError, setControlError] = useState(null); + const [busy, setBusy] = useState(false); + const control = (operation: () => Promise) => { + setControlError(null); + setBusy(true); + operation() + .catch((error: unknown) => setControlError(error instanceof Error ? error.message : String(error))) + .finally(() => setBusy(false)); }; + const stop = () => control(() => actions.run(() => window.agentCompany.stopRun({ runId: run.id }))); + const cancel = () => control(() => actions.run(() => window.agentCompany.cancelQueuedRun({ orgId: run.orgId, runId: run.id }))); + const again = () => control(() => actions.run(() => window.agentCompany.runTask({ orgId: run.orgId, taskId: run.taskId }))); return (
@@ -136,9 +143,16 @@ function RunDetail({ run, tasks, agents, providers, providerRequests, actions, o
{run.id}
{runLabel(run.state)} - {run.state === "running" && } + {run.state === "running" && } + {run.state === "queued" && } + {run.state === "failed" && } + {(run.state === "stopped" || run.state === "superseded") && }
- {stopError &&
{stopError}
} + {run.state === "queued" &&
{t("A cancelled session can start again on the next heartbeat; block the task to keep it stopped.")}
} + {(run.state === "failed" || run.state === "stopped" || run.state === "superseded") && ( +
{t("A new session on the task picks up from what earlier sessions did and where they stopped.")}
+ )} + {controlError &&
{controlError}
}
{t("Agent")}
{agent?.name ?? "—"} ({agent?.role ?? "—"})
diff --git a/src/shared/desktop-api.ts b/src/shared/desktop-api.ts index f7ee6b7e..1d6da67d 100644 --- a/src/shared/desktop-api.ts +++ b/src/shared/desktop-api.ts @@ -205,6 +205,8 @@ export interface DesktopApi { setTaskBlocked(input: SetTaskBlockedInput): Promise>; /** What deleting a task would remove, keep and close. */ taskDeleteImpact(input: { orgId: string; taskId: string }): Promise>; + /** Cancels a run that has not started yet. */ + cancelQueuedRun(input: { orgId: string; runId: string }): Promise>; chatSpaceHistory(input: { orgId: string; spaceId: string; beforeMessageId: string | null } & ({ limit: number } | { fromMessageId: string })): Promise>; revokeConnector(input: RevokeConnectorInput): Promise>; search(input: SearchInput): Promise>; diff --git a/src/shared/ipc.ts b/src/shared/ipc.ts index e8dd3ef5..13989bd9 100644 --- a/src/shared/ipc.ts +++ b/src/shared/ipc.ts @@ -85,6 +85,7 @@ export const IPC_CHANNELS = { chatSpaceHistory: "agentcompany:chat-space-history", setTaskBlocked: "agentcompany:set-task-blocked", taskDeleteImpact: "agentcompany:task-delete-impact", + cancelQueuedRun: "agentcompany:cancel-queued-run", revokeConnector: "agentcompany:revoke-connector", search: "agentcompany:search", rendererEvent: "agentcompany:renderer-event", diff --git a/tests/e2e/ceo-chat.spec.ts b/tests/e2e/ceo-chat.spec.ts index 7ea9ea49..5730eb1a 100644 --- a/tests/e2e/ceo-chat.spec.ts +++ b/tests/e2e/ceo-chat.spec.ts @@ -82,6 +82,9 @@ test("the pinned CEO conversation is the Messages front door", async () => { await expect(page.getByText("Goal: Ship the FlowPilot MVP.", { exact: true })).toBeVisible(); await page.getByRole("button", { name: /Operationalize goal: Ship the FlowPilot MVP\./ }).first().click(); await taskDialog.getByRole("button", { name: /^Open run/ }).first().click(); + // A queued run can be cancelled before it starts; then it offers to resume. + await page.getByRole("button", { name: "Cancel run" }).click(); + await expect(page.getByRole("button", { name: "Resume" })).toBeVisible(); await expect(page.getByRole("button", { name: "Open task Operationalize goal: Ship the FlowPilot MVP." })).toBeVisible(); await page.getByRole("button", { name: "Open task Operationalize goal: Ship the FlowPilot MVP." }).click(); await expect(taskDialog).toBeVisible(); diff --git a/tests/integration/run-cancel.test.ts b/tests/integration/run-cancel.test.ts new file mode 100644 index 00000000..1130a042 --- /dev/null +++ b/tests/integration/run-cancel.test.ts @@ -0,0 +1,45 @@ +import { randomUUID } from "node:crypto"; +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 { ChatService } from "../../src/main/services/chat-service"; +import { cancelQueuedRun } from "../../src/main/services/run-cancel"; + +const testVault = { + encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), + decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"), +}; + +describe("cancelling a queued run", () => { + test("stops it before it starts without spending the retry budget, and closes a chat reply it was for", () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-run-cancel-${randomUUID()}.db`)); + const providers = new ProviderService(connection.db, providerAdapters, testVault); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + 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; + }); + const kickoff = connection.db.prepare("select id from task_runs where state = 'queued'").get() as { id: string }; + connection.transaction(() => cancelQueuedRun(connection.db, orgId, kickoff.id)); + expect(connection.db.prepare("select state, error_message from task_runs where id = ?").get(kickoff.id)) + .toEqual({ state: "stopped", error_message: "Cancelled before it started." }); + expect(() => cancelQueuedRun(connection.db, orgId, kickoff.id)).toThrow("That run is no longer queued."); + + const employeeId = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id; + const dm = (connection.db.prepare("select id from chat_spaces where org_id = ? and space_type = 'direct'").get(orgId) as { id: string }).id; + connection.transaction(() => new ChatService(connection.db).sendMessage({ orgId, spaceId: dm, threadId: null, senderType: "employee", senderId: employeeId, bodyMd: "Status?", attachmentFileIds: [] })); + const wake = connection.db.prepare("select id, task_id from task_runs where trigger = 'chat_mention'").get() as { id: string; task_id: string }; + connection.transaction(() => cancelQueuedRun(connection.db, orgId, wake.id)); + expect(connection.db.prepare("select status from tasks where id = ?").get(wake.task_id)).toEqual({ status: "done" }); + connection.close(); + }); +});