Q57: quitting stops running sessions and waits for them to record their end
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
d72333a04c
commit
3bc0d811cb
4 files changed
+71
-9
No files matched your search
@@ -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");
|
||||
|
||||
@@ -5,18 +5,33 @@ export const RUN_STOPPED_MESSAGE = "Stopped by the operator.";
|
||||
|
||||
const runSignals = new AsyncLocalStorage<AbortSignal>();
|
||||
const controllers = new Map<string, AbortController>();
|
||||
const settling = new Map<string, Promise<unknown>>();
|
||||
|
||||
/** 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);
|
||||
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<number> {
|
||||
const pending = [...settling.values()];
|
||||
for (const id of [...controllers.keys()]) stopRun(id, reason);
|
||||
let timer: ReturnType<typeof setTimeout> | 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<T>(operation: Promise<T>): Promise<T> {
|
||||
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<T>((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));
|
||||
});
|
||||
|
||||
Reference in new issue
Block a user