diff --git a/src/auto-reply/reply/reply-turn-admission.test.ts b/src/auto-reply/reply/reply-turn-admission.test.ts index 3aeb1512e161..dba69f8464be 100644 --- a/src/auto-reply/reply/reply-turn-admission.test.ts +++ b/src/auto-reply/reply/reply-turn-admission.test.ts @@ -494,7 +494,7 @@ describe("reply turn admission", () => { }, ); - it("waits through deferred owner release retries beyond one settle slice", async () => { + it("keeps deferred owner release retries from retaining a successor", async () => { vi.useFakeTimers(); try { const sessionKey = "agent:main:telegram:topic:deferred-recovery-release"; @@ -524,15 +524,15 @@ describe("reply turn admission", () => { } const applySessionEntryReplacements = sessionAccessor.applySessionEntryReplacements; let failures = 0; - vi.spyOn(sessionAccessor, "applySessionEntryReplacements").mockImplementation( - async (params) => { - if (failures < 15) { + const accessorSpy = vi + .spyOn(sessionAccessor, "applySessionEntryReplacements") + .mockImplementation(async (params) => { + if (failures < 3) { failures += 1; - throw new Error("transient session-store failure"); + throw new Error("SQLite session entry changed before replacement"); } return await applySessionEntryReplacements(params); - }, - ); + }); owner.operation.complete(); const successor = admitTestReplyTurn({ @@ -545,10 +545,9 @@ describe("reply turn admission", () => { void successor.then(() => { successorSettled = true; }); - await vi.advanceTimersByTimeAsync(REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS + 1); - expect(successorSettled).toBe(false); - - await vi.advanceTimersByTimeAsync(20_000); + await vi.advanceTimersByTimeAsync(100); + expect(successorSettled).toBe(true); + accessorSpy.mockRestore(); const admitted = await successor; expect(admitted.status).toBe("owned"); if (admitted.status === "owned") { diff --git a/src/auto-reply/reply/reply-turn-admission.ts b/src/auto-reply/reply/reply-turn-admission.ts index ee6a28f428df..f7c48f43eb7e 100644 --- a/src/auto-reply/reply/reply-turn-admission.ts +++ b/src/auto-reply/reply/reply-turn-admission.ts @@ -62,19 +62,16 @@ async function releaseReplyRecoveryOwner( if (!lease) { return undefined; } - let settleDeferredRelease: ( - pending: MainSessionRecoveryPendingTarget | undefined, - ) => void = () => {}; - const deferredRelease = new Promise((resolve) => { - settleDeferredRelease = resolve; - }); try { return await releaseMainSessionRecoveryOwner(lease, { - onDeferredSuccess: settleDeferredRelease, + onDeferredSuccess: scheduleMainSessionRecoveryPendingTarget, }); } catch (error) { log.warn(`failed to release main-session recovery reply owner: ${formatErrorMessage(error)}`); - return await deferredRelease; + // The durable owner schedules exact-token retries. A completed reply must + // not keep its successor barrier and lifecycle admission until that + // background repair wins a contested SQLite write. + return undefined; } } diff --git a/src/config/sessions/session-accessor.sqlite-entry-store.ts b/src/config/sessions/session-accessor.sqlite-entry-store.ts index be5a75141c89..3d654ed311be 100644 --- a/src/config/sessions/session-accessor.sqlite-entry-store.ts +++ b/src/config/sessions/session-accessor.sqlite-entry-store.ts @@ -233,7 +233,7 @@ export function readExactSessionEntryRow( return entry ? { entry, legacyKeys: [], row } : undefined; } -export function readExactSessionEntryJsonForCanonicalRepair( +export function readExactSessionEntryJson( database: Pick, sessionKey: string, ): string | undefined { diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts index 3247a992542e..2e334a6df8d2 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts @@ -25,7 +25,7 @@ import { import { sqliteSessionEntriesEqual } from "./session-accessor.sqlite-entry-equality.js"; import { deleteSessionEntryRows, - readExactSessionEntryJsonForCanonicalRepair, + readExactSessionEntryJson, readExactSessionEntryRow, readSessionEntryStore, } from "./session-accessor.sqlite-entry-store.js"; @@ -54,7 +54,7 @@ export function shouldRemoveSessionEntry( } if ( removal.expectedEntry !== undefined && - JSON.stringify(entry) !== JSON.stringify(removal.expectedEntry) + !sqliteSessionEntriesEqual(entry, removal.expectedEntry) ) { return false; } @@ -328,7 +328,7 @@ export async function projectSessionEntryLifecycleMutation( const sessionKey = removal.exactStoredKey ? removal.sessionKey : removal.sessionKey.trim(); let entry = removal.exactStoredKey || sessionKey ? store[sessionKey] : undefined; if (removal.expectedRawEntryJson !== undefined) { - const currentRawEntryJson = readExactSessionEntryJsonForCanonicalRepair(database, sessionKey); + const currentRawEntryJson = readExactSessionEntryJson(database, sessionKey); if (currentRawEntryJson !== removal.expectedRawEntryJson) { throw new Error( `SQLite session entry changed before raw lifecycle removal for ${sessionKey}`, diff --git a/src/config/sessions/session-accessor.sqlite-projection.ts b/src/config/sessions/session-accessor.sqlite-projection.ts index d8c10b99e8e8..952c1ef53c9d 100644 --- a/src/config/sessions/session-accessor.sqlite-projection.ts +++ b/src/config/sessions/session-accessor.sqlite-projection.ts @@ -37,7 +37,7 @@ import { sqliteSessionEntriesEqual } from "./session-accessor.sqlite-entry-equal import { deleteLegacySessionEntryRows, deleteSessionEntryRows, - readExactSessionEntryJsonForCanonicalRepair, + readExactSessionEntryJson, readExactSessionEntryRow, readSessionEntryCount, readSessionEntryStore, @@ -210,10 +210,7 @@ function readProjectedRemovalEntry( : readExactSessionEntryRow(database, projected.sessionKey) )?.entry; } - if ( - readExactSessionEntryJsonForCanonicalRepair(database, projected.sessionKey) !== - expectedRawEntryJson - ) { + if (readExactSessionEntryJson(database, projected.sessionKey) !== expectedRawEntryJson) { throw new Error( `SQLite session entry changed before raw lifecycle removal for ${projected.sessionKey}`, ); diff --git a/src/config/sessions/session-accessor.sqlite-replacement-projection.test.ts b/src/config/sessions/session-accessor.sqlite-replacement-projection.test.ts new file mode 100644 index 000000000000..a9f88573c762 --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-replacement-projection.test.ts @@ -0,0 +1,60 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { cleanupTempDirs, makeTempDir } from "../../../test/helpers/temp-dir.js"; + +const { readExactSessionEntryJsonMock } = vi.hoisted(() => ({ + readExactSessionEntryJsonMock: vi.fn(), +})); + +vi.mock("./session-accessor.sqlite-entry-store.js", async () => { + const actual = await vi.importActual( + "./session-accessor.sqlite-entry-store.js", + ); + readExactSessionEntryJsonMock.mockImplementation(actual.readExactSessionEntryJson); + return { ...actual, readExactSessionEntryJson: readExactSessionEntryJsonMock }; +}); + +const { applySessionEntryReplacements, loadSessionEntry, upsertSessionEntryCore } = + await import("./session-accessor.js"); + +describe("session entry replacement compare-and-swap", () => { + const tempDirs: string[] = []; + let storePath: string; + + beforeEach(() => { + storePath = `${makeTempDir(tempDirs, "replacement-cas")}/openclaw-agent.sqlite`; + }); + + afterEach(() => { + readExactSessionEntryJsonMock.mockReset(); + cleanupTempDirs(tempDirs); + }); + + it("refuses to replace a selected row whose bytes disappear before the snapshot completes", async () => { + const scope = { sessionKey: "agent:main:vanishing-row", storePath }; + await upsertSessionEntryCore(scope, { + model: "base", + sessionId: "vanishing-row", + updatedAt: 10, + }); + // A concurrent writer can delete the row between hydrating the snapshot entry and reading + // its persisted bytes. Both the snapshot and the transaction then observe "no bytes", so a + // missing-vs-missing compare would agree and rewrite the stale entry into the deleted key. + readExactSessionEntryJsonMock.mockReturnValue(undefined); + + await expect( + applySessionEntryReplacements({ + sessionKeys: [scope.sessionKey], + storePath, + update: (entries) => ({ + replacements: entries.map(({ entry, sessionKey }) => ({ + entry: { ...entry, model: "resurrected" }, + sessionKey, + })), + result: undefined, + }), + }), + ).rejects.toThrow("changed before replacement"); + + expect(loadSessionEntry(scope)).toMatchObject({ model: "base", sessionId: "vanishing-row" }); + }); +}); diff --git a/src/config/sessions/session-accessor.sqlite-replacement-projection.ts b/src/config/sessions/session-accessor.sqlite-replacement-projection.ts index eaa27f0a0a56..f6ffa3fd6b40 100644 --- a/src/config/sessions/session-accessor.sqlite-replacement-projection.ts +++ b/src/config/sessions/session-accessor.sqlite-replacement-projection.ts @@ -11,6 +11,7 @@ import type { } from "./session-accessor.sqlite-contract.js"; import { deleteLegacySessionEntryRows, + readExactSessionEntryJson, readExactSessionEntryRow, readSessionEntryStore, writeSessionEntry, @@ -85,8 +86,21 @@ async function applySqliteSessionEntryReplacementProjection( const replacementAuthorityKeys = selectedStatuses ? new Set(entries.map(({ sessionKey }) => sessionKey)) : selectedKeys; + // Compare persisted row bytes, never a hydrated entry. Participants and owner live in + // their own table/columns, and each selection reader projects a different subset of them, + // so an entry-object compare can differ from the transaction re-read with no write at all + // and wedge the row's repairs forever. const expectedEntryJson = new Map( - entries.map(({ sessionKey, entry }) => [sessionKey, JSON.stringify(entry)]), + entries.map(({ sessionKey }) => { + const rawEntryJson = readExactSessionEntryJson(database, sessionKey); + if (rawEntryJson === undefined) { + // The row vanished between hydrating the snapshot and reading its bytes. Fail closed: + // a selected key must hold bytes, or a later missing-vs-missing compare would pass and + // rewrite the stale entry into a concurrently deleted key. + throw new Error(`SQLite session entry changed before replacement for ${sessionKey}`); + } + return [sessionKey, rawEntryJson]; + }), ); const operation = await params.update(entries); const replacements = normalize(operation.replacements); @@ -153,7 +167,10 @@ async function applySqliteSessionEntryReplacementProjection( const transactionEntries = new Map(); for (const sessionKey of validationKeys) { const transactionEntry = readExactSessionEntryRow(transactionDb, sessionKey)?.entry; - if (JSON.stringify(transactionEntry) !== expectedEntryJson.get(sessionKey)) { + if ( + readExactSessionEntryJson(transactionDb, sessionKey) !== + expectedEntryJson.get(sessionKey) + ) { throw new Error(`SQLite session entry changed before replacement for ${sessionKey}`); } if (transactionEntry) { diff --git a/src/config/sessions/session-accessor.test.ts b/src/config/sessions/session-accessor.test.ts index 301b851abcc0..af942e6846ab 100644 --- a/src/config/sessions/session-accessor.test.ts +++ b/src/config/sessions/session-accessor.test.ts @@ -76,6 +76,7 @@ import { } from "./session-accessor.sqlite-entry-store.js"; import { loadExactSessionEntry, replaceSessionEntrySync } from "./session-accessor.sqlite-entry.js"; import { importSqliteSessionRows } from "./session-accessor.sqlite-import.js"; +import { recordSessionParticipant } from "./session-accessor.sqlite-participants.js"; import { applySessionEntryCanonicalReplacements } from "./session-accessor.sqlite-replacement-projection.js"; import { appendTranscriptEventSync, @@ -2500,6 +2501,35 @@ describe("session accessor seam", () => { expect(loadSessionEntry(scope)).toMatchObject({ model: "newer", updatedAt: 20 }); }); + it("replaces a status-selected entry whose participants are projected only inside the transaction", async () => { + const scope = { sessionKey: "agent:main:participant-replacement", storePath }; + await upsertSessionEntryCore(scope, { + sessionId: "participant-replacement", + status: "running", + updatedAt: 10, + }); + // No owner or createdActor, so this participant survives owner filtering and the + // transaction-side read hydrates fields the status-selected snapshot never sees. + recordSessionParticipant(scope, { + actor: { id: "8167215807", type: "human" }, + source: "channel", + }); + + await applySessionEntryReplacements({ + statuses: ["running"], + storePath, + update: (entries) => ({ + replacements: entries.map(({ entry, sessionKey }) => ({ + entry: { ...entry, abortedLastRun: true }, + sessionKey, + })), + result: undefined, + }), + }); + + expect(loadSessionEntry(scope)).toMatchObject({ abortedLastRun: true }); + }); + it("awaits lifecycle builders outside transactions while keeping their commit indivisible", async () => { const scope = { sessionKey: "agent:main:lifecycle-prepare", storePath }; await upsertSessionEntryCore(scope, { diff --git a/src/logging/diagnostic-session-recovery.ts b/src/logging/diagnostic-session-recovery.ts index 0666b7132b0b..c0abb8f9ecab 100644 --- a/src/logging/diagnostic-session-recovery.ts +++ b/src/logging/diagnostic-session-recovery.ts @@ -14,8 +14,6 @@ type DiagnosticSessionRecoverySkipReason = | "missing_session_ref" | "stale_session_state"; -type DiagnosticSessionRecoveryNoopReason = "no_active_work"; - export type StuckSessionRecoveryRequest = { sessionId?: string; sessionKey?: string; @@ -67,7 +65,7 @@ export type StuckSessionRecoveryOutcome = | (DiagnosticSessionRecoveryBaseOutcome & { status: "released"; action: "release_lane"; - reason?: "stale_lane_task"; + reason?: "no_active_work" | "stale_lane_task"; released: number; queuedCount?: number; }) @@ -78,11 +76,6 @@ export type StuckSessionRecoveryOutcome = activeCount?: number; queuedCount?: number; }) - | (DiagnosticSessionRecoveryBaseOutcome & { - status: "noop"; - action: "none"; - reason: DiagnosticSessionRecoveryNoopReason; - }) | (DiagnosticSessionRecoveryBaseOutcome & { status: "failed"; action: "none"; @@ -96,11 +89,7 @@ export function recoveryOutcomeMutatesSessionState( if (!outcome) { return false; } - return ( - outcome.status === "aborted" || - outcome.status === "released" || - (outcome.status === "noop" && outcome.reason === "no_active_work") - ); + return outcome.status === "aborted" || outcome.status === "released"; } export function recoveryOutcomeClearsQueuedSessionState( @@ -108,8 +97,7 @@ export function recoveryOutcomeClearsQueuedSessionState( ): boolean { return ( outcome.status === "released" || - (outcome.status === "aborted" && outcome.released > 0 && (outcome.queuedCount ?? 0) === 0) || - (outcome.status === "noop" && outcome.reason === "no_active_work") + (outcome.status === "aborted" && outcome.released > 0 && (outcome.queuedCount ?? 0) === 0) ); } diff --git a/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts b/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts index d9cf513b14ad..5c36fe0284f8 100644 --- a/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts +++ b/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts @@ -743,7 +743,7 @@ describe("stuck session recovery", () => { ]); }); - it("reports when recovery finds no active work to release", async () => { + it("releases stale processing state when recovery finds no active work", async () => { mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue(undefined); mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(undefined); mocks.isEmbeddedAgentRunActive.mockReturnValue(false); @@ -757,10 +757,34 @@ describe("stuck session recovery", () => { expect(mocks.resetCommandLane).toHaveBeenCalledWith("session:agent:main:main"); expect(warnLogMessages()).toEqual([ - "stuck session recovery outcome: status=noop action=none sessionId=stale-session sessionKey=agent:main:main lane=session:agent:main:main reason=no_active_work", + "stuck session recovery: sessionId=stale-session sessionKey=agent:main:main age=180s action=release_lane aborted=false drained=true released=0", + "stuck session recovery outcome: status=released action=release_lane sessionId=stale-session sessionKey=agent:main:main lane=session:agent:main:main reason=no_active_work released=0", ]); }); + it("keeps observing an active run that neither aborted nor released", async () => { + mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue("active-session"); + mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue("active-session"); + mocks.isEmbeddedAgentRunActive.mockReturnValue(true); + mocks.abortEmbeddedAgentRun.mockReturnValue(false); + mocks.forceClearEmbeddedAgentRun.mockReturnValue(false); + mocks.resetCommandLane.mockReturnValue(0); + + const outcome = await recoverStuckDiagnosticSession({ + sessionId: "active-session", + sessionKey: "agent:main:main", + ageMs: 180_000, + allowActiveAbort: true, + }); + + expect(outcome).toMatchObject({ + status: "skipped", + action: "observe_only", + reason: "active_embedded_run", + activeSessionId: "active-session", + }); + }); + it("clears stale queued processing state even when the lane has no active work", async () => { mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue(undefined); mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(undefined); @@ -777,7 +801,7 @@ describe("stuck session recovery", () => { expect(mocks.resetCommandLane).toHaveBeenCalledWith("session:agent:main:main"); expect(warnLogMessages()).toEqual([ "stuck session recovery: sessionId=stale-session sessionKey=agent:main:main age=180s action=release_lane aborted=false drained=true released=0", - "stuck session recovery outcome: status=released action=release_lane sessionId=stale-session sessionKey=agent:main:main lane=session:agent:main:main released=0", + "stuck session recovery outcome: status=released action=release_lane sessionId=stale-session sessionKey=agent:main:main lane=session:agent:main:main reason=no_active_work released=0", ]); }); diff --git a/src/logging/diagnostic-stuck-session-recovery.runtime.ts b/src/logging/diagnostic-stuck-session-recovery.runtime.ts index 581627d5e0bf..80955a4b6fba 100644 --- a/src/logging/diagnostic-stuck-session-recovery.runtime.ts +++ b/src/logging/diagnostic-stuck-session-recovery.runtime.ts @@ -359,9 +359,9 @@ export async function recoverStuckDiagnosticSession( ? resetCommandLane(sessionLane) : 0; - const clearStaleQueuedSession = !aborted && released === 0 && (params.queueDepth ?? 0) > 0; + const clearStaleSession = !aborted && released === 0 && !activeSessionId; - if (aborted || forceCleared || released > 0 || clearStaleQueuedSession) { + if (aborted || forceCleared || released > 0 || clearStaleSession) { const action = aborted || forceCleared ? "abort_embedded_run" : "release_lane"; const stoppedFields = formatStoppedCronSessionDiagnosticFields( resolveCronSessionDiagnosticContext({ sessionKey: params.sessionKey, activeSessionId }), @@ -396,17 +396,21 @@ export async function recoverStuckDiagnosticSession( sessionKey: params.sessionKey, released, lane: sessionLane ?? undefined, + ...(clearStaleSession ? { reason: "no_active_work" as const } : {}), }; diag.warn(`stuck session recovery outcome: ${formatRecoveryOutcome(outcome)}`); return outcome; } + // An active run that neither aborted nor released still owns its work. Reporting + // recovery here would clear the session's diagnostic state out from under it. const outcome: StuckSessionRecoveryOutcome = { - status: "noop", - action: "none", - reason: "no_active_work", + status: "skipped", + action: "observe_only", + reason: "active_embedded_run", sessionId: params.sessionId, sessionKey: params.sessionKey, - lane: sessionLane ?? undefined, + activeSessionId, + activeWorkKind: "embedded_run", }; diag.warn(`stuck session recovery outcome: ${formatRecoveryOutcome(outcome)}`); return outcome; diff --git a/src/logging/diagnostic.test.ts b/src/logging/diagnostic.test.ts index cbce8ee9afbd..b6b1b94e3d51 100644 --- a/src/logging/diagnostic.test.ts +++ b/src/logging/diagnostic.test.ts @@ -1900,9 +1900,10 @@ describe("stuck session diagnostics threshold", () => { it("clears queued diagnostic state after no-active-work recovery", async () => { const events: DiagnosticEventPayload[] = []; const recoverStuckSession = vi.fn().mockResolvedValue({ - status: "noop", - action: "none", + status: "released", + action: "release_lane", reason: "no_active_work", + released: 0, sessionId: "s1", sessionKey: "main", }); @@ -1932,8 +1933,8 @@ describe("stuck session diagnostics threshold", () => { expect(state.queueDepth).toBe(0); requireMatchingRecord( events, - { type: "session.state", state: "idle", reason: "stuck_recovery:noop", queueDepth: 0 }, - "noop state clear event", + { type: "session.state", state: "idle", reason: "stuck_recovery:released", queueDepth: 0 }, + "released state clear event", ); }); @@ -1984,9 +1985,10 @@ describe("stuck session diagnostics threshold", () => { const events: DiagnosticEventPayload[] = []; let resolveRecovery: | ((outcome: { - status: "noop"; - action: "none"; + status: "released"; + action: "release_lane"; reason: "no_active_work"; + released: number; sessionId: string; sessionKey: string; }) => void) @@ -1994,9 +1996,10 @@ describe("stuck session diagnostics threshold", () => { const recoverStuckSession = vi.fn( () => new Promise<{ - status: "noop"; - action: "none"; + status: "released"; + action: "release_lane"; reason: "no_active_work"; + released: number; sessionId: string; sessionKey: string; }>((resolve) => { @@ -2033,9 +2036,10 @@ describe("stuck session diagnostics threshold", () => { ); resolveRecovery?.({ - status: "noop", - action: "none", + status: "released", + action: "release_lane", reason: "no_active_work", + released: 0, sessionId: "s1", sessionKey: "main", });