From 3bc0d811cb402db960e3e6a326197a84412b483e 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 10:09:16 -0400 Subject: [PATCH] Q57: quitting stops running sessions and waits for them to record their end Co-Authored-By: Claude Opus 5.5 --- src/main/index.ts | 18 ++++++++++-- src/main/services/composio/composio-client.ts | 6 ++-- src/main/services/run-control.ts | 27 ++++++++++++++--- tests/unit/run-control.test.ts | 29 +++++++++++++++++++ 4 files changed, 71 insertions(+), 9 deletions(-) create mode 100644 tests/unit/run-control.test.ts diff --git a/src/main/index.ts b/src/main/index.ts index c2294ec9..02398236 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -11,6 +11,9 @@ import { SkillService } from "./services/skill-service"; import { startProviderBackgroundWorker } from "./app/provider-background-worker"; import { recoverInterruptedWork } from "./app/startup-recovery"; import { startTaskQueueWorker } from "./app/task-queue-worker"; +import { stopAllRunsAndWait } from "./services/run-control"; + +const QUIT_GRACE_MS = 10_000; const __dirname = dirname(fileURLToPath(import.meta.url)); @@ -61,10 +64,21 @@ if (!app.requestSingleInstanceLock()) { win.focus(); }); - app.on("before-quit", () => { + let shutdown: "running" | "draining" | "drained" = "running"; + app.on("before-quit", (event) => { + if (shutdown === "drained") return; + // Hold every exit request until running sessions are stopped and have recorded how they ended. + event.preventDefault(); + if (shutdown === "draining") return; + shutdown = "draining"; stopProviderBackgroundWorker(); stopTaskQueueWorker(); - connection.close(); + void stopAllRunsAndWait("The app was closing.", QUIT_GRACE_MS).then((unsettled) => { + if (unsettled > 0) console.warn(`[quit] ${unsettled} run(s) did not settle within ${QUIT_GRACE_MS}ms; startup recovery will re-queue them.`); + connection.close(); + shutdown = "drained"; + app.quit(); + }); }); }); diff --git a/src/main/services/composio/composio-client.ts b/src/main/services/composio/composio-client.ts index 47e6abb3..9cc63871 100644 --- a/src/main/services/composio/composio-client.ts +++ b/src/main/services/composio/composio-client.ts @@ -1,5 +1,5 @@ import { net } from "electron"; -import { currentRunSignal, RUN_STOPPED_MESSAGE, runStopReason } from "../run-control"; +import { currentRunSignal, stopReasonOf } from "../run-control"; const BASE_URL = "https://backend.composio.dev/api/v3"; @@ -156,12 +156,12 @@ export class ComposioClient { const request = net.request({ method, url }); const runSignal = currentRunSignal(); if (runSignal?.aborted) { - reject(new Error(runStopReason() ?? RUN_STOPPED_MESSAGE)); + reject(new Error(stopReasonOf(runSignal))); return; } runSignal?.addEventListener("abort", () => { request.abort(); - reject(new Error(runStopReason() ?? RUN_STOPPED_MESSAGE)); + reject(new Error(stopReasonOf(runSignal))); }, { once: true }); request.setHeader("x-api-key", this.apiKey); request.setHeader("Accept", "application/json"); diff --git a/src/main/services/run-control.ts b/src/main/services/run-control.ts index af14cd6a..11a83240 100644 --- a/src/main/services/run-control.ts +++ b/src/main/services/run-control.ts @@ -5,18 +5,33 @@ export const RUN_STOPPED_MESSAGE = "Stopped by the operator."; const runSignals = new AsyncLocalStorage(); const controllers = new Map(); +const settling = 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); + const running = runSignals.run(controller.signal, operation); + settling.set(runId, running.catch(() => undefined)); try { - return await runSignals.run(controller.signal, operation); + return await running; } finally { controllers.delete(runId); + settling.delete(runId); } } +/** Stops every run and waits, up to the timeout, for each to record how it ended; returns how many were still unsettled. */ +export async function stopAllRunsAndWait(reason: string, timeoutMs: number): Promise { + const pending = [...settling.values()]; + for (const id of [...controllers.keys()]) stopRun(id, reason); + let timer: ReturnType | undefined; + const timedOut = new Promise<"timeout">((resolve) => { timer = setTimeout(() => resolve("timeout"), timeoutMs); }); + const outcome = await Promise.race([Promise.allSettled(pending).then(() => "settled" as const), timedOut]); + clearTimeout(timer); + return outcome === "settled" ? 0 : settling.size; +} + /** Stops one in-flight run with a reason; false when it is not running in this process. */ export function stopRun(runId: string, reason: string = RUN_STOPPED_MESSAGE): boolean { const controller = controllers.get(runId); @@ -40,7 +55,11 @@ export function currentRunSignal(): AbortSignal | undefined { /** Why the current run was stopped, or null while it may continue. */ export function runStopReason(): string | null { const signal = currentRunSignal(); - if (!signal?.aborted) return null; + return signal?.aborted ? stopReasonOf(signal) : null; +} + +/** The reason a stopped run's signal carries. */ +export function stopReasonOf(signal: AbortSignal): string { return signal.reason instanceof Error ? signal.reason.message : RUN_STOPPED_MESSAGE; } @@ -53,9 +72,9 @@ export function throwIfRunStopped(): void { export function untilStopped(operation: Promise): Promise { const signal = currentRunSignal(); if (!signal) return operation; - if (signal.aborted) return Promise.reject(new Error(runStopReason() ?? RUN_STOPPED_MESSAGE)); + if (signal.aborted) return Promise.reject(new Error(stopReasonOf(signal))); return new Promise((resolve, reject) => { - const onAbort = () => reject(new Error(runStopReason() ?? RUN_STOPPED_MESSAGE)); + const onAbort = () => reject(new Error(stopReasonOf(signal))); signal.addEventListener("abort", onAbort, { once: true }); operation.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort)); }); diff --git a/tests/unit/run-control.test.ts b/tests/unit/run-control.test.ts new file mode 100644 index 00000000..9bcfdc89 --- /dev/null +++ b/tests/unit/run-control.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, test } from "vitest"; +import { stopAllRunsAndWait, untilStopped, withRunControl } from "../../src/main/services/run-control"; + +describe("run control", () => { + test("shutdown stops every run and waits until each has recorded its ending", async () => { + const recorded: string[] = []; + const session = (id: string) => withRunControl(id, async () => { + try { + await untilStopped(new Promise(() => {})); + } catch (error) { + await new Promise((resolve) => setTimeout(resolve, 20)); + recorded.push(`${id}: ${(error as Error).message}`); + throw error; + } + }); + const runs = [session("run_a"), session("run_b")].map((run) => run.catch(() => undefined)); + expect(await stopAllRunsAndWait("The app was closing.", 2_000)).toBe(0); + expect(recorded.sort()).toEqual(["run_a: The app was closing.", "run_b: The app was closing."]); + await Promise.all(runs); + }); + + test("a run that ignores the stop is reported as unsettled after the grace period", async () => { + let release = () => {}; + const stubborn = withRunControl("run_stubborn", () => new Promise((resolve) => { release = resolve; })); + expect(await stopAllRunsAndWait("The app was closing.", 30)).toBe(1); + release(); + await stubborn; + }); +});