From d72333a04cf6e923e8307a9bb3f049c2e69dbc74 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:02:46 -0400 Subject: [PATCH] Q56: disconnecting a provider stops the sessions using it Co-Authored-By: Claude Opus 5.5 --- src/main/services/composio/composio-client.ts | 6 ++--- src/main/services/provider-service.ts | 5 +++++ src/main/services/run-control.ts | 22 +++++++++++++------ src/main/services/task-execution-service.ts | 15 +++++++------ tests/integration/run-stop.test.ts | 15 +++++++++++++ 5 files changed, 46 insertions(+), 17 deletions(-) diff --git a/src/main/services/composio/composio-client.ts b/src/main/services/composio/composio-client.ts index 00656efc..47e6abb3 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 } from "../run-control"; +import { currentRunSignal, RUN_STOPPED_MESSAGE, runStopReason } 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(RUN_STOPPED_MESSAGE)); + reject(new Error(runStopReason() ?? RUN_STOPPED_MESSAGE)); return; } runSignal?.addEventListener("abort", () => { request.abort(); - reject(new Error(RUN_STOPPED_MESSAGE)); + reject(new Error(runStopReason() ?? RUN_STOPPED_MESSAGE)); }, { once: true }); request.setHeader("x-api-key", this.apiKey); request.setHeader("Accept", "application/json"); diff --git a/src/main/services/provider-service.ts b/src/main/services/provider-service.ts index d5b1ae61..5882d45f 100644 --- a/src/main/services/provider-service.ts +++ b/src/main/services/provider-service.ts @@ -17,6 +17,7 @@ import { ProviderRequestStore } from "./provider-request-store"; import { ProviderResponseRunner, type ProviderResponseCreateInput } from "./provider-response-runner"; import { launchProviderSignInWindow, type LaunchProviderSignInWindowInput } from "./provider-sign-in-window"; import type { ToolTurnFn } from "./provider-tool-turn"; +import { stopRun } from "./run-control"; import { AUTH_STATE_CONFIGURED, AUTH_STATE_REVOKED, @@ -157,6 +158,10 @@ export class ProviderService { set credential_id = null, enabled = 0, auth_state = ?, health_state = ?, last_error = null, last_checked_at = null where id = ? and org_id = ? `).run(AUTH_STATE_REVOKED, HEALTH_STATE_UNCHECKED, input.providerId, input.orgId); + // Sessions mid-turn on this provider stop now instead of finishing on a revoked account. + const label = (this.db.prepare("select label from providers where id = ?").get(input.providerId) as { label: string }).label; + const running = this.db.prepare("select id from task_runs where provider_id = ? and state = 'running'").all(input.providerId) as Array<{ id: string }>; + for (const run of running) stopRun(run.id, `${label} was disconnected.`); this.audit.record({ orgId: input.orgId, entityType: "provider", diff --git a/src/main/services/run-control.ts b/src/main/services/run-control.ts index 20686724..af14cd6a 100644 --- a/src/main/services/run-control.ts +++ b/src/main/services/run-control.ts @@ -17,12 +17,12 @@ export async function withRunControl(runId: string, operation: () => Promise< } } -/** Stops one in-flight run; false when it is not running in this process. */ -export function stopRun(runId: string): boolean { +/** 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); if (!controller) return false; - console.info(`[run-control] stopping run ${runId}`); - controller.abort(); + console.info(`[run-control] stopping run ${runId}: ${reason}`); + controller.abort(new Error(reason)); return true; } @@ -37,17 +37,25 @@ export function currentRunSignal(): AbortSignal | undefined { return runSignals.getStore(); } +/** 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.reason instanceof Error ? signal.reason.message : RUN_STOPPED_MESSAGE; +} + export function throwIfRunStopped(): void { - if (currentRunSignal()?.aborted) throw new Error(RUN_STOPPED_MESSAGE); + const reason = runStopReason(); + if (reason) throw new Error(reason); } /** 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)); + if (signal.aborted) return Promise.reject(new Error(runStopReason() ?? RUN_STOPPED_MESSAGE)); return new Promise((resolve, reject) => { - const onAbort = () => reject(new Error(RUN_STOPPED_MESSAGE)); + const onAbort = () => reject(new Error(runStopReason() ?? 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 c75b529a..0d9497d8 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -13,7 +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 { runStopReason, withRunControl } from "./run-control"; import type { AgentToolContext } from "./agent-tools"; import { buildAgentInstructions, @@ -283,9 +283,10 @@ export class TaskExecutionService { } 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; + const stopReason = runStopReason(); + if (stopReason) { + this.markStopped(orgId, runId, providerRequestId, stopReason, { taskId: task.id, agentId: agent.id, providerId: provider.id }); + throw new Error(stopReason); } this.failRun(orgId, runId, providerRequestId, message); if (isTransientProviderFailure(message)) this.cooldowns.park(provider, message); @@ -423,13 +424,13 @@ export class TaskExecutionService { } /** 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 { + private markStopped(orgId: string, runId: string, providerRequestId: string | null, reason: string, 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); + `).run(TASK_RUN_STATE.STOPPED, providerRequestId, new Date().toISOString(), reason, runId, orgId); this.closeChatWakeTask(orgId, runId); - this.audit.record({ orgId, entityType: "task_run", entityId: runId, actorType: "employee", actorId: null, action: "stopped", payload }); + this.audit.record({ orgId, entityType: "task_run", entityId: runId, actorType: "system", actorId: null, action: "stopped", payload: { ...payload, reason } }); } /** Closes a queued run that no longer matches its task; it is not a failure and spends no retry budget. */ diff --git a/tests/integration/run-stop.test.ts b/tests/integration/run-stop.test.ts index 44e31038..393bb4da 100644 --- a/tests/integration/run-stop.test.ts +++ b/tests/integration/run-stop.test.ts @@ -93,6 +93,21 @@ describe("stopping runs", () => { failing.connection.close(); }); + test("disconnecting a provider stops its running sessions, even when the turn ends with a final answer", async () => { + let revoke = () => {}; + const { connection, execution, orgId, claim } = company(async () => { + revoke(); + return { text: "All done.", toolCalls: [], rawResponseJson: "{}" }; + }); + const runId = claim(); + const providerId = (connection.db.prepare("select provider_id from task_runs where id = ?").get(runId) as { provider_id: string }).provider_id; + const providers = new ProviderService(connection.db, providerAdapters, testVault); + revoke = () => providers.revokeCredential({ orgId, providerId }); + await expect(execution.executeRun(orgId, runId)).rejects.toThrow("OpenAI was disconnected."); + expect(connection.db.prepare("select state, error_message from task_runs where id = ?").get(runId)).toEqual({ state: "stopped", error_message: "OpenAI was disconnected." }); + 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));