feat(FN-4743): complete Step 3 — wire mail cleanup into maintenance

Fusion-Task-Id: FN-4743
Fusion-Task-Lineage: 50d1d8e4-66d5-4299-843b-28072a904277

packages/engine/src/__tests__/self-healing-mail-cleanup.test.ts

@@ -0,0 +1,92 @@
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { rm } from "node:fs/promises";
+
+import { Database, MessageStore } from "@fusion/core";
+
+import { SelfHealingManager } from "../self-healing.js";
+
+describe("FN-4743: self-healing mail cleanup maintenance", () => {
+  let tmpRoot: string;
+  let db: Database;
+  let messageStore: MessageStore;
+
+  beforeAll(() => {
+    tmpRoot = mkdtempSync(join(tmpdir(), "fusion-self-healing-mail-cleanup-"));
+    const fusionDir = join(tmpRoot, ".fusion");
+    db = new Database(fusionDir, { inMemory: true });
+    db.init();
+    messageStore = new MessageStore(db);
+  });
+
+  beforeEach(() => {
+    db.exec("DELETE FROM messages");
+  });
+
+  afterAll(async () => {
+    db.close();
+    await rm(tmpRoot, { recursive: true, force: true });
+  });
+
+  function buildManager(mailAutoCleanupDays?: number, includeMessageStore = true) {
+    const store = {
+      getSettings: vi.fn(async () => ({ maintenanceIntervalMs: 0, globalPause: false, enginePaused: false, mailAutoCleanupDays })),
+    } as any;
+
+    const manager = new SelfHealingManager(store, {
+      rootDir: tmpRoot,
+      messageStore: includeMessageStore ? messageStore : undefined,
+    });
+    vi.spyOn(manager as any, "pruneWorktrees").mockResolvedValue(undefined);
+    vi.spyOn(manager as any, "cleanupOrphans").mockResolvedValue(undefined);
+    vi.spyOn(manager as any, "cleanupOrphanedBranches").mockResolvedValue(undefined);
+    vi.spyOn(manager as any, "checkpointWal").mockReturnValue(undefined);
+    vi.spyOn(manager as any, "enforceWorktreeCap").mockResolvedValue(undefined);
+    vi.spyOn(manager, "archiveStaleDoneTasks").mockResolvedValue(0);
+
+    return manager;
+  }
+
+  it("removes only stale messages when mailAutoCleanupDays is enabled", async () => {
+    const stale = messageStore.sendMessage({ fromId: "user-1", fromType: "user", toId: "agent-1", toType: "agent", content: "stale", type: "user-to-agent" });
+    const fresh = messageStore.sendMessage({ fromId: "agent-1", fromType: "agent", toId: "user-1", toType: "user", content: "fresh", type: "agent-to-user" });
+
+    const staleTimestamp = new Date(Date.now() - 12 * 86_400_000).toISOString();
+    const freshTimestamp = new Date(Date.now() - 1 * 86_400_000).toISOString();
+    db.prepare("UPDATE messages SET updatedAt = ? WHERE id = ?").run(staleTimestamp, stale.id);
+    db.prepare("UPDATE messages SET updatedAt = ? WHERE id = ?").run(freshTimestamp, fresh.id);
+
+    const manager = buildManager(7, true);
+    await (manager as any).runMaintenance();
+
+    expect(messageStore.getMessage(stale.id)).toBeNull();
+    expect(messageStore.getMessage(fresh.id)).not.toBeNull();
+  });
+
+  it("is a no-op when mailAutoCleanupDays is off or undefined", async () => {
+    const oldMessage = messageStore.sendMessage({ fromId: "user-1", fromType: "user", toId: "agent-1", toType: "agent", content: "keep", type: "user-to-agent" });
+    const oldTimestamp = new Date(Date.now() - 100 * 86_400_000).toISOString();
+    db.prepare("UPDATE messages SET updatedAt = ? WHERE id = ?").run(oldTimestamp, oldMessage.id);
+
+    const managerOff = buildManager(0, true);
+    await (managerOff as any).runMaintenance();
+    expect(messageStore.getMessage(oldMessage.id)).not.toBeNull();
+
+    const managerUndefined = buildManager(undefined, true);
+    await (managerUndefined as any).runMaintenance();
+    expect(messageStore.getMessage(oldMessage.id)).not.toBeNull();
+  });
+
+  it("is a no-op when messageStore option is omitted", async () => {
+    const oldMessage = messageStore.sendMessage({ fromId: "user-2", fromType: "user", toId: "agent-2", toType: "agent", content: "keep-no-store", type: "user-to-agent" });
+    const oldTimestamp = new Date(Date.now() - 100 * 86_400_000).toISOString();
+    db.prepare("UPDATE messages SET updatedAt = ? WHERE id = ?").run(oldTimestamp, oldMessage.id);
+
+    const manager = buildManager(7, false);
+    await (manager as any).runMaintenance();
+
+    expect(messageStore.getMessage(oldMessage.id)).not.toBeNull();
+  });
+});

