refactor: add restart recovery lifecycle seam (#93735)

scripts/check-session-accessor-boundary.mjs

@@ -90,6 +90,7 @@ export const migratedSessionAccessorWriteFiles = new Set([
   "src/agents/command/session-store.ts",
   "src/agents/embedded-agent-runner/run.ts",
   "src/agents/embedded-agent-runner/run/attempt.ts",
+  "src/agents/main-session-restart-recovery.ts",
   "src/auto-reply/reply/abort-cutoff.runtime.ts",
   "src/auto-reply/reply/agent-runner-cli-dispatch.ts",
   "src/auto-reply/reply/agent-runner-execution.ts",

src/agents/main-session-restart-recovery.ts

@@ -15,8 +15,8 @@ import {
   resolveAllAgentSessionStoreTargetsSync,
   resolveSessionFilePath,
   resolveSessionTranscriptPathInDir,
-  updateSessionStore,
 } from "../config/sessions.js";
+import { applyRestartRecoveryLifecycle } from "../config/sessions/session-accessor.js";
 import type { OpenClawConfig } from "../config/types.openclaw.js";
 import { callGateway } from "../gateway/call.js";
 import { readSessionMessagesAsync } from "../gateway/session-transcript-readers.js";
@@ -213,13 +213,13 @@ export async function markRestartAbortedMainSessions(params: {
   }
 
   for (const storePath of storePaths) {
-    await updateSessionStore(
+    const storeResult = await applyRestartRecoveryLifecycle({
       storePath,
-      (store) => {
-        for (const [sessionKey, entry] of Object.entries(store)) {
-          if (!entry) {
-            continue;
-          }
+      requireWriteSuccess: true,
+      update: (entries) => {
+        const replacements: Array<{ sessionKey: string; entry: SessionEntry }> = [];
+        const counts = { marked: 0, skipped: 0 };
+        for (const { sessionKey, entry } of entries) {
           const registeredActiveRuns = listAgentRunsForSession({
             sessionKey,
             sessionId: entry.sessionId,
@@ -248,7 +248,7 @@ export async function markRestartAbortedMainSessions(params: {
             continue;
           }
           if (shouldSkipMainRecovery(entry, sessionKey)) {
-            result.skipped++;
+            counts.skipped++;
             continue;
           }
           const wasRunning = entry.status === "running";
@@ -288,12 +288,14 @@ export async function markRestartAbortedMainSessions(params: {
               : a.runId.localeCompare(b.runId),
           );
           entry.updatedAt = Date.now();
-          store[sessionKey] = entry;
-          result.marked++;
+          replacements.push({ sessionKey, entry });
+          counts.marked++;
         }
+        return { result: counts, replacements };
       },
-      { skipMaintenance: true, requireWriteSuccess: true },
-    );
+    });
+    result.marked += storeResult.marked;
+    result.skipped += storeResult.skipped;
   }
 
   if (result.marked > 0) {
@@ -327,18 +329,17 @@ export async function markStartupOrphanedMainSessionsForRecovery(params: {
     providedActiveSessionKeys ?? normalizeStringSet(listActiveEmbeddedRunSessionKeys());
 
   for (const storePath of await resolveRestartRecoveryStorePaths(params)) {
-    await updateSessionStore(
+    const storeResult = await applyRestartRecoveryLifecycle({
       storePath,
-      (store) => {
-        for (const [sessionKey, entry] of Object.entries(store)) {
-          if (!entry) {
-            continue;
-          }
+      update: (entries) => {
+        const replacements: Array<{ sessionKey: string; entry: SessionEntry }> = [];
+        const counts = { marked: 0, skipped: 0 };
+        for (const { sessionKey, entry } of entries) {
           if (entry.status !== "running" || entry.abortedLastRun === true) {
             continue;
           }
           if (shouldSkipMainRecovery(entry, sessionKey)) {
-            result.skipped++;
+            counts.skipped++;
             continue;
           }
           const updatedAt = normalizeFiniteTimestamp(entry.updatedAt);
@@ -361,12 +362,14 @@ export async function markStartupOrphanedMainSessionsForRecovery(params: {
           }
           entry.abortedLastRun = true;
           entry.updatedAt = Date.now();
-          store[sessionKey] = entry;
-          result.marked++;
+          replacements.push({ sessionKey, entry });
+          counts.marked++;
         }
+        return { result: counts, replacements };
       },
-      { skipMaintenance: true },
-    );
+    });
+    result.marked += storeResult.marked;
+    result.skipped += storeResult.skipped;
   }
 
   if (result.marked > 0) {
@@ -438,12 +441,13 @@ async function markSessionFailed(params: {
   sessionKey: string;
   reason: string;
 }): Promise<void> {
-  await updateSessionStore(
-    params.storePath,
-    (store) => {
-      const entry = store[params.sessionKey];
+  await applyRestartRecoveryLifecycle({
+    storePath: params.storePath,
+    update: (entries) => {
+      const current = entries.find((entry) => entry.sessionKey === params.sessionKey);
+      const entry = current?.entry;
       if (!entry || entry.status !== "running") {
-        return;
+        return { result: undefined };
       }
       entry.status = "failed";
       entry.abortedLastRun = true;
@@ -458,10 +462,12 @@ async function markSessionFailed(params: {
       entry.pendingFinalDeliveryContext = undefined;
       entry.restartRecoveryDeliveryContext = undefined;
       entry.restartRecoveryDeliveryRunId = undefined;
-      store[params.sessionKey] = entry;
+      return {
+        result: undefined,
+        replacements: [{ sessionKey: params.sessionKey, entry }],
+      };
     },
-    { skipMaintenance: true },
-  );
+  });
   log.warn(`marked interrupted main session failed: ${params.sessionKey} (${params.reason})`);
 }
 
@@ -594,12 +600,13 @@ async function resumeMainSession(params: {
       params: agentParams,
       timeoutMs: 10_000,
     });
-    await updateSessionStore(
-      params.storePath,
-      (store) => {
-        const entry = store[params.sessionKey];
+    await applyRestartRecoveryLifecycle({
+      storePath: params.storePath,
+      update: (entries) => {
+        const current = entries.find((entry) => entry.sessionKey === params.sessionKey);
+        const entry = current?.entry;
         if (!entry) {
-          return;
+          return { result: undefined };
         }
         const now = Date.now();
         entry.abortedLastRun = false;
@@ -621,10 +628,12 @@ async function resumeMainSession(params: {
             entry.pendingFinalDeliveryContext = undefined;
           }
         }
-        store[params.sessionKey] = entry;
+        return {
+          result: undefined,
+          replacements: [{ sessionKey: params.sessionKey, entry }],
+        };
       },
-      { skipMaintenance: true },
-    );
+    });
     log.info(
       `resumed interrupted main session: ${params.sessionKey}${
         sanitizedPendingText ? " (with pending payload)" : ""
@@ -653,15 +662,17 @@ export async function markRestartAbortedMainSessionsFromLocks(params: {
   }
 
   const storePath = path.join(sessionsDir, "sessions.json");
-  await updateSessionStore(
+  const storeResult = await applyRestartRecoveryLifecycle({
     storePath,
-    (store) => {
-      for (const [sessionKey, entry] of Object.entries(store)) {
-        if (!entry || entry.status !== "running") {
+    update: (entries) => {
+      const replacements: Array<{ sessionKey: string; entry: SessionEntry }> = [];
+      const counts = { marked: 0, skipped: 0 };
+      for (const { sessionKey, entry } of entries) {
+        if (entry.status !== "running") {
           continue;
         }
         if (shouldSkipMainRecovery(entry, sessionKey)) {
-          result.skipped++;
+          counts.skipped++;
           continue;
         }
         const entryLockPaths = resolveEntryTranscriptLockPaths({ entry, sessionsDir });
@@ -669,12 +680,14 @@ export async function markRestartAbortedMainSessionsFromLocks(params: {
           continue;
         }
         entry.abortedLastRun = true;
-        store[sessionKey] = entry;
-        result.marked++;
+        replacements.push({ sessionKey, entry });
+        counts.marked++;
       }
+      return { result: counts, replacements };
     },
-    { skipMaintenance: true },
-  );
+  });
+  result.marked += storeResult.marked;
+  result.skipped += storeResult.skipped;
 
   if (result.marked > 0) {
     log.warn(`marked ${result.marked} interrupted main session(s) from stale transcript locks`);

src/config/sessions/session-accessor.test.ts

@@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest";
 import { onSessionTranscriptUpdate } from "../../sessions/transcript-events.js";
 import type { OpenClawConfig } from "../types.openclaw.js";
 import {
+  applyRestartRecoveryLifecycle,
   appendTranscriptMessage,
   appendTranscriptEvent,
   applySessionEntryLifecycleMutation,
@@ -448,6 +449,62 @@ describe("session accessor file-backed seam", () => {
     });
   });
 
+  it("applies restart recovery replacements without exposing mutable store rows", async () => {
+    fs.writeFileSync(
+      storePath,
+      JSON.stringify(
+        {
+          "agent:main:main": {
+            sessionId: "session-1",
+            status: "running",
+            updatedAt: 10,
+          },
+          "agent:main:other": {
+            sessionId: "session-2",
+            status: "running",
+            updatedAt: 20,
+          },
+        } satisfies Record<string, SessionEntry>,
+        null,
+        2,
+      ),
+      "utf8",
+    );
+
+    const result = await applyRestartRecoveryLifecycle({
+      storePath,
+      update: (entries) => {
+        const main = entries.find((entry) => entry.sessionKey === "agent:main:main");
+        const other = entries.find((entry) => entry.sessionKey === "agent:main:other");
+        if (other) {
+          other.entry.status = "failed";
+        }
+        if (!main) {
+          return { result: { replaced: false } };
+        }
+        main.entry.abortedLastRun = true;
+        main.entry.updatedAt = 30;
+        return {
+          result: { replaced: true },
+          replacements: [{ sessionKey: main.sessionKey, entry: main.entry }],
+        };
+      },
+    });
+
+    expect(result).toEqual({ replaced: true });
+    const store = loadSessionStore(storePath);
+    expect(store["agent:main:main"]).toMatchObject({
+      abortedLastRun: true,
+      sessionId: "session-1",
+      updatedAt: 30,
+    });
+    expect(store["agent:main:other"]).toMatchObject({
+      sessionId: "session-2",
+      status: "running",
+      updatedAt: 20,
+    });
+  });
+
   it("cleans scoped lifecycle entries and unreferenced transcript artifacts", async () => {
     const nowMs = Date.now();
     const oldDate = new Date(nowMs - 600_000);

src/config/sessions/session-accessor.ts

@@ -287,6 +287,27 @@ export type SessionEntryPatchContext = {
   existingEntry?: SessionEntry;
 };
 
+export type RestartRecoveryLifecycleEntry = {
+  /** Exact persisted key for the restart recovery candidate row. */
+  sessionKey: string;
+  /** Detached entry snapshot; mutating it does not persist unless returned as a replacement. */
+  entry: SessionEntry;
+};
+
+export type RestartRecoveryLifecycleReplacement = {
+  /** Exact persisted key to replace. Missing keys are ignored. */
+  sessionKey: string;
+  /** Full replacement row to persist for this restart recovery lifecycle step. */
+  entry: SessionEntry;
+};
+
+export type RestartRecoveryLifecycleUpdate<T> = {
+  /** Caller-owned result returned after replacements are persisted. */
+  result: T;
+  /** Exact rows to replace inside the storage transaction. */
+  replacements?: Iterable<RestartRecoveryLifecycleReplacement>;
+};
+
 export type SessionEntryCreateWithTranscriptContext = {
   /** Current entry under the requested key before creation, if any. */
   existingEntry?: SessionEntry;
@@ -590,6 +611,46 @@ export async function applySessionPatchProjection<
   return await applyFileSessionEntryPatchProjection(params);
 }
 
+/**
+ * Applies restart-recovery lifecycle replacements without exposing the backing
+ * store shape. The file backend runs selection and replacement under one writer
+ * lock; the SQLite backend can map the same callback to a transaction.
+ */
+export async function applyRestartRecoveryLifecycle<T>(params: {
+  storePath: string;
+  update: (
+    entries: RestartRecoveryLifecycleEntry[],
+  ) => Promise<RestartRecoveryLifecycleUpdate<T>> | RestartRecoveryLifecycleUpdate<T>;
+  requireWriteSuccess?: boolean;
+  skipMaintenance?: boolean;
+}): Promise<T> {
+  const writerResult = await updateSessionStore(
+    params.storePath,
+    async (store) => {
+      const entries = Object.entries(store).map(([sessionKey, entry]) => ({
+        sessionKey,
+        entry: structuredClone(entry),
+      }));
+      const operation = await params.update(entries);
+      let changed = false;
+      for (const replacement of operation.replacements ?? []) {
+        if (!Object.hasOwn(store, replacement.sessionKey)) {
+          continue;
+        }
+        store[replacement.sessionKey] = structuredClone(replacement.entry);
+        changed = true;
+      }
+      return { changed, result: operation.result };
+    },
+    {
+      requireWriteSuccess: params.requireWriteSuccess,
+      skipMaintenance: params.skipMaintenance ?? true,
+      skipSaveWhenResult: (result) => !result.changed,
+    },
+  );
+  return writerResult.result;
+}
+
 /** Removes entries and orphan transcript artifacts owned by a named session lifecycle. */
 export async function cleanupSessionLifecycleArtifacts(
   params: SessionLifecycleArtifactCleanupParams,

test/scripts/check-session-accessor-boundary.test.ts

@@ -65,6 +65,7 @@ describe("session accessor boundary guard", () => {
         "src/agents/command/session-store.ts",
         "src/agents/embedded-agent-runner/run.ts",
         "src/agents/embedded-agent-runner/run/attempt.ts",
+        "src/agents/main-session-restart-recovery.ts",
         "src/auto-reply/reply/abort-cutoff.runtime.ts",
         "src/auto-reply/reply/agent-runner-cli-dispatch.ts",
         "src/auto-reply/reply/agent-runner-execution.ts",