fix: unblock replies after recovery owner release conflict (#126507)

* fix: unblock replies after recovery owner release conflict

* fix(diagnostics): keep an unreclaimed active run as an observed skip

Removing the noop outcome left recoverStuckDiagnosticSession able to fall off
the end of its try block when an active run neither aborted nor released,
returning undefined from a Promise<StuckSessionRecoveryOutcome> (tsgo TS2366).
Report that residual path as skipped/active_embedded_run so the watchdog never
clears diagnostic state for work that still owns its session.

* fix(sessions): compare persisted row bytes in session-entry replacement CAS

The replacement projection snapshotted entries with the status/store readers
(no participant projection) and revalidated inside the transaction with
readExactSessionEntryRow, which merges session_participants into the entry.
Any row with a participant that is not its owner therefore serialized
differently on the two sides, so the compare-and-swap threw "SQLite session
entry changed before replacement" on every attempt with no concurrent write
at all. Startup orphan marking selects by status, so such a session could
never be repaired: it stayed status=running across restarts, its rotated
session id never persisted, and every later turn failed with "changed while
starting work", retried by the ingress spool for 24h.

Compare the persisted entry_json bytes on both sides, the same raw-bytes CAS
the sibling lifecycle and projection paths already use, so separately mutable
decorations (participants today, owner columns next) cannot invalidate a
logical-session write. Renames the raw reader to readExactSessionEntryJson
now that it is the general CAS reader rather than repair-only, and aligns
shouldRemoveSessionEntry with the participants-excluding equality its own
callers already use.

* fix(sessions): fail closed when a selected replacement row has no persisted bytes

The raw-bytes CAS could not distinguish "row unchanged" from "row gone". If a
selected row was deleted between hydrating the snapshot entry and reading its
persisted bytes, both the snapshot and the transaction read undefined, the
compare agreed, the transaction's source map stayed empty, and the stale
replacement was written back into the deleted key.

A selected key must hold bytes, so treat a missing snapshot read as the
conflict it is. Reported by ClawSweeper on the previous head.

---------

Co-authored-by: VACInc <3279061+VACInc@users.noreply.github.com>

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") {

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<MainSessionRecoveryPendingTarget | undefined>((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;
   }
 }
 

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<OpenClawAgentDatabase, "db">,
   sessionKey: string,
 ): string | undefined {

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}`,

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}`,
     );

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<typeof import("./session-accessor.sqlite-entry-store.js")>(
+    "./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" });
+  });
+});

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<T, TReplacement>(
     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<T, TReplacement>(
         const transactionEntries = new Map<string, SessionEntry>();
         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) {

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, {

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)
   );
 }
 

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",
     ]);
   });
 

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;

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",
       });