packages/engine/src/runtimes/in-process-runtime.ts

@@ -657,6 +657,7 @@ export class InProcessRuntime
         leaseManager: this.leaseManager,
         hasActiveAgentExecution: (agentId: string) => this.heartbeatMonitor?.getTrackedAgents().includes(agentId) ?? false,
         chatStore: this.chatStore,
+        messageStore: this.messageStore,
         restartDurableAgentHeartbeat: async (agentId: string, context: { reason: string; attempt: number }) => {
           if (!this.heartbeatMonitor) {
             return false;

packages/engine/src/self-healing.ts

@@ -26,7 +26,7 @@ import { exec, execSync } from "node:child_process";
 import { promisify } from "node:util";
 import { existsSync, mkdirSync, readdirSync, rmSync, statSync, writeFileSync } from "node:fs";
 import { isAbsolute, join, relative, resolve } from "node:path";
-import { getInReviewStallReason, getStalePausedReviewSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, type AgentStore, type ChatStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority } from "@fusion/core";
+import { getInReviewStallReason, getStalePausedReviewSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, type AgentStore, type ChatStore, type MessageStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority } from "@fusion/core";
 import type { MeshLeaseManager } from "./mesh-lease-manager.js";
 import { createLogger } from "./logger.js";
 import { getRegisteredWorktreePaths, isUsableTaskWorktree, removeWorktree, resolveWorktreeBackend, scanIdleWorktrees, scanOrphanedBranches } from "./worktree-pool.js";
@@ -199,6 +199,8 @@ export interface SelfHealingOptions {
   autoRecoveryDispatcher?: AutoRecoveryDispatcher;
   /** Optional ChatStore for maintenance chat-retention cleanup. */
   chatStore?: ChatStore;
+  /** Optional MessageStore for maintenance mail-retention cleanup. */
+  messageStore?: MessageStore;
 }
 
 const APPROVED_TRIAGE_RECOVERY_GRACE_MS = 60_000;
@@ -971,6 +973,22 @@ export class SelfHealingManager {
             log.log(`Maintenance batch 1 step "cleanup-old-chats" succeeded — sessions=${sessionsDeleted} rooms=${roomsDeleted}`);
           },
         },
+        {
+          name: "cleanup-old-mail",
+          fn: async () => {
+            const value = Number(settings.mailAutoCleanupDays ?? 0);
+            if (!Number.isFinite(value) || value <= 0) {
+              log.log(`Skipping cleanup-old-mail: setting=${String(settings.mailAutoCleanupDays ?? 0)}`);
+              return;
+            }
+            if (!this.options.messageStore) {
+              log.log("Skipping cleanup-old-mail: messageStore unavailable");
+              return;
+            }
+            const { messagesDeleted } = this.options.messageStore.cleanupOldMessages(value * 86_400_000);
+            log.log(`Maintenance batch 1 step "cleanup-old-mail" succeeded — messagesDeleted=${messagesDeleted}`);
+          },
+        },
         { name: "checkpoint-wal", fn: () => Promise.resolve(this.checkpointWal()) },
         { name: "enforce-worktree-cap", fn: () => this.enforceWorktreeCap() },
       ];