fix(tasks): fail closed on registry restore errors (#105031)

src/cli/gateway-cli/run-loop.test.ts

@@ -1365,6 +1365,83 @@ describe("runGatewayLoop", () => {
     });
   });
 
+  it("keeps the process alive and retries after task-registry restore fails during restart", async () => {
+    vi.clearAllMocks();
+    reloadTaskRegistryFromStore.mockReset();
+    reloadTaskRegistryFromStore.mockImplementationOnce(() => {
+      throw new Error("task registry restore failed");
+    });
+    peekGatewaySigusr1RestartReason.mockReturnValue(undefined);
+    respawnGatewayProcessForUpdate.mockReturnValue({
+      mode: "disabled",
+      detail: "OPENCLAW_NO_RESPAWN",
+    });
+
+    try {
+      await withIsolatedSignals(async ({ captureSignal }) => {
+        const closeFirst = createCloseMock();
+        const closeSecond = createCloseMock();
+        const { runtime, exited } = createRuntimeWithExitSignal();
+        let resolveSecondStart: (() => void) | null = null;
+        const startedSecond = new Promise<void>((resolve) => {
+          resolveSecondStart = resolve;
+        });
+        const start = vi
+          .fn()
+          .mockResolvedValueOnce({ close: closeFirst })
+          .mockImplementationOnce(async () => {
+            resolveSecondStart?.();
+            return { close: closeSecond };
+          });
+
+        const { runGatewayLoop } = await import("./run-loop.js");
+        void runGatewayLoop({
+          start: start as unknown as Parameters<typeof runGatewayLoop>[0]["start"],
+          runtime: runtime as unknown as Parameters<typeof runGatewayLoop>[0]["runtime"],
+        });
+        await waitForLoopCondition(
+          () => start.mock.calls.length === 1,
+          "expected initial gateway start",
+        );
+        const sigusr1 = captureSignal("SIGUSR1");
+        const sigterm = captureSignal("SIGTERM");
+
+        try {
+          sigusr1();
+          await waitForLoopCondition(
+            () =>
+              gatewayLog.error.mock.calls.some(([message]) =>
+                String(message).includes("gateway startup failed: task registry restore failed."),
+              ),
+            "expected failed task-registry restore to be logged",
+          );
+
+          expectRestartCloseCall(closeFirst, 90_000);
+          expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(1);
+          expect(start).toHaveBeenCalledTimes(1);
+          expect(runtime.exit).not.toHaveBeenCalled();
+
+          sigusr1();
+          await startedSecond;
+
+          expect(reloadTaskRegistryFromStore).toHaveBeenCalledTimes(2);
+          expect(start).toHaveBeenCalledTimes(2);
+          expect(runtime.exit).not.toHaveBeenCalled();
+        } finally {
+          sigterm();
+          await expect(exited).resolves.toBe(0);
+        }
+
+        expect(closeSecond).toHaveBeenCalledWith({
+          reason: "gateway stopping",
+          restartExpectedMs: null,
+        });
+      });
+    } finally {
+      reloadTaskRegistryFromStore.mockReset();
+    }
+  });
+
   it("uses the default restart drain timeout when config omits deferralTimeoutMs", async () => {
     vi.clearAllMocks();
     loadConfig.mockReturnValue({ gateway: { reload: {} } });

src/cli/gateway-cli/run-loop.ts

@@ -918,11 +918,13 @@ export async function runGatewayLoop(params: {
     // SIGTERM/SIGINT still exit after a graceful shutdown.
     let isFirstStart = true;
     for (;;) {
-      await onIteration();
+      // The restart hook reopens admission before reloading durable state. Clear
+      // its local mirror first so a failed reload cannot skip the next drain.
       restartDrainingMarked = false;
-      startupStartedAt = Date.now();
       let startupFailedBeforeServerHandle = false;
       try {
+        await onIteration();
+        startupStartedAt = Date.now();
         await params.beginBoot?.(startupStartedAt);
         server = await params.start({ startupStartedAt });
         startupFailedWithoutServerHandle = false;

src/tasks/task-registry.store.test.ts

@@ -35,6 +35,7 @@ import {
   listFreshTasksForOwnerKey,
   markTaskTerminalById,
   maybeDeliverTaskStateChangeUpdate,
+  reloadTaskRegistryFromStore,
   resetTaskRegistryForTests,
   updateTaskNotifyPolicyById,
 } from "./task-registry.js";
@@ -179,24 +180,28 @@ describe("task-registry store runtime", () => {
     expect(latestSnapshot.tasks.get("task-restored")?.task).toBe("Restored task");
   });
 
-  it("logs restore parser failures and keeps the registry empty", async () => {
+  it("logs restore parser failures and keeps the failure sticky", async () => {
     const warnLogs = createWarnLogCapture("openclaw-task-registry-restore-test");
     const invalidValue = "not-requested";
+    const loadSnapshot = vi.fn(() => {
+      throw new Error(`Invalid persisted task delivery status: ${JSON.stringify(invalidValue)}`);
+    });
     try {
       configureTaskRegistryRuntime({
         store: {
-          loadSnapshot: () => {
-            throw new Error(
-              `Invalid persisted task delivery status: ${JSON.stringify(invalidValue)}`,
-            );
-          },
+          loadSnapshot,
           saveSnapshot: () => {},
         },
       });
 
-      expect(findTaskByRunId("run-restored")).toBeUndefined();
+      expect(() => findTaskByRunId("run-restored")).toThrow(
+        `Task registry restore failed: Invalid persisted task delivery status: "${invalidValue}"`,
+      );
       expect(await warnLogs.findText(invalidValue)).toContain(invalidValue);
-      expect(getTaskById("task-restored")).toBeUndefined();
+      expect(() => getTaskById("task-restored")).toThrow(
+        `Task registry restore failed: Invalid persisted task delivery status: "${invalidValue}"`,
+      );
+      expect(loadSnapshot).toHaveBeenCalledTimes(1);
     } finally {
       warnLogs.cleanup();
     }
@@ -223,13 +228,92 @@ describe("task-registry store runtime", () => {
       },
     });
 
-    expect(findTaskByRunId("run-restored")).toBeUndefined();
+    expect(() => findTaskByRunId("run-restored")).toThrow(
+      `Task registry restore failed: Invalid persisted task delivery status: "${invalidValue}"`,
+    );
     expect(warn).toHaveBeenCalledTimes(1);
     expect(String(warn.mock.calls[0]?.[0])).toContain(
       `Failed to restore task registry: Invalid persisted task delivery status: "${invalidValue}"`,
     );
   });
 
+  it("blocks writes until an explicit reload recovers the registry", () => {
+    const storedTask = createStoredTask();
+    let restoreShouldFail = true;
+    const loadSnapshot = vi.fn(() => {
+      if (restoreShouldFail) {
+        throw new Error("SQLITE_CORRUPT: malformed task registry");
+      }
+      return {
+        tasks: new Map([[storedTask.taskId, storedTask]]),
+        deliveryStates: new Map(),
+      };
+    });
+    const upsertTaskWithDeliveryState = vi.fn();
+    configureTaskRegistryRuntime({
+      store: {
+        loadSnapshot,
+        saveSnapshot: () => {},
+        upsertTaskWithDeliveryState,
+      },
+    });
+
+    expect(() =>
+      createTaskRecord({
+        runtime: "cron",
+        ownerKey: "agent:main:main",
+        scopeKind: "session",
+        task: "Must not overwrite hidden durable tasks",
+        status: "queued",
+        deliveryStatus: "not_applicable",
+        notifyPolicy: "silent",
+      }),
+    ).toThrow("Task registry restore failed: SQLITE_CORRUPT: malformed task registry");
+    expect(() => getTaskById(storedTask.taskId)).toThrow(
+      "Task registry restore failed: SQLITE_CORRUPT: malformed task registry",
+    );
+    expect(upsertTaskWithDeliveryState).not.toHaveBeenCalled();
+    expect(loadSnapshot).toHaveBeenCalledTimes(1);
+
+    restoreShouldFail = false;
+    reloadTaskRegistryFromStore();
+
+    expect(getTaskById(storedTask.taskId)).toMatchObject({ taskId: storedTask.taskId });
+    expect(loadSnapshot).toHaveBeenCalledTimes(2);
+  });
+
+  it("clears a sticky restore failure during the test reset boundary", () => {
+    const failedLoad = vi.fn(() => {
+      throw new Error("SQLITE_IOERR: failed to read task registry");
+    });
+    configureTaskRegistryRuntime({
+      store: {
+        loadSnapshot: failedLoad,
+        saveSnapshot: () => {},
+      },
+    });
+
+    expect(() => getTaskById("task-restored")).toThrow(
+      "Task registry restore failed: SQLITE_IOERR: failed to read task registry",
+    );
+    resetTaskRegistryForTests({ persist: false });
+
+    const cleanLoad = vi.fn(() => ({
+      tasks: new Map<string, TaskRecord>(),
+      deliveryStates: new Map<string, TaskDeliveryState>(),
+    }));
+    configureTaskRegistryRuntime({
+      store: {
+        loadSnapshot: cleanLoad,
+        saveSnapshot: () => {},
+      },
+    });
+
+    expect(getTaskById("task-restored")).toBeUndefined();
+    expect(failedLoad).toHaveBeenCalledTimes(1);
+    expect(cleanLoad).toHaveBeenCalledTimes(1);
+  });
+
   it("uses scoped owner lookups for fresh owner task reads", () => {
     const storedTask = createStoredTask();
     const loadSnapshot = vi.fn(() => ({

src/tasks/task-registry.test.ts

@@ -15,7 +15,9 @@ import {
 import type { SessionBindingRecord } from "../infra/outbound/session-binding-service.js";
 import { peekSystemEvents, resetSystemEventsForTest } from "../infra/system-events.js";
 import {
+  beginGatewayRestartSignalAdmission,
   getActiveGatewayRootWorkCount,
+  markGatewayRestartDraining,
   resetGatewayWorkAdmission,
   tryBeginGatewaySuspendAdmission,
 } from "../process/gateway-work-admission.js";
@@ -3690,6 +3692,114 @@ describe("task-registry", () => {
     });
   });
 
+  it("reattaches the lifecycle listener after recovering from an initial restore failure", async () => {
+    await withTaskRegistryTempDir(async () => {
+      resetTaskRegistryMemoryForTest();
+      const runId = "run-restore-listener";
+      const storedTask: TaskRecord = {
+        taskId: "task-restore-listener",
+        runtime: "acp",
+        requesterSessionKey: "agent:main:main",
+        ownerKey: "agent:main:main",
+        scopeKind: "session",
+        runId,
+        task: "Resume lifecycle tracking after restore recovery",
+        status: "running",
+        deliveryStatus: "not_applicable",
+        notifyPolicy: "silent",
+        createdAt: 100,
+        startedAt: 100,
+        lastEventAt: 100,
+      };
+      let restoreShouldFail = true;
+      configureTaskRegistryRuntime({
+        store: {
+          loadSnapshot: () => {
+            if (restoreShouldFail) {
+              throw new Error("SQLITE_IOERR: initial task restore failed");
+            }
+            return {
+              tasks: new Map([[storedTask.taskId, storedTask]]),
+              deliveryStates: new Map(),
+            };
+          },
+          saveSnapshot: () => {},
+        },
+      });
+
+      expect(() => getTaskById(storedTask.taskId)).toThrow(
+        "Task registry restore failed: SQLITE_IOERR: initial task restore failed",
+      );
+      restoreShouldFail = false;
+      reloadTaskRegistryFromStore();
+
+      emitAgentEvent({
+        runId,
+        stream: "lifecycle",
+        data: {
+          phase: "end",
+          endedAt: 250,
+        },
+      });
+
+      expectRecordFields(requireTaskByRunId(runId), {
+        status: "succeeded",
+        endedAt: 250,
+      });
+    });
+  });
+
+  it("does not hide a failed reload behind the restart-draining delivery fallback", async () => {
+    await withTaskRegistryTempDir(async () => {
+      resetTaskRegistryMemoryForTest();
+      const storedTask: TaskRecord = {
+        taskId: "task-reload-failure",
+        runtime: "acp",
+        requesterSessionKey: "agent:main:main",
+        ownerKey: "agent:main:main",
+        scopeKind: "session",
+        runId: "run-reload-failure",
+        task: "Keep restore failures visible",
+        status: "succeeded",
+        deliveryStatus: "pending",
+        notifyPolicy: "done_only",
+        createdAt: 100,
+        endedAt: 200,
+        lastEventAt: 200,
+      };
+      let restoreError: Error | null = null;
+      configureTaskRegistryRuntime({
+        store: {
+          loadSnapshot: () => {
+            if (restoreError) {
+              throw restoreError;
+            }
+            return {
+              tasks: new Map([[storedTask.taskId, storedTask]]),
+              deliveryStates: new Map(),
+            };
+          },
+          saveSnapshot: () => {},
+        },
+      });
+      expect(getTaskById(storedTask.taskId)?.taskId).toBe(storedTask.taskId);
+
+      beginGatewayRestartSignalAdmission();
+      const pendingDelivery = maybeDeliverTaskTerminalUpdate(storedTask.taskId);
+      await Promise.resolve();
+
+      restoreError = new Error("SQLITE_CORRUPT: task reload failed");
+      expect(() => reloadTaskRegistryFromStore()).toThrow(
+        "Task registry restore failed: SQLITE_CORRUPT: task reload failed",
+      );
+      markGatewayRestartDraining();
+
+      await expect(pendingDelivery).rejects.toThrow(
+        "Task registry restore failed: SQLITE_CORRUPT: task reload failed",
+      );
+    });
+  });
+
   it("summarizes inspectable task audit findings", async () => {
     await withTaskRegistryTempDir(async () => {
       resetTaskRegistryMemoryForTest();

src/tasks/task-registry.ts

@@ -77,7 +77,12 @@ const taskIdsByRelatedSessionKey = taskRegistryProcessState.taskIdsByRelatedSess
 const tasksWithPendingDelivery = taskRegistryProcessState.tasksWithPendingDelivery;
 let listenerStarted = false;
 let listenerStop: (() => void) | null = null;
-let restoreAttempted = false;
+type TaskRegistryRestoreState =
+  | { status: "uninitialized" }
+  | { status: "restoring" }
+  | { status: "ready" }
+  | { status: "failed"; error: Error };
+let taskRegistryRestoreState: TaskRegistryRestoreState = { status: "uninitialized" };
 const taskFlowSyncRetryTimers = new Map<string, ReturnType<typeof setTimeout>>();
 type TaskRegistryDeliveryRuntime = Pick<
   typeof import("./task-registry-delivery-runtime.js"),
@@ -1208,36 +1213,54 @@ function clearTaskFlowSyncRetries(): void {
 }
 
 function restoreTaskRegistryOnce() {
-  if (restoreAttempted) {
-    return;
+  switch (taskRegistryRestoreState.status) {
+    case "ready":
+      return;
+    case "failed":
+      throw taskRegistryRestoreState.error;
+    case "restoring":
+      throw new Error("Task registry restore is already in progress.");
+    case "uninitialized":
+      break;
   }
-  restoreAttempted = true;
+  taskRegistryRestoreState = { status: "restoring" };
   try {
     const restored = getTaskRegistryStore().loadSnapshot();
-    if (restored.tasks.size === 0 && restored.deliveryStates.size === 0) {
-      return;
-    }
+    const restoredTasks = new Map<string, TaskRecord>();
     for (const [taskId, task] of restored.tasks.entries()) {
-      tasks.set(taskId, normalizeTaskTimestamps(task));
+      restoredTasks.set(taskId, normalizeTaskTimestamps(task));
     }
-    for (const [taskId, state] of restored.deliveryStates.entries()) {
+    const restoredDeliveryStates = new Map(restored.deliveryStates);
+
+    clearTaskRegistryMemory();
+    for (const [taskId, task] of restoredTasks.entries()) {
+      tasks.set(taskId, task);
+    }
+    for (const [taskId, state] of restoredDeliveryStates.entries()) {
       taskDeliveryStates.set(taskId, state);
     }
     rebuildRunIdIndex();
     rebuildOwnerKeyIndex();
     rebuildParentFlowIdIndex();
     rebuildRelatedSessionKeyIndex();
-    emitTaskRegistryObserverEvent(() => ({
-      kind: "restored",
-      tasks: snapshotTaskRecords(tasks),
-    }));
+    taskRegistryRestoreState = { status: "ready" };
+    if (restoredTasks.size > 0 || restoredDeliveryStates.size > 0) {
+      emitTaskRegistryObserverEvent(() => ({
+        kind: "restored",
+        tasks: snapshotTaskRecords(tasks),
+      }));
+    }
   } catch (error) {
+    clearTaskRegistryMemory();
     const message = formatErrorMessage(error);
+    const restoreError = new Error(`Task registry restore failed: ${message}`, { cause: error });
+    taskRegistryRestoreState = { status: "failed", error: restoreError };
     // Compact console logs omit structured metadata, so keep the rejected value visible there too.
     log.warn("Failed to restore task registry", {
       error: message,
       consoleMessage: `Failed to restore task registry: ${message}`,
     });
+    throw restoreError;
   }
 }
 
@@ -1248,8 +1271,8 @@ export function ensureTaskRegistryReady() {
 
 export function reloadTaskRegistryFromStore(): void {
   clearTaskRegistryMemory();
-  restoreAttempted = false;
-  restoreTaskRegistryOnce();
+  taskRegistryRestoreState = { status: "uninitialized" };
+  ensureTaskRegistryReady();
 }
 
 function updateTask(taskId: string, patch: Partial<TaskRecord>): TaskRecord | null {
@@ -1428,6 +1451,7 @@ async function runTaskDeliveryWithIndependentAdmission(
   taskId: string,
   deliver: () => Promise<TaskRecord | null>,
 ): Promise<TaskRecord | null> {
+  ensureTaskRegistryReady();
   let admitted = false;
   try {
     return await runWithGatewayIndependentRootWorkAdmission(async () => {
@@ -1439,6 +1463,7 @@ async function runTaskDeliveryWithIndependentAdmission(
     // restart closes admission. An already-admitted delivery still reports its
     // own failures instead of hiding them behind a concurrent restart.
     if (!admitted && isGatewayRestartDraining()) {
+      ensureTaskRegistryReady();
       const current = tasks.get(taskId);
       return current ? cloneTaskRecord(current) : null;
     }
@@ -2675,7 +2700,7 @@ export function deleteTaskRecordById(taskId: string): boolean {
 
 export function resetTaskRegistryForTests(opts?: { persist?: boolean }) {
   clearTaskRegistryMemory();
-  restoreAttempted = false;
+  taskRegistryRestoreState = { status: "uninitialized" };
   resetTaskRegistryRuntimeForTests();
   if (listenerStop) {
     listenerStop();