File size: 3,906 Bytes
f778c12 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | /** Cleanup helpers for failed ACP spawn flows. */
import {
callInProcessGatewayTool,
getInProcessGatewayToolContext,
runWithGatewayToolCleanupContext,
} from "../../agents/tools/in-process-gateway.js";
import { resolveSessionStorePathCore } from "../../config/sessions/paths.js";
import { loadSessionEntry } from "../../config/sessions/session-accessor.js";
import type { SessionEntry } from "../../config/sessions/types.js";
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import { logVerbose } from "../../globals.js";
import { runExclusiveSessionLifecycleMutation } from "../../sessions/session-lifecycle-admission.js";
import { isAcpOwnerRepairRequired } from "./manager.runtime-owner.js";
/** Roll back only the provisional session owned by this failed spawn. */
export async function cleanupFailedAcpSpawn(params: {
cfg: OpenClawConfig;
sessionKey: string;
agentId: string;
sessionEntry?: SessionEntry;
deleteTranscript: boolean;
closeRuntimeOnFailure?: () => Promise<void>;
}): Promise<void> {
if (!params.sessionEntry) {
return;
}
const { sessionId, lifecycleRevision } = params.sessionEntry;
const storePath = resolveSessionStorePathCore(params.cfg.session?.store, {
agentId: params.agentId,
});
const assertCurrent = () => {
const current = loadSessionEntry({
storePath,
sessionKey: params.sessionKey,
agentId: params.agentId,
clone: false,
});
if (current?.sessionId !== sessionId || current?.lifecycleRevision !== lifecycleRevision) {
throw new Error(`ACP provisional session ${params.sessionKey} changed before cleanup.`);
}
};
let deletionStarted = false;
const cancellation = new AbortController();
try {
await runWithGatewayToolCleanupContext(async () => {
assertCurrent();
const context = getInProcessGatewayToolContext();
if (!context) {
throw new Error("ACP provisional cleanup Gateway is unavailable.");
}
await callInProcessGatewayTool(
"sessions.delete",
{
key: params.sessionKey,
agentId: params.agentId,
expectedSessionId: sessionId,
...(lifecycleRevision ? { expectedLifecycleRevision: lifecycleRevision } : {}),
deleteTranscript: params.deleteTranscript,
emitLifecycleHooks: false,
},
// New /acp rows can have no revision yet. The private guard preserves
// exact absence as well as recorded revisions throughout deletion.
{
sessionMutationCommitGuard: () => {
cancellation.signal.throwIfAborted();
context.requestEntryLifetime?.signal.throwIfAborted();
assertCurrent();
// The router calls this after lazy preparation, before handler entry.
deletionStarted = true;
},
signal: cancellation.signal,
timeoutMs: 10_000,
},
);
});
} catch (error) {
// A retired owner cannot dispatch cleanup. Retain its existing local handle
// release, but never repeat a close after canonical deletion has started.
if (!deletionStarted) {
// A timed-out preparation may still settle; fence its late handler entry.
cancellation.abort(error);
}
if (!deletionStarted && params.closeRuntimeOnFailure) {
await runExclusiveSessionLifecycleMutation({
scope: storePath,
identities: [params.sessionKey, sessionId],
run: async () => {
assertCurrent();
await params.closeRuntimeOnFailure!();
},
}).catch((releaseError: unknown) => {
if (isAcpOwnerRepairRequired(releaseError)) {
throw releaseError;
}
logVerbose(`acp-spawn: provisional runtime cleanup failed: ${String(releaseError)}`);
});
}
logVerbose(`acp-spawn: provisional session cleanup failed: ${String(error)}`);
}
}
|