diff --git a/src/agents/agent-command.live-model-switch.test.ts b/src/agents/agent-command.live-model-switch.test.ts index 44f677c4e843..394c93f69b92 100644 --- a/src/agents/agent-command.live-model-switch.test.ts +++ b/src/agents/agent-command.live-model-switch.test.ts @@ -328,7 +328,9 @@ vi.mock("../infra/agent-events.js", () => ({ clearAgentRunContext: (...args: unknown[]) => state.clearAgentRunContextMock(...args), emitAgentEvent: (...args: unknown[]) => state.emitAgentEventMock(...args), getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", onAgentEvent: vi.fn(), + registerAgentEventLifecycleRotationHandler: vi.fn(), registerAgentRunContext: (...args: unknown[]) => state.registerAgentRunContextMock(...args), withAgentRunLifecycleGeneration: (_generation: string, run: () => unknown) => run(), })); diff --git a/src/agents/embedded-agent-runner/run-state.ts b/src/agents/embedded-agent-runner/run-state.ts index 3a3e0f9793ea..0c401c7d8f59 100644 --- a/src/agents/embedded-agent-runner/run-state.ts +++ b/src/agents/embedded-agent-runner/run-state.ts @@ -12,6 +12,10 @@ import { resolveActiveReplyRunSessionId, type ReplyBackendQueueMessageOptions, } from "../../auto-reply/reply/reply-run-registry.js"; +import { + isAgentEventLifecycleGenerationCurrent, + registerAgentEventLifecycleRotationHandler, +} from "../../infra/agent-events.js"; import { resolveGlobalSingleton } from "../../shared/global-singleton.js"; /** @@ -63,6 +67,7 @@ const EMBEDDED_RUN_STATE_KEY = Symbol.for("openclaw.embeddedRunState"); const embeddedRunState = resolveGlobalSingleton(EMBEDDED_RUN_STATE_KEY, () => ({ activeRuns: new Map(), activeRunsByRunId: new Map(), + activeRunLifecycleGenerations: new WeakMap(), retainedAbortabilityRunIds: new Set(), snapshots: new Map(), sessionIdsByKey: new Map(), @@ -79,6 +84,12 @@ export const ACTIVE_EMBEDDED_RUNS = export const ACTIVE_EMBEDDED_RUNS_BY_RUN_ID = embeddedRunState.activeRunsByRunId ?? (embeddedRunState.activeRunsByRunId = new Map()); +export const ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS = + embeddedRunState.activeRunLifecycleGenerations ?? + (embeddedRunState.activeRunLifecycleGenerations = new WeakMap< + EmbeddedAgentQueueHandle, + string + >()); export const RETAINED_EMBEDDED_RUN_ABORTABILITY_RUN_IDS = embeddedRunState.retainedAbortabilityRunIds ?? (embeddedRunState.retainedAbortabilityRunIds = new Set()); @@ -104,6 +115,71 @@ export const EMBEDDED_RUN_WAITERS = embeddedRunState.waiters ?? (embeddedRunState.waiters = new Map>()); +function evictPriorLifecycleEmbeddedRuns(): void { + const staleHandles = new Set(); + for (const [sessionId, handle] of ACTIVE_EMBEDDED_RUNS) { + const lifecycleGeneration = ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.get(handle); + if (lifecycleGeneration && isAgentEventLifecycleGenerationCurrent(lifecycleGeneration)) { + continue; + } + staleHandles.add(handle); + if (ACTIVE_EMBEDDED_RUNS.get(sessionId) === handle) { + ACTIVE_EMBEDDED_RUNS.delete(sessionId); + } + ACTIVE_EMBEDDED_RUN_SNAPSHOTS.delete(sessionId); + } + for (const [runId, handle] of ACTIVE_EMBEDDED_RUNS_BY_RUN_ID) { + const lifecycleGeneration = ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.get(handle); + if (lifecycleGeneration && isAgentEventLifecycleGenerationCurrent(lifecycleGeneration)) { + continue; + } + staleHandles.add(handle); + // This index only gates the separately owned chat abort controller; absence + // is abortable. Keeping it would let stale ownership influence new work. + if (ACTIVE_EMBEDDED_RUNS_BY_RUN_ID.get(runId) === handle) { + ACTIVE_EMBEDDED_RUNS_BY_RUN_ID.delete(runId); + RETAINED_EMBEDDED_RUN_ABORTABILITY_RUN_IDS.delete(runId); + } + } + for (const [sessionKey, sessionId] of ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_KEY) { + if (!ACTIVE_EMBEDDED_RUNS.has(sessionId)) { + ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_KEY.delete(sessionKey); + } + } + for (const [sessionFile, sessionId] of ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_FILE) { + if (!ACTIVE_EMBEDDED_RUNS.has(sessionId)) { + ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_FILE.delete(sessionFile); + } + } + for (const [sessionId, waiters] of EMBEDDED_RUN_WAITERS) { + if (ACTIVE_EMBEDDED_RUNS.has(sessionId)) { + continue; + } + EMBEDDED_RUN_WAITERS.delete(sessionId); + for (const waiter of waiters) { + if (waiter.timer) { + clearTimeout(waiter.timer); + } + waiter.resolve(true); + } + } + const abortErrors: unknown[] = []; + // Remove stale ownership first so synchronous abort callbacks may register a + // replacement without the cleanup above erasing that current-generation run. + for (const handle of staleHandles) { + try { + handle.abort("restart"); + } catch (error) { + abortErrors.push(error); + } + } + if (abortErrors.length > 0) { + throw new AggregateError(abortErrors, "Failed to abort stale embedded agent runs"); + } +} + +registerAgentEventLifecycleRotationHandler("embedded-agent-runs", evictPriorLifecycleEmbeddedRuns); + /** Counts active embedded runs while including auto-reply registry runs for shared sessions. */ export function getActiveEmbeddedRunCount(): number { let activeCount = ACTIVE_EMBEDDED_RUNS.size; @@ -137,6 +213,20 @@ export function listActiveEmbeddedRunSessionIds(): string[] { ].toSorted((a, b) => a.localeCompare(b)); } +export function setActiveEmbeddedRunLifecycleGeneration( + handle: EmbeddedAgentQueueHandle, + lifecycleGeneration: string, +): string { + // A delayed re-registration must not transfer an old driver into the new + // Gateway lifecycle and suppress orphan recovery again. + const existingLifecycleGeneration = ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.get(handle); + if (existingLifecycleGeneration !== undefined) { + return existingLifecycleGeneration; + } + ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.set(handle, lifecycleGeneration); + return lifecycleGeneration; +} + /** Resolves the current session id for an active run after resets or compaction. */ export function resolveActiveEmbeddedRunSessionId(sessionKey: string): string | undefined { const normalizedSessionKey = sessionKey.trim(); diff --git a/src/agents/embedded-agent-runner/run/attempt-after-turn.ts b/src/agents/embedded-agent-runner/run/attempt-after-turn.ts index 1add41067c8f..3e3cb858c367 100644 --- a/src/agents/embedded-agent-runner/run/attempt-after-turn.ts +++ b/src/agents/embedded-agent-runner/run/attempt-after-turn.ts @@ -3,6 +3,7 @@ */ import { OPENCLAW_EMBEDDED_CONTEXT_ENGINE_HOST } from "../../../context-engine/host-compat.js"; import type { ContextEngine } from "../../../context-engine/types.js"; +import { captureAgentRunLifecycleGeneration } from "../../../infra/agent-events.js"; import { freezeDiagnosticTraceContext } from "../../../infra/diagnostic-trace-context.js"; import { formatErrorMessage } from "../../../infra/errors.js"; import type { getGlobalHookRunner } from "../../../plugins/hook-runner-global.js"; @@ -185,7 +186,11 @@ export async function completeEmbeddedAttemptAfterTurn( if (rotation.rotated) { sessionIdUsed = rotation.sessionId ?? sessionIdUsed; sessionFileUsed = rotation.sessionFile ?? sessionFileUsed; - updateActiveEmbeddedRunSessionFile(attempt.sessionId, sessionFileUsed); + updateActiveEmbeddedRunSessionFile( + attempt.sessionId, + sessionFileUsed, + attempt.lifecycleGeneration ?? captureAgentRunLifecycleGeneration(attempt.runId), + ); log.info( `[compaction] rotated active transcript after automatic compaction ` + `(sessionKey=${attempt.sessionKey ?? attempt.sessionId})`, diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts b/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts index 9c11cc855f80..e77bfea5b256 100644 --- a/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts +++ b/src/agents/embedded-agent-runner/run/attempt-stream-prepare.ts @@ -3,6 +3,7 @@ */ import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../../../auto-reply/tokens.js"; +import { captureAgentRunLifecycleGeneration } from "../../../infra/agent-events.js"; import { freezeDiagnosticTraceContext, type DiagnosticTraceContext, @@ -30,6 +31,7 @@ import { type ToolSearchTargetTranscriptProjection, } from "../../tool-search.js"; import { log } from "../logger.js"; +import { setActiveEmbeddedRunLifecycleGeneration } from "../run-state.js"; import { clearActiveEmbeddedRun, type EmbeddedAgentQueueHandle, @@ -374,6 +376,10 @@ export function prepareEmbeddedAttemptStream(input: { abort: (reason) => abortActiveRunExternally(reason), }; attempt.replyOperation?.attachBackend(queueHandle); + setActiveEmbeddedRunLifecycleGeneration( + queueHandle, + attempt.lifecycleGeneration ?? captureAgentRunLifecycleGeneration(attempt.runId), + ); setActiveEmbeddedRun(attempt.sessionId, queueHandle, attempt.sessionKey, attempt.sessionFile); return { diff --git a/src/agents/embedded-agent-runner/runs.generation.test.ts b/src/agents/embedded-agent-runner/runs.generation.test.ts new file mode 100644 index 000000000000..53bb410829d8 --- /dev/null +++ b/src/agents/embedded-agent-runner/runs.generation.test.ts @@ -0,0 +1,305 @@ +import { importFreshModule } from "openclaw/plugin-sdk/test-fixtures"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { testing as replyRunTesting } from "../../auto-reply/reply/reply-run-registry.test-support.js"; +import { + getAgentEventLifecycleGeneration, + rotateAgentEventLifecycleGeneration, +} from "../../infra/agent-events.js"; +import { + listActiveEmbeddedRunSessionIds, + listActiveEmbeddedRunSessionKeys, + setActiveEmbeddedRunLifecycleGeneration, + type EmbeddedAgentQueueHandle, +} from "./run-state.js"; +import { + clearActiveEmbeddedRun, + isEmbeddedAgentRunAbortableForRunId, + queueEmbeddedAgentMessageWithOutcomeAsync, + resolveActiveEmbeddedRunHandleSessionId, + resolveActiveEmbeddedRunHandleSessionIdBySessionFile, + setActiveEmbeddedRun, + updateActiveEmbeddedRunSessionFile, +} from "./runs.js"; +import { testing } from "./runs.test-support.js"; + +const lifecycleMock = vi.hoisted(() => { + let generationSequence = 0; + let generation = `test-generation-${generationSequence}`; + const handlers = new Map void>(); + return { + get: () => generation, + isCurrent: (candidate: string) => candidate === generation, + register: (key: string, handler: (nextGeneration: string) => void) => { + handlers.set(key, handler); + }, + reset: () => { + generationSequence += 1; + generation = `test-generation-${generationSequence}`; + }, + rotate: () => { + generationSequence += 1; + generation = `test-generation-${generationSequence}`; + const errors: unknown[] = []; + for (const handler of handlers.values()) { + try { + handler(generation); + } catch (error) { + errors.push(error); + } + } + if (errors.length > 0) { + throw new AggregateError(errors, "Failed to retire stale agent lifecycle owners"); + } + return generation; + }, + }; +}); + +vi.mock("../../infra/agent-events.js", async (importOriginal) => ({ + ...(await importOriginal()), + getAgentEventLifecycleGeneration: lifecycleMock.get, + isAgentEventLifecycleGenerationCurrent: lifecycleMock.isCurrent, + registerAgentEventLifecycleRotationHandler: lifecycleMock.register, + rotateAgentEventLifecycleGeneration: lifecycleMock.rotate, +})); + +function createRunHandle(params: { + abort?: EmbeddedAgentQueueHandle["abort"]; + queueMessage: EmbeddedAgentQueueHandle["queueMessage"]; + runId: string; +}): EmbeddedAgentQueueHandle { + return { + kind: "embedded", + runId: params.runId, + queueMessage: params.queueMessage, + isStreaming: () => true, + isAbortable: () => false, + isCompacting: () => false, + abort: params.abort ?? (() => {}), + }; +} + +describe("embedded run registry lifecycle generations", () => { + afterEach(() => { + testing.resetActiveEmbeddedRuns(); + replyRunTesting.resetReplyRunRegistry(); + lifecycleMock.reset(); + }); + + it("rejects a delayed prior-lifecycle registration for a current session owner", async () => { + const priorLifecycleGeneration = getAgentEventLifecycleGeneration(); + const staleQueueMessage = vi.fn(async () => {}); + const staleAbort = vi.fn(); + const staleHandle = createRunHandle({ + abort: staleAbort, + queueMessage: staleQueueMessage, + runId: "stale-run", + }); + setActiveEmbeddedRunLifecycleGeneration(staleHandle, priorLifecycleGeneration); + + rotateAgentEventLifecycleGeneration(); + const currentQueueMessage = vi.fn(async () => {}); + const currentAbort = vi.fn(); + setActiveEmbeddedRun( + "shared-session", + createRunHandle({ + abort: currentAbort, + queueMessage: currentQueueMessage, + runId: "current-run", + }), + "agent:main:current", + "/tmp/current-session.jsonl", + ); + + setActiveEmbeddedRun( + "shared-session", + staleHandle, + "agent:main:stale", + "/tmp/stale-session.jsonl", + ); + updateActiveEmbeddedRunSessionFile( + "shared-session", + "/tmp/stale-update.jsonl", + priorLifecycleGeneration, + ); + + await expect( + queueEmbeddedAgentMessageWithOutcomeAsync("shared-session", "still live"), + ).resolves.toMatchObject({ queued: true, target: "embedded_run" }); + expect(currentQueueMessage).toHaveBeenCalledOnce(); + expect(staleQueueMessage).not.toHaveBeenCalled(); + expect(staleAbort).toHaveBeenCalledWith("restart"); + expect(currentAbort).not.toHaveBeenCalled(); + expect(listActiveEmbeddedRunSessionIds()).toContain("shared-session"); + expect(listActiveEmbeddedRunSessionKeys()).toEqual(["agent:main:current"]); + expect(resolveActiveEmbeddedRunHandleSessionId("agent:main:stale")).toBeUndefined(); + expect( + resolveActiveEmbeddedRunHandleSessionIdBySessionFile("/tmp/stale-update.jsonl"), + ).toBeUndefined(); + expect(isEmbeddedAgentRunAbortableForRunId("current-run")).toBe(false); + expect(isEmbeddedAgentRunAbortableForRunId("stale-run")).toBe(true); + }); + + it("rejects a delayed prior-lifecycle registration without a replacement owner", async () => { + const priorLifecycleGeneration = getAgentEventLifecycleGeneration(); + const staleQueueMessage = vi.fn(async () => {}); + const staleAbort = vi.fn(); + const staleHandle = createRunHandle({ + abort: staleAbort, + queueMessage: staleQueueMessage, + runId: "stale-run", + }); + setActiveEmbeddedRunLifecycleGeneration(staleHandle, priorLifecycleGeneration); + + rotateAgentEventLifecycleGeneration(); + setActiveEmbeddedRun( + "stale-session", + staleHandle, + "agent:main:stale", + "/tmp/stale-session.jsonl", + ); + + await expect( + queueEmbeddedAgentMessageWithOutcomeAsync("stale-session", "should not arrive"), + ).resolves.toMatchObject({ queued: false, reason: "no_active_run" }); + expect(staleQueueMessage).not.toHaveBeenCalled(); + expect(staleAbort).toHaveBeenCalledWith("restart"); + expect(listActiveEmbeddedRunSessionIds()).not.toContain("stale-session"); + expect(listActiveEmbeddedRunSessionKeys()).not.toContain("agent:main:stale"); + expect(resolveActiveEmbeddedRunHandleSessionId("agent:main:stale")).toBeUndefined(); + expect( + resolveActiveEmbeddedRunHandleSessionIdBySessionFile("/tmp/stale-session.jsonl"), + ).toBeUndefined(); + expect(isEmbeddedAgentRunAbortableForRunId("stale-run")).toBe(true); + }); + + it("propagates failure to abort a delayed prior-lifecycle registration", () => { + const priorLifecycleGeneration = getAgentEventLifecycleGeneration(); + const staleHandle = createRunHandle({ + abort: () => { + throw new Error("stale abort failed"); + }, + queueMessage: vi.fn(async () => {}), + runId: "stale-run", + }); + setActiveEmbeddedRunLifecycleGeneration(staleHandle, priorLifecycleGeneration); + + rotateAgentEventLifecycleGeneration(); + + expect(() => setActiveEmbeddedRun("stale-session", staleHandle, "agent:main:stale")).toThrow( + "stale abort failed", + ); + expect(listActiveEmbeddedRunSessionIds()).not.toContain("stale-session"); + }); + + it("lets a current-lifecycle owner replace a stale session owner", async () => { + const staleQueueMessage = vi.fn(async () => {}); + const staleAbort = vi.fn(); + const staleHandle = createRunHandle({ + abort: staleAbort, + queueMessage: staleQueueMessage, + runId: "stale-run", + }); + setActiveEmbeddedRun( + "shared-session", + staleHandle, + "agent:main:stale", + "/tmp/stale-session.jsonl", + ); + + rotateAgentEventLifecycleGeneration(); + const currentQueueMessage = vi.fn(async () => {}); + const currentAbort = vi.fn(); + setActiveEmbeddedRun( + "shared-session", + createRunHandle({ + abort: currentAbort, + queueMessage: currentQueueMessage, + runId: "current-run", + }), + "agent:main:current", + "/tmp/current-session.jsonl", + ); + clearActiveEmbeddedRun("shared-session", staleHandle, "agent:main:stale"); + + await expect( + queueEmbeddedAgentMessageWithOutcomeAsync("shared-session", "now current"), + ).resolves.toMatchObject({ queued: true, target: "embedded_run" }); + expect(currentQueueMessage).toHaveBeenCalledOnce(); + expect(staleQueueMessage).not.toHaveBeenCalled(); + expect(staleAbort).toHaveBeenCalledOnce(); + expect(staleAbort).toHaveBeenCalledWith("restart"); + expect(currentAbort).not.toHaveBeenCalled(); + expect(listActiveEmbeddedRunSessionIds()).toContain("shared-session"); + expect(listActiveEmbeddedRunSessionKeys()).toEqual(["agent:main:current"]); + expect(resolveActiveEmbeddedRunHandleSessionId("agent:main:stale")).toBeUndefined(); + expect( + resolveActiveEmbeddedRunHandleSessionIdBySessionFile("/tmp/stale-session.jsonl"), + ).toBeUndefined(); + }); + + it("preserves a current owner registered synchronously by stale abort", async () => { + const currentQueueMessage = vi.fn(async () => {}); + const currentAbort = vi.fn(); + const currentHandle = createRunHandle({ + abort: currentAbort, + queueMessage: currentQueueMessage, + runId: "current-run", + }); + const staleAbort = vi.fn(() => { + setActiveEmbeddedRun( + "shared-session", + currentHandle, + "agent:main:current", + "/tmp/current-session.jsonl", + ); + }); + const staleHandle = createRunHandle({ + abort: staleAbort, + queueMessage: vi.fn(async () => {}), + runId: "stale-run", + }); + setActiveEmbeddedRun( + "shared-session", + staleHandle, + "agent:main:stale", + "/tmp/stale-session.jsonl", + ); + + rotateAgentEventLifecycleGeneration(); + + await expect( + queueEmbeddedAgentMessageWithOutcomeAsync("shared-session", "current survives"), + ).resolves.toMatchObject({ queued: true, target: "embedded_run" }); + expect(staleAbort).toHaveBeenCalledWith("restart"); + expect(currentAbort).not.toHaveBeenCalled(); + expect(currentQueueMessage).toHaveBeenCalledOnce(); + expect(listActiveEmbeddedRunSessionKeys()).toEqual(["agent:main:current"]); + }); + + it("evicts reply operations created by a prior hot-loaded module instance", async () => { + const replyRunsA = await importFreshModule< + typeof import("../../auto-reply/reply/reply-run-registry.js") + >(import.meta.url, "../../auto-reply/reply/reply-run-registry.js?scope=generation-a"); + const operation = replyRunsA.createReplyOperation({ + sessionKey: "agent:main:hot-loaded", + sessionId: "hot-loaded-session", + resetTriggered: false, + }); + const cancel = vi.fn(); + operation.setPhase("running"); + operation.attachBackend({ + kind: "embedded", + cancel, + isStreaming: () => true, + }); + + const replyRunsB = await importFreshModule< + typeof import("../../auto-reply/reply/reply-run-registry.js") + >(import.meta.url, "../../auto-reply/reply/reply-run-registry.js?scope=generation-b"); + rotateAgentEventLifecycleGeneration(); + + expect(cancel).toHaveBeenCalledWith("restart"); + expect(replyRunsB.isReplyRunActiveForSessionId("hot-loaded-session")).toBe(false); + }); +}); diff --git a/src/agents/embedded-agent-runner/runs.test.ts b/src/agents/embedded-agent-runner/runs.test.ts index fe4ef109dfdc..8c3e27857ada 100644 --- a/src/agents/embedded-agent-runner/runs.test.ts +++ b/src/agents/embedded-agent-runner/runs.test.ts @@ -10,6 +10,7 @@ import { isReplyRunActiveForSessionId, } from "../../auto-reply/reply/reply-run-registry.js"; import { testing as replyRunTesting } from "../../auto-reply/reply/reply-run-registry.test-support.js"; +import { getAgentEventLifecycleGeneration } from "../../infra/agent-events.js"; import { setDiagnosticsEnabledForProcess } from "../../infra/diagnostic-events.js"; import { resetDiagnosticRunActivityForTest } from "../../logging/diagnostic-run-activity.js"; import { markDiagnosticToolStartedForTest } from "../../logging/diagnostic-run-activity.test-support.js"; @@ -437,6 +438,7 @@ describe("embedded-agent runner run registry", () => { updateActiveEmbeddedRunSessionFile( "session-file-diagnostics", "/tmp/openclaw-run-registry-rotated.jsonl", + getAgentEventLifecycleGeneration(), ); expect(getDiagnosticSessionState({ sessionId: "session-file-diagnostics" }).sessionFile).toBe( diff --git a/src/agents/embedded-agent-runner/runs.ts b/src/agents/embedded-agent-runner/runs.ts index 76ce5e82c51f..989957e6b73e 100644 --- a/src/agents/embedded-agent-runner/runs.ts +++ b/src/agents/embedded-agent-runner/runs.ts @@ -17,6 +17,10 @@ import { type ReplyOperationPhase, waitForReplyRunEndBySessionId, } from "../../auto-reply/reply/reply-run-registry.js"; +import { + getAgentEventLifecycleGeneration, + isAgentEventLifecycleGenerationCurrent, +} from "../../infra/agent-events.js"; import { getDiagnosticSessionActivitySnapshot, markDiagnosticEmbeddedRunEnded, @@ -33,6 +37,7 @@ import { resolveTimerTimeoutMs } from "../../shared/number-coercion.js"; import { ACTIVE_EMBEDDED_RUNS, ACTIVE_EMBEDDED_RUNS_BY_RUN_ID, + ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS, ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_FILE, ACTIVE_EMBEDDED_RUN_SESSION_IDS_BY_KEY, ACTIVE_EMBEDDED_RUN_SNAPSHOTS, @@ -42,6 +47,7 @@ import { EMBEDDED_RUN_WAITERS, getActiveEmbeddedRunCount, RETAINED_EMBEDDED_RUN_ABORTABILITY_RUN_IDS, + setActiveEmbeddedRunLifecycleGeneration, type ActiveEmbeddedRunSnapshot, type AbandonedEmbeddedRun, type EmbeddedAgentQueueHandle, @@ -835,6 +841,22 @@ export function setActiveEmbeddedRun( sessionKey?: string, sessionFile?: string, ) { + const currentLifecycleGeneration = getAgentEventLifecycleGeneration(); + const incomingLifecycleGeneration = setActiveEmbeddedRunLifecycleGeneration( + handle, + currentLifecycleGeneration, + ); + // The immutable handle generation rejects delayed stale registration even + // when rotation left no replacement owner in the session slot. + if (!isAgentEventLifecycleGenerationCurrent(incomingLifecycleGeneration)) { + try { + handle.abort("restart"); + } catch (error) { + diag.warn(`stale run registration abort failed: sessionId=${sessionId} err=${String(error)}`); + throw error; + } + return; + } const previousHandle = ACTIVE_EMBEDDED_RUNS.get(sessionId); const wasActive = previousHandle !== undefined; if (previousHandle) { @@ -845,6 +867,7 @@ export function setActiveEmbeddedRun( if (handle.runId) { ACTIVE_EMBEDDED_RUNS_BY_RUN_ID.set(handle.runId, handle); } + clearActiveRunSessionKeys(sessionId); setActiveRunSessionKey(sessionKey, sessionId); clearActiveRunSessionFiles(sessionId); setActiveRunSessionFile(sessionFile, sessionId); @@ -874,8 +897,14 @@ export function updateActiveEmbeddedRunSnapshot( export function updateActiveEmbeddedRunSessionFile( sessionId: string, sessionFile: string | undefined, + lifecycleGeneration: string, ): void { - if (!ACTIVE_EMBEDDED_RUNS.has(sessionId)) { + const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId); + if ( + handle === undefined || + ACTIVE_EMBEDDED_RUN_LIFECYCLE_GENERATIONS.get(handle) !== lifecycleGeneration || + !isAgentEventLifecycleGenerationCurrent(lifecycleGeneration) + ) { return; } clearActiveRunSessionFiles(sessionId); diff --git a/src/agents/embedded-agent-subscribe.handlers.lifecycle.test.ts b/src/agents/embedded-agent-subscribe.handlers.lifecycle.test.ts index 0c15048dc834..b39d598a5b75 100644 --- a/src/agents/embedded-agent-subscribe.handlers.lifecycle.test.ts +++ b/src/agents/embedded-agent-subscribe.handlers.lifecycle.test.ts @@ -26,6 +26,9 @@ const BEFORE_AGENT_FINALIZE_EVENT = { vi.mock("../infra/agent-events.js", () => ({ emitAgentEvent: emitAgentEventMock, + getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", + registerAgentEventLifecycleRotationHandler: vi.fn(), })); function createContext( diff --git a/src/agents/main-session-restart-recovery.test.ts b/src/agents/main-session-restart-recovery.test.ts index 8495dd7039d9..db48867117aa 100644 --- a/src/agents/main-session-restart-recovery.test.ts +++ b/src/agents/main-session-restart-recovery.test.ts @@ -4,6 +4,7 @@ import os from "node:os"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { GatewayClientRequestError } from "../../packages/gateway-client/src/index.js"; +import { createReplyOperation } from "../auto-reply/reply/reply-run-registry.js"; import type { InternalSessionEntry as SessionEntry } from "../config/sessions.js"; import * as sessionAccessor from "../config/sessions/session-accessor.js"; import { @@ -19,6 +20,7 @@ import { getAgentEventLifecycleGeneration, registerAgentRunContext, resetAgentEventsForTest, + rotateAgentEventLifecycleGeneration, } from "../infra/agent-events.js"; import { getActiveGatewayRootWorkCount, @@ -32,6 +34,14 @@ import { runExclusiveSessionLifecycleMutation, } from "../sessions/session-lifecycle-admission.js"; import { createDeferred } from "../test-utils/deferred.js"; +import { setActiveEmbeddedRunLifecycleGeneration } from "./embedded-agent-runner/run-state.js"; +import { + clearActiveEmbeddedRun, + queueEmbeddedAgentMessageWithOutcomeAsync, + resolveActiveEmbeddedRunHandleSessionId, + setActiveEmbeddedRun, + type EmbeddedAgentQueueHandle, +} from "./embedded-agent-runner/runs.js"; import { INTERNAL_RUNTIME_CONTEXT_BEGIN, INTERNAL_RUNTIME_CONTEXT_END, @@ -2035,6 +2045,179 @@ describe("main-session-restart-recovery", () => { expect(store["agent:main:already-marked"]?.abortedLastRun).toBe(false); }); + it.each([ + ["current owner before delayed stale registration", "current-first"], + ["stale owner before current registration", "stale-first"], + ] as const)("keeps a live session running with %s", async (_label, registrationOrder) => { + const sessionsDir = await makeSessionsDir(); + const cutoff = Date.now(); + const sessionKey = "agent:main:generation-race"; + const sessionId = "generation-race-session"; + await writeStore(sessionsDir, { + [sessionKey]: { + sessionId, + updatedAt: cutoff - 10_000, + status: "running", + }, + }); + + const createHandle = (runId: string): EmbeddedAgentQueueHandle => ({ + kind: "embedded", + runId, + queueMessage: async () => {}, + isStreaming: () => true, + isCompacting: () => false, + abort: () => {}, + }); + const priorLifecycleGeneration = getAgentEventLifecycleGeneration(); + const staleHandle = createHandle("stale-generation-run"); + setActiveEmbeddedRunLifecycleGeneration(staleHandle, priorLifecycleGeneration); + if (registrationOrder === "stale-first") { + setActiveEmbeddedRun(sessionId, staleHandle, sessionKey); + } + + rotateAgentEventLifecycleGeneration(); + const currentHandle = createHandle("current-generation-run"); + setActiveEmbeddedRun(sessionId, currentHandle, sessionKey); + if (registrationOrder === "current-first") { + setActiveEmbeddedRun(sessionId, staleHandle, sessionKey); + } + + try { + await expect( + recoverStartupOrphanedMainSessions({ stateDir: tmpDir, updatedBeforeMs: cutoff }), + ).resolves.toEqual({ marked: 0, recovered: 0, failed: 0, skipped: 0 }); + expect(callGateway).not.toHaveBeenCalled(); + expect( + loadSessionEntry({ + sessionKey, + storePath: path.join(sessionsDir, "sessions.json"), + }), + ).toMatchObject({ status: "running" }); + } finally { + clearActiveEmbeddedRun(sessionId, currentHandle, sessionKey); + clearActiveEmbeddedRun(sessionId, staleHandle, sessionKey); + } + }); + + it("reconciles only prior-lifecycle running sessions after an in-process restart", async () => { + const sessionsDir = await makeSessionsDir(); + const cutoff = Date.now(); + const abandonedKey = "agent:main:abandoned-client"; + const liveKey = "agent:main:live-client"; + await writeStore(sessionsDir, { + [abandonedKey]: { + sessionId: "abandoned-session", + updatedAt: cutoff - 10_000, + status: "running", + }, + [liveKey]: { + sessionId: "live-session", + updatedAt: cutoff - 10_000, + status: "running", + }, + }); + await writeTranscript(sessionsDir, "abandoned-session", [ + { role: "system", content: "the client disappeared before the turn became resumable" }, + ]); + + const createHandle = ( + runId: string, + queueMessage: EmbeddedAgentQueueHandle["queueMessage"] = async () => {}, + abort: EmbeddedAgentQueueHandle["abort"] = () => {}, + ): EmbeddedAgentQueueHandle => ({ + kind: "embedded", + runId, + queueMessage, + isStreaming: () => true, + isCompacting: () => false, + abort, + }); + const abandonedReply = createReplyOperation({ + sessionKey: abandonedKey, + sessionId: "abandoned-session", + resetTriggered: false, + }); + const abandonedReplyQueue = vi.fn(async () => {}); + const abandonedReplyCancel = vi.fn(); + abandonedReply.setPhase("running"); + abandonedReply.attachBackend({ + kind: "embedded", + cancel: abandonedReplyCancel, + isStreaming: () => true, + queueMessage: abandonedReplyQueue, + }); + const abandonedEmbeddedQueue = vi.fn(async () => {}); + const abandonedEmbeddedAbort = vi.fn(); + const abandonedHandle = createHandle( + "abandoned-run", + abandonedEmbeddedQueue, + abandonedEmbeddedAbort, + ); + setActiveEmbeddedRun("abandoned-session", abandonedHandle, abandonedKey); + + const firstRecovery = recoverStartupOrphanedMainSessions({ + stateDir: tmpDir, + updatedBeforeMs: cutoff, + }); + // Advance ownership while the async store discovery above is pending. The + // older scan must drop the stale owner without overlooking this new live one. + rotateAgentEventLifecycleGeneration(); + setActiveEmbeddedRun("abandoned-session", abandonedHandle, abandonedKey); + + await expect( + queueEmbeddedAgentMessageWithOutcomeAsync("abandoned-session", "do not route stale"), + ).resolves.toMatchObject({ queued: false, reason: "no_active_run" }); + expect(abandonedEmbeddedQueue).not.toHaveBeenCalled(); + expect(abandonedEmbeddedAbort).toHaveBeenCalledWith("restart"); + expect(abandonedReplyQueue).not.toHaveBeenCalled(); + expect(abandonedReplyCancel).toHaveBeenCalledWith("restart"); + expect(resolveActiveEmbeddedRunHandleSessionId(abandonedKey)).toBeUndefined(); + + const liveReply = createReplyOperation({ + sessionKey: liveKey, + sessionId: "live-session", + resetTriggered: false, + }); + const liveAbort = vi.fn(); + const liveHandle = createHandle("live-run", undefined, liveAbort); + setActiveEmbeddedRun("live-session", liveHandle, liveKey); + try { + const first = await firstRecovery; + + expect(first).toEqual({ marked: 1, recovered: 0, failed: 1, skipped: 0 }); + expect(callGateway).not.toHaveBeenCalled(); + expect( + loadSessionEntry({ + sessionKey: abandonedKey, + storePath: path.join(sessionsDir, "sessions.json"), + }), + ).toMatchObject({ + status: "failed", + abortedLastRun: true, + }); + expect( + loadSessionEntry({ + sessionKey: liveKey, + storePath: path.join(sessionsDir, "sessions.json"), + }), + ).toMatchObject({ + status: "running", + }); + expect(liveAbort).not.toHaveBeenCalled(); + expect(liveReply.abortSignal.aborted).toBe(false); + + await expect( + recoverStartupOrphanedMainSessions({ stateDir: tmpDir, updatedBeforeMs: cutoff }), + ).resolves.toEqual({ marked: 0, recovered: 0, failed: 0, skipped: 0 }); + } finally { + clearActiveEmbeddedRun("abandoned-session", abandonedHandle, abandonedKey); + clearActiveEmbeddedRun("live-session", liveHandle, liveKey); + abandonedReply.complete(); + liveReply.complete(); + } + }); + it("recovers only the configured store for duplicate startup-orphaned session keys", async () => { const cutoff = Date.now(); const defaultSessionsDir = await makeSessionsDir(); diff --git a/src/agents/main-session-restart-recovery.ts b/src/agents/main-session-restart-recovery.ts index 9fb26a57386d..79253731626e 100644 --- a/src/agents/main-session-restart-recovery.ts +++ b/src/agents/main-session-restart-recovery.ts @@ -364,6 +364,9 @@ export async function markStartupOrphanedMainSessionsForRecovery(params: { ? undefined : normalizeStringSet(params.activeSessionKeys); const updatedBeforeMs = normalizeFiniteTimestamp(params.updatedBeforeMs); + // Lifecycle rotation synchronously evicts stale owners, so this same registry + // view drives both operational routing and recovery suppression. Re-read it at + // each check so a newer owner can still fence an older async recovery scan. const resolveActiveSessionIds = () => providedActiveSessionIds ?? normalizeStringSet(listActiveEmbeddedRunSessionIds()); const resolveActiveSessionKeys = () => diff --git a/src/agents/subagent-registry.announce-loop-guard.test.ts b/src/agents/subagent-registry.announce-loop-guard.test.ts index 0be460dca1f1..d53fbf0a3def 100644 --- a/src/agents/subagent-registry.announce-loop-guard.test.ts +++ b/src/agents/subagent-registry.announce-loop-guard.test.ts @@ -54,7 +54,10 @@ vi.mock("../gateway/call.js", () => ({ })); vi.mock("../infra/agent-events.js", () => ({ + getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", onAgentEvent: mocks.onAgentEvent, + registerAgentEventLifecycleRotationHandler: vi.fn(), })); vi.mock("./subagent-registry.store.sqlite.js", () => ({ diff --git a/src/agents/subagent-registry.archive.e2e.test.ts b/src/agents/subagent-registry.archive.e2e.test.ts index e41fb7cdef3e..5fa6b9aa8038 100644 --- a/src/agents/subagent-registry.archive.e2e.test.ts +++ b/src/agents/subagent-registry.archive.e2e.test.ts @@ -57,8 +57,11 @@ vi.mock("../tasks/task-status-access.js", () => ({ })); vi.mock("../infra/agent-events.js", () => ({ + getAgentEventLifecycleGeneration: () => "test-generation", getAgentRunContext: vi.fn(() => undefined), + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", onAgentEvent: vi.fn((_handler: unknown) => noop), + registerAgentEventLifecycleRotationHandler: vi.fn(), })); vi.mock("../config/config.js", async () => { diff --git a/src/agents/subagent-registry.mocks.shared.ts b/src/agents/subagent-registry.mocks.shared.ts index 5f9f35668966..a3133a29cbb6 100644 --- a/src/agents/subagent-registry.mocks.shared.ts +++ b/src/agents/subagent-registry.mocks.shared.ts @@ -21,5 +21,8 @@ vi.mock("../gateway/call.js", () => ({ })); vi.mock("../infra/agent-events.js", () => ({ + getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", onAgentEvent: sharedMocks.onAgentEvent, + registerAgentEventLifecycleRotationHandler: vi.fn(), })); diff --git a/src/agents/subagent-registry.steer-restart.test.ts b/src/agents/subagent-registry.steer-restart.test.ts index 6e4cf07d23fd..eb113b1377fa 100644 --- a/src/agents/subagent-registry.steer-restart.test.ts +++ b/src/agents/subagent-registry.steer-restart.test.ts @@ -50,6 +50,9 @@ vi.mock("../gateway/call.js", () => ({ })); vi.mock("../infra/agent-events.js", () => ({ + getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", + registerAgentEventLifecycleRotationHandler: vi.fn(), onAgentEvent: vi.fn((handler: typeof lifecycleHandler) => { lifecycleHandler = handler; return noop; diff --git a/src/agents/subagent-registry.test.ts b/src/agents/subagent-registry.test.ts index 47ea99ae9f5e..f1f345277c4d 100644 --- a/src/agents/subagent-registry.test.ts +++ b/src/agents/subagent-registry.test.ts @@ -211,8 +211,11 @@ vi.mock("../gateway/call.js", () => ({ })); vi.mock("../infra/agent-events.js", () => ({ + getAgentEventLifecycleGeneration: () => "test-generation", getAgentRunContext: mocks.getAgentRunContext, + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", onAgentEvent: mocks.onAgentEvent, + registerAgentEventLifecycleRotationHandler: vi.fn(), })); vi.mock("../config/config.js", () => { diff --git a/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts b/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts index 067b7e31b5dd..f790cefb5cab 100644 --- a/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts +++ b/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts @@ -537,7 +537,10 @@ vi.mock("../../bindings/records.js", () => ({ vi.mock("../../infra/agent-events.js", () => ({ emitAgentAuditEvent: (params: unknown) => agentEventMocks.emitAgentAuditEvent(params), emitAgentEvent: (params: unknown) => agentEventMocks.emitAgentEvent(params), + getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", onAgentEvent: (listener: unknown) => agentEventMocks.onAgentEvent(listener), + registerAgentEventLifecycleRotationHandler: vi.fn(), })); vi.mock("../../plugins/conversation-binding.js", () => ({ buildPluginBindingDeclinedText: () => "Plugin binding request was declined.", diff --git a/src/auto-reply/reply/reply-run-registry.ts b/src/auto-reply/reply/reply-run-registry.ts index 87d5ef39bcd7..280ba7e757ce 100644 --- a/src/auto-reply/reply/reply-run-registry.ts +++ b/src/auto-reply/reply/reply-run-registry.ts @@ -5,6 +5,11 @@ import { isAgentRunRestartAbortReason, } from "../../agents/run-termination.js"; import { createAbortError } from "../../infra/abort-signal.js"; +import { + getAgentEventLifecycleGeneration, + isAgentEventLifecycleGenerationCurrent, + registerAgentEventLifecycleRotationHandler, +} from "../../infra/agent-events.js"; import type { ImageContent } from "../../llm/types.js"; import { getDiagnosticSessionActivitySnapshot, @@ -123,6 +128,8 @@ type ReplyOperationResult = export type ReplyOperation = { readonly key: ReplyRunKey; readonly sessionId: string; + /** Gateway lifecycle that admitted this process-local owner. */ + readonly lifecycleGeneration?: string; readonly routeThreadId?: string | number; readonly abortSignal: AbortSignal; readonly resetTriggered: boolean; @@ -234,6 +241,7 @@ type ReplyRunState = { waitKeysBySessionId: Map; waitersByKey: Map>; followupAdmissionBarriersByKey: Map; + evictOperationByOperation?: WeakMap void>; }; const REPLY_RUN_STATE_KEY = Symbol.for("openclaw.replyRunRegistry"); @@ -245,8 +253,12 @@ const replyRunState = resolveGlobalSingleton(REPLY_RUN_STATE_KEY, waitKeysBySessionId: new Map(), waitersByKey: new Map>(), followupAdmissionBarriersByKey: new Map(), + evictOperationByOperation: new WeakMap void>(), })); replyRunState.followupAdmissionBarriersByKey ??= new Map(); +const evictReplyOperationByOperation = + replyRunState.evictOperationByOperation ?? + (replyRunState.evictOperationByOperation = new WeakMap void>()); export const REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS = 15_000; // Terminal results must release the lane even if the owner never resumes. @@ -490,7 +502,20 @@ function updateFollowupAdmissionSessionId(sessionKey: string, sessionId: string) } } -function clearReplyRunState(params: { sessionKey: string; sessionId: string }): void { +function clearReplyRunState(params: { + sessionKey: string; + sessionId: string; + operation: ReplyOperation; +}): void { + if (replyRunState.activeRunsByKey.get(params.sessionKey) !== params.operation) { + if ( + replyRunState.activeKeysBySessionId.get(params.sessionId) === params.sessionKey && + replyRunState.activeSessionIdsByKey.get(params.sessionKey) !== params.sessionId + ) { + replyRunState.activeKeysBySessionId.delete(params.sessionId); + } + return; + } replyRunState.activeRunsByKey.delete(params.sessionKey); replyRunState.activeSessionIdsByKey.delete(params.sessionKey); if (replyRunState.activeKeysBySessionId.get(params.sessionId) === params.sessionKey) { @@ -550,6 +575,7 @@ export function createReplyOperation(params: { let terminalRecovery = false; let acceptedSteeredInboundAudio = false; const startedAtMs = Date.now(); + const lifecycleGeneration = getAgentEventLifecycleGeneration(); let lastActivityAtMs = startedAtMs; const upstreamAbortSignal = params.upstreamAbortSignal; let upstreamAbortHandler: (() => void) | undefined; @@ -580,6 +606,7 @@ export function createReplyOperation(params: { terminalSettleTimer.clear(); finalizationLease.clear(); expireReplyOperationByOperation.delete(operation); + evictReplyOperationByOperation.delete(operation); detachUpstreamAbort(); const registeredBarrier = afterClearBarrier ? registerFollowupAdmissionBarrier( @@ -598,6 +625,7 @@ export function createReplyOperation(params: { clearReplyRunState({ sessionKey: currentSessionKey, sessionId: currentSessionId, + operation, }); if (!registeredBarrier) { flushReplyOperationAfterClear(operation, currentSessionId); @@ -642,6 +670,7 @@ export function createReplyOperation(params: { get sessionId() { return currentSessionId; }, + lifecycleGeneration, get routeThreadId() { return params.routeThreadId; }, @@ -939,6 +968,33 @@ export function createReplyOperation(params: { }, }); + evictReplyOperationByOperation.set(operation, () => { + if (stateCleared) { + return; + } + if (!result) { + setResult({ kind: "aborted", code: "aborted_for_restart" }); + phase = "aborted"; + } + abortInternally(createAgentRunRestartAbortError()); + let cancelError: unknown; + let cancelFailed = false; + try { + getAttachedBackend(operation)?.cancel("restart"); + } catch (error) { + cancelFailed = true; + cancelError = error; + diag.warn( + `reply run lifecycle eviction cancel failed: sessionKey=${currentSessionKey} error=${String(error)}`, + ); + } finally { + clearState(); + } + if (cancelFailed) { + throw cancelError; + } + }); + replyRunState.activeRunsByKey.set(sessionKey, operation); replyRunState.activeSessionIdsByKey.set(sessionKey, currentSessionId); replyRunState.activeKeysBySessionId.set(currentSessionId, sessionKey); @@ -1295,6 +1351,67 @@ export function listActiveReplyRunSessionKeys(): string[] { return [...replyRunState.activeSessionIdsByKey.keys()]; } +function evictPriorLifecycleReplyRuns(): void { + const errors: unknown[] = []; + for (const operation of replyRunState.activeRunsByKey.values()) { + if ( + operation.lifecycleGeneration && + isAgentEventLifecycleGenerationCurrent(operation.lifecycleGeneration) + ) { + continue; + } + const evict = evictReplyOperationByOperation.get(operation); + if (evict) { + try { + evict(); + } catch (error) { + errors.push(error); + try { + clearReplyRunState({ + sessionKey: operation.key, + sessionId: operation.sessionId, + operation, + }); + } catch (clearError) { + errors.push(clearError); + } + } + continue; + } + // Pre-generation hot-loaded operations have no retained callback, but their + // public method still closes over the module instance that owns the backend. + try { + if (!operation.abortForRestart()) { + errors.push(new Error(`Stale reply operation was not abortable: ${operation.key}`)); + } + } catch (error) { + errors.push(error); + } + // Admission stays occupied until the old closure clears it. If abort + // synchronously clears and replaces the slot, its captured stateCleared + // makes this completion idempotent instead of erasing the replacement. + try { + operation.complete(); + } catch (error) { + errors.push(error); + } + try { + clearReplyRunState({ + sessionKey: operation.key, + sessionId: operation.sessionId, + operation, + }); + } catch (error) { + errors.push(error); + } + } + if (errors.length > 0) { + throw new AggregateError(errors, "Failed to abort stale reply runs"); + } +} + +registerAgentEventLifecycleRotationHandler("reply-runs", evictPriorLifecycleReplyRuns); + const replyRunRegistryTestApi = { resetReplyRunRegistry(): void { for (const [sessionKey, sessionId] of replyRunState.activeSessionIdsByKey) { diff --git a/src/cli/gateway-cli/run-loop.test.ts b/src/cli/gateway-cli/run-loop.test.ts index ddb5106ff17f..9f6a0e2376ce 100644 --- a/src/cli/gateway-cli/run-loop.test.ts +++ b/src/cli/gateway-cli/run-loop.test.ts @@ -211,6 +211,9 @@ vi.mock("../../tasks/runtime-internal.js", () => ({ })); vi.mock("../../infra/agent-events.js", () => ({ + getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", + registerAgentEventLifecycleRotationHandler: vi.fn(), rotateAgentEventLifecycleGeneration: () => rotateAgentEventLifecycleGeneration(), })); diff --git a/src/commands/agent.acp.test.ts b/src/commands/agent.acp.test.ts index 793ac9f2a363..53cfb9a20878 100644 --- a/src/commands/agent.acp.test.ts +++ b/src/commands/agent.acp.test.ts @@ -26,10 +26,14 @@ const agentEventMocks = vi.hoisted(() => { } }), getAgentEventLifecycleGeneration: vi.fn(() => "test-generation"), + isAgentEventLifecycleGenerationCurrent: vi.fn( + (generation: string) => generation === "test-generation", + ), onAgentEvent: vi.fn((handler: (event: AgentEvent) => void) => { handlers.add(handler); return () => handlers.delete(handler); }), + registerAgentEventLifecycleRotationHandler: vi.fn(), registerAgentRunContext: vi.fn(), withAgentRunLifecycleGeneration: vi.fn((_generation: string, run: () => unknown) => run()), }; diff --git a/src/gateway/server-methods/agent.test-harness.ts b/src/gateway/server-methods/agent.test-harness.ts index 49fbb9ec1d52..c3b40910dd4e 100644 --- a/src/gateway/server-methods/agent.test-harness.ts +++ b/src/gateway/server-methods/agent.test-harness.ts @@ -221,6 +221,9 @@ vi.mock("../../infra/agent-events.js", () => ({ getAgentEventLifecycleGeneration: () => mocks.lifecycleGeneration, getAgentRunContext: vi.fn(() => undefined), hasProjectedAgentRunForSession: vi.fn(() => false), + isAgentEventLifecycleGenerationCurrent: (generation: string) => + generation === mocks.lifecycleGeneration, + registerAgentEventLifecycleRotationHandler: vi.fn(), registerAgentRunContext: mocks.registerAgentRunContext, onAgentEvent: vi.fn(), })); diff --git a/src/infra/agent-events.ts b/src/infra/agent-events.ts index d4d1757ca023..f9a68be8f941 100644 --- a/src/infra/agent-events.ts +++ b/src/infra/agent-events.ts @@ -167,6 +167,7 @@ type AgentEventState = { } >; lifecycleGeneration: string; + lifecycleRotationHandlers?: Map void>; }; const AGENT_EVENT_STATE_KEY = Symbol.for("openclaw.agentEvents.state"); @@ -223,9 +224,25 @@ export function getAgentEventLifecycleGeneration(): string { return getAgentEventState().lifecycleGeneration; } +export function isAgentEventLifecycleGenerationCurrent(lifecycleGeneration: string): boolean { + return lifecycleGeneration === getAgentEventState().lifecycleGeneration; +} + +/** Registers process-local state cleanup at the gateway lifecycle boundary. */ +export function registerAgentEventLifecycleRotationHandler( + key: string, + handler: (lifecycleGeneration: string) => void, +): void { + const state = getAgentEventState(); + const handlers = + state.lifecycleRotationHandlers ?? + (state.lifecycleRotationHandlers = new Map void>()); + handlers.set(key, handler); +} + /** Rejects work that no longer belongs to the active gateway lifecycle. */ export function assertAgentRunLifecycleGenerationCurrent(lifecycleGeneration: string): void { - if (lifecycleGeneration === getAgentEventState().lifecycleGeneration) { + if (isAgentEventLifecycleGenerationCurrent(lifecycleGeneration)) { return; } throw createAbortError("Agent run belongs to a stale gateway lifecycle"); @@ -244,6 +261,18 @@ export function captureAgentRunLifecycleGeneration(runId: string): string { export function rotateAgentEventLifecycleGeneration(): string { const state = getAgentEventState(); state.lifecycleGeneration = randomUUID(); + // Rotation is the liveness choke point: after it returns, no prior-generation + // owner is operationally reachable. Recovery and runtime consumers therefore + // agree that only current-generation owners can drive or receive work. + const errors: unknown[] = []; + notifyListeners( + state.lifecycleRotationHandlers?.values() ?? [], + state.lifecycleGeneration, + (error) => errors.push(error), + ); + if (errors.length > 0) { + throw new AggregateError(errors, "Failed to retire stale agent lifecycle owners"); + } return state.lifecycleGeneration; } diff --git a/src/plugins/wired-hooks-after-tool-call.e2e.test.ts b/src/plugins/wired-hooks-after-tool-call.e2e.test.ts index 5be8cf71a45d..a44a164932c6 100644 --- a/src/plugins/wired-hooks-after-tool-call.e2e.test.ts +++ b/src/plugins/wired-hooks-after-tool-call.e2e.test.ts @@ -22,6 +22,9 @@ vi.mock("../infra/agent-events.js", () => ({ emitAgentCommandOutputEvent: vi.fn(), emitAgentItemEvent: vi.fn(), emitAgentEvent: vi.fn(), + getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", + registerAgentEventLifecycleRotationHandler: vi.fn(), })); function createToolHandlerCtx(params: { diff --git a/src/plugins/wired-hooks-compaction.test.ts b/src/plugins/wired-hooks-compaction.test.ts index 0524bf6f5130..5fa8bfcd126a 100644 --- a/src/plugins/wired-hooks-compaction.test.ts +++ b/src/plugins/wired-hooks-compaction.test.ts @@ -19,6 +19,9 @@ vi.mock("../plugins/hook-runner-global.js", () => ({ vi.mock("../infra/agent-events.js", () => ({ emitAgentEvent: hookMocks.emitAgentEvent, + getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", + registerAgentEventLifecycleRotationHandler: vi.fn(), })); import { diff --git a/src/tasks/cron-run-continuation-cleanup.test.ts b/src/tasks/cron-run-continuation-cleanup.test.ts index a5dbc27c00b0..015abf5ec399 100644 --- a/src/tasks/cron-run-continuation-cleanup.test.ts +++ b/src/tasks/cron-run-continuation-cleanup.test.ts @@ -17,6 +17,9 @@ vi.mock("../config/sessions/session-accessor.js", () => ({ })); vi.mock("../infra/agent-events.js", () => ({ getAgentEventLifecycleGeneration: () => "current-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => + generation === "current-generation", + registerAgentEventLifecycleRotationHandler: vi.fn(), })); vi.mock("../infra/session-delivery-queue.js", () => ({ loadPendingSessionDeliveries: mocks.loadPendingSessionDeliveries, diff --git a/src/tui/embedded-backend.test.ts b/src/tui/embedded-backend.test.ts index c8cb761b2cbd..69f0bdd4503e 100644 --- a/src/tui/embedded-backend.test.ts +++ b/src/tui/embedded-backend.test.ts @@ -97,6 +97,9 @@ vi.mock("../agents/btw.js", () => ({ })); vi.mock("../infra/agent-events.js", () => ({ + getAgentEventLifecycleGeneration: () => "test-generation", + isAgentEventLifecycleGenerationCurrent: (generation: string) => generation === "test-generation", + registerAgentEventLifecycleRotationHandler: vi.fn(), onAgentEvent: (listener: (evt: unknown) => void) => { registeredListener = listener; return () => {