fix crash-only startup recovery

Author: YKDZ

apps/app-e2e/docker-compose.yml

@@ -10,6 +10,12 @@ services:
   redis:
     image: redis
     container_name: cat_e2e_redis
+    command:
+      - redis-server
+      - --appendonly
+      - "yes"
+      - --appendfsync
+      - everysec
     healthcheck:
       test:
         - CMD-SHELL

apps/app/docker-compose.yml

@@ -20,6 +20,14 @@ services:
 
   redis:
     image: redis
+    command:
+      - redis-server
+      - --appendonly
+      - "yes"
+      - --appendfsync
+      - everysec
+    volumes:
+      - redis-data:/data
     healthcheck:
       test:
         - CMD-SHELL
@@ -89,3 +97,6 @@ services:
       resources:
         limits:
           memory: 2G
+
+volumes:
+  redis-data:

apps/app/src/server/initialize.spec.ts

@@ -1,6 +1,9 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
 const mocks = vi.hoisted(() => {
+  const calls: string[] = [];
+  const activeRunIds = ["run-active"];
+  const fakeRecoverCrashedAgentRuns = Symbol("recoverCrashedAgentRuns");
   const fakeApp = {
     all: vi.fn(),
   };
@@ -72,14 +75,21 @@ const mocks = vi.hoisted(() => {
   const fakeCleanupHandle = { stop: vi.fn() };
   const fakePluginLoader = { kind: "plugin-loader" };
   const defaultPluginIds = ["password-auth-provider", "json-file-handler"];
+  const messageGatewayStart = vi.fn();
+  const serverError = vi.fn();
+  const serverInfo = vi.fn();
+  const serverWarn = vi.fn();
 
   return {
+    activeRunIds,
     assertSearchRuntimeHealth: vi.fn().mockResolvedValue(fakeDatabaseSummary),
+    calls,
     createAppPluginLoader: vi.fn(() => fakePluginLoader),
     createDefaultGraphRuntime: vi.fn(),
     createRuntimeBackends: vi.fn().mockResolvedValue(fakeBackends),
     ensureDB: vi.fn().mockResolvedValue(undefined),
     ensureRootUser: vi.fn().mockResolvedValue(undefined),
+    executeCommand: vi.fn(),
     executeQuery: vi.fn(),
     fakeApp,
     fakeBackends,
@@ -91,6 +101,7 @@ const mocks = vi.hoisted(() => {
     fakePluginLoader,
     fakePluginManager,
     fakeProfile,
+    fakeRecoverCrashedAgentRuns,
     fakeRouteRegistry,
     getCurrentRedisHandle: vi.fn().mockReturnValue(undefined),
     getDbHandle: vi.fn().mockResolvedValue(fakeDrizzleDB),
@@ -99,6 +110,9 @@ const mocks = vi.hoisted(() => {
       appMethodRegistry: { id: "registry" },
     })),
     getFirstRegisteredUser: Symbol("getFirstRegisteredUser"),
+    getGlobalGraphRuntimeOrNull: vi.fn(() => ({
+      scheduler: { getActiveRunIds: vi.fn(() => activeRunIds) },
+    })),
     getSetting: Symbol("getSetting"),
     grantFirstUserSuperadmin: vi.fn().mockResolvedValue(undefined),
     initCacheStore: vi.fn(),
@@ -107,23 +121,29 @@ const mocks = vi.hoisted(() => {
     initSessionStore: vi.fn(),
     initAllVectorStorage: vi.fn().mockResolvedValue(undefined),
     MessageGateway: class {
-      public start = vi.fn();
+      public start = messageGatewayStart;
     },
+    messageGatewayStart,
     pluginManagerGet: vi.fn(() => fakePluginManager),
     pluginManagerClear: vi.fn(),
     pluginManagerInstallDefaults: vi.fn().mockResolvedValue(undefined),
     registerAuditHandler: vi.fn(),
     registerBuiltinAgents: vi.fn().mockResolvedValue(undefined),
     registerDomainEventHandlers: vi.fn(),
     registerVectorizationConsumer: vi.fn().mockResolvedValue(undefined),
+    recoverCrashedAgentRuns: fakeRecoverCrashedAgentRuns,
     resolveRuntimeProfile: vi.fn(() => fakeProfile),
     seedSystemRoles: vi.fn().mockResolvedValue(undefined),
     serverLogger: {
       withSituation: () => ({
-        error: vi.fn(),
-        info: vi.fn(),
+        error: serverError,
+        info: serverInfo,
+        warn: serverWarn,
       }),
     },
+    serverError,
+    serverInfo,
+    serverWarn,
     setVectorizationQueue: vi.fn(),
     startPostgresRuntimeCleanup: vi.fn(() => fakeCleanupHandle),
     wireEntityStateFetchers: vi.fn(),
@@ -144,6 +164,7 @@ vi.mock("@cat/db", () => ({
 }));
 
 vi.mock("@cat/domain", () => ({
+  executeCommand: mocks.executeCommand,
   executeQuery: mocks.executeQuery,
   getCurrentRedisHandle: mocks.getCurrentRedisHandle,
   getDbHandle: mocks.getDbHandle,
@@ -152,6 +173,7 @@ vi.mock("@cat/domain", () => ({
   initCacheStore: mocks.initCacheStore,
   initRuntimeState: mocks.initRuntimeState,
   initSessionStore: mocks.initSessionStore,
+  recoverCrashedAgentRuns: mocks.recoverCrashedAgentRuns,
   resolveRuntimeProfile: mocks.resolveRuntimeProfile,
 }));
 
@@ -200,6 +222,7 @@ vi.mock("@cat/vcs", () => ({
 
 vi.mock("@cat/workflow", () => ({
   createDefaultGraphRuntime: mocks.createDefaultGraphRuntime,
+  getGlobalGraphRuntimeOrNull: mocks.getGlobalGraphRuntimeOrNull,
 }));
 
 vi.mock("./default-plugins/catalog", () => ({
@@ -224,13 +247,31 @@ import { initializeApp } from "./initialize";
 describe("initializeApp", () => {
   beforeEach(() => {
     vi.clearAllMocks();
+    mocks.calls.length = 0;
     mocks.executeQuery
       .mockResolvedValueOnce(null)
       .mockResolvedValueOnce("CAT")
       .mockResolvedValueOnce("http://localhost:3000/");
+    mocks.executeCommand.mockImplementation(async (_ctx, command) => {
+      if (command === mocks.recoverCrashedAgentRuns) {
+        mocks.calls.push("recover-runs");
+        return { recoveredRunIds: ["run-1"] };
+      }
+      return undefined;
+    });
+    mocks.registerDomainEventHandlers.mockImplementation(() => {
+      mocks.calls.push("domain-handlers");
+    });
+    mocks.registerVectorizationConsumer.mockImplementation(async () => {
+      mocks.calls.push("vectorization-consumer");
+    });
+    mocks.createDefaultGraphRuntime.mockImplementation(() => {
+      mocks.calls.push("graph-runtime");
+    });
     Reflect.deleteProperty(globalThis, "app");
     globalThis.inited = false;
     globalThis.redis = undefined;
+    Reflect.deleteProperty(globalThis, "messageGateway");
     Reflect.deleteProperty(globalThis, "pluginManager");
     globalThis.runtimeCleanup = undefined;
   });
@@ -285,6 +326,21 @@ describe("initializeApp", () => {
       mocks.fakePluginManager,
       mocks.getDefaultPluginIds(),
     );
+    expect(mocks.calls).toEqual([
+      "domain-handlers",
+      "recover-runs",
+      "vectorization-consumer",
+      "graph-runtime",
+    ]);
+    expect(mocks.executeCommand).toHaveBeenCalledWith(
+      { db: mocks.fakeDrizzleClient },
+      mocks.recoverCrashedAgentRuns,
+      { activeRunIds: mocks.activeRunIds },
+    );
+    expect(mocks.serverWarn).toHaveBeenCalledWith(
+      { recoveredRunIds: ["run-1"] },
+      "Recovered crashed workflow runs",
+    );
     expect(mocks.initAllVectorStorage).not.toHaveBeenCalled();
     expect(mocks.registerVectorizationConsumer).toHaveBeenCalledWith(
       mocks.fakeBackends.vectorizationQueue,
@@ -299,4 +355,49 @@ describe("initializeApp", () => {
 
     exitSpy.mockRestore();
   });
+
+  it("exits before readiness when run recovery fails", async () => {
+    const exitError = new Error("process.exit called");
+    const exitSpy = vi
+      .spyOn(process, "exit")
+      .mockImplementation(
+        (_code?: Parameters<typeof process.exit>[0]): never => {
+          throw exitError;
+        },
+      );
+    mocks.executeCommand.mockRejectedValueOnce(new Error("db recovery failed"));
+
+    await expect(initializeApp()).rejects.toBe(exitError);
+
+    expect(exitSpy).toHaveBeenCalledWith(1);
+    expect(globalThis.inited).not.toBe(true);
+    expect(mocks.registerVectorizationConsumer).not.toHaveBeenCalled();
+    expect(mocks.createDefaultGraphRuntime).not.toHaveBeenCalled();
+    expect(mocks.messageGatewayStart).not.toHaveBeenCalled();
+
+    exitSpy.mockRestore();
+  });
+
+  it("exits before readiness when queue recovery fails", async () => {
+    const exitError = new Error("process.exit called");
+    const exitSpy = vi
+      .spyOn(process, "exit")
+      .mockImplementation(
+        (_code?: Parameters<typeof process.exit>[0]): never => {
+          throw exitError;
+        },
+      );
+    mocks.registerVectorizationConsumer.mockRejectedValueOnce(
+      new Error("redis recovery failed"),
+    );
+
+    await expect(initializeApp()).rejects.toBe(exitError);
+
+    expect(exitSpy).toHaveBeenCalledWith(1);
+    expect(globalThis.inited).not.toBe(true);
+    expect(mocks.createDefaultGraphRuntime).not.toHaveBeenCalled();
+    expect(mocks.messageGatewayStart).not.toHaveBeenCalled();
+
+    exitSpy.mockRestore();
+  });
 });

apps/app/src/server/initialize.ts

@@ -2,12 +2,14 @@ import { registerBuiltinAgents } from "@cat/agent";
 import app from "@cat/app-api/app";
 import { ensureDB, ensureRootUser } from "@cat/db";
 import {
+  executeCommand,
   executeQuery,
   getFirstRegisteredUser,
   getSetting,
   getDbHandle,
   getCurrentRedisHandle,
   initRuntimeState,
+  recoverCrashedAgentRuns,
   resolveRuntimeProfile,
   type DrizzleClient,
   initCacheStore,
@@ -31,7 +33,10 @@ import {
 } from "@cat/server-shared";
 import { assertPromise } from "@cat/shared";
 import { getDefaultRegistries, wireEntityStateFetchers } from "@cat/vcs";
-import { createDefaultGraphRuntime } from "@cat/workflow";
+import {
+  createDefaultGraphRuntime,
+  getGlobalGraphRuntimeOrNull,
+} from "@cat/workflow";
 import { access } from "fs/promises";
 import { join } from "path";
 
@@ -141,6 +146,23 @@ export const initializeApp = async (): Promise<void> => {
     });
 
     registerDomainEventHandlers(drizzleDB.client, { pluginManager });
+
+    const existingRuntime = getGlobalGraphRuntimeOrNull();
+    const activeRunIds = existingRuntime?.scheduler.getActiveRunIds() ?? [];
+    const crashRecovery = await executeCommand(
+      { db: drizzleDB.client },
+      recoverCrashedAgentRuns,
+      { activeRunIds },
+    );
+    if (crashRecovery.recoveredRunIds.length > 0) {
+      logger
+        .withSituation("SERVER")
+        .warn(
+          { recoveredRunIds: crashRecovery.recoveredRunIds },
+          "Recovered crashed workflow runs",
+        );
+    }
+
     await registerVectorizationConsumer(backends.vectorizationQueue);
 
     const messageGateway = new MessageGateway({