Q56: disconnecting a provider stops the sessions using it
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
9e1f8da11c
commit
d72333a04c
5 files changed
+46
-17
No files matched your search
@@ -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");
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -17,12 +17,12 @@ export async function withRunControl<T>(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<T>(operation: Promise<T>): Promise<T> {
|
||||
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<T>((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));
|
||||
});
|
||||
|
||||
@@ -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. */
|
||||
|
||||
@@ -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<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
|
||||
Reference in new issue
Block a user