feat(FN-4545): complete Step 4-5 — add message-delivery recovery handler wiring

Fusion-Task-Id: FN-4545 Fusion-Task-Lineage: 95108429-618e-4a6d-9fa9-7ac2596665a2
2026-05-14 20:41:12 -07:00
parent c37d3b404c
commit 50fe141f5b
4 changed files with 153 additions and 25 deletions
--- a/packages/engine/src/agent-tools.ts
+++ b/packages/engine/src/agent-tools.ts
@@ -23,6 +23,7 @@ import { createLogger } from "./logger.js";
 import { fetchWebContent, WebFetchError } from "./web-fetch.js";
 import type { RunAuditor } from "./run-audit.js";
 import { computeApprovalDedupeKey } from "./agent-action-gate.js";
+import { MessageDeliveryAutoRecoveryHandler } from "./auto-recovery-handlers/message-delivery.js";

 // ── Tool parameter schemas (canonical definitions) ────────────────────────

@@ -1830,7 +1831,15 @@ export function createDelegateTaskTool(
 * @param fromAgentId - The agent ID sending the message
 * @returns ToolDefinition for the `fn_send_message` tool
 */
-export function createSendMessageTool(messageStore: MessageStore, fromAgentId: string): ToolDefinition {
+export function createSendMessageTool(
+  messageStore: MessageStore,
+  fromAgentId: string,
+  options?: { autoRecovery?: ProjectSettings["autoRecovery"]; runAudit?: RunAuditor },
+): ToolDefinition {
+  const deliveryHandler = new MessageDeliveryAutoRecoveryHandler({
+    runAudit: options?.runAudit ?? { database: async () => {}, git: async () => {}, filesystem: async () => {} },
+  });
+
  return {
    name: "fn_send_message",
    label: "Send Message",
@@ -1841,7 +1850,6 @@ export function createSendMessageTool(messageStore: MessageStore, fromAgentId: s
    parameters: sendMessageParams,
    // eslint-disable-next-line @typescript-eslint/no-explicit-any
    execute: async (_id: string, params: Static<typeof sendMessageParams>, _signal?: any, _onUpdate?: any, _ctx?: any) => {
-      // Validate content length
      const content = params.content.trim();
      if (content.length === 0) {
        return {
@@ -1873,22 +1881,32 @@ export function createSendMessageTool(messageStore: MessageStore, fromAgentId: s
          };
        }

-        const message = messageStore.sendMessage({
-          fromId: fromAgentId,
-          fromType: "agent",
-          toId: recipient.id,
-          toType: recipient.type,
-          content,
-          type: messageType,
-          ...(replyToMessageId ? { metadata: { replyTo: { messageId: replyToMessageId } } } : {}),
-        });
+        const result = await deliveryHandler.runWithBoundedRetry({
+          run: async () => Promise.resolve(messageStore.sendMessage({
+            fromId: fromAgentId,
+            fromType: "agent",
+            toId: recipient.id,
+            toType: recipient.type,
+            content,
+            type: messageType,
+            ...(replyToMessageId ? { metadata: { replyTo: { messageId: replyToMessageId } } } : {}),
+          })),
+          correlation: { kind: "direct", fromAgentId, toId: recipient.id },
+        }, options?.autoRecovery ?? { mode: "deterministic-only", maxRetries: 3 });
+
+        if (result.outcome === "parked") {
+          return {
+            content: [{ type: "text" as const, text: `ERROR: Failed to send message: ${result.error.message}` }],
+            details: {},
+          };
+        }

        return {
          content: [{
            type: "text" as const,
-            text: `Message sent to ${recipient.id === DASHBOARD_USER_ID ? DASHBOARD_USER_ID : params.to_id} (ID: ${message.id})`,
+            text: `Message sent to ${recipient.id === DASHBOARD_USER_ID ? DASHBOARD_USER_ID : params.to_id} (ID: ${result.value.id})`,
          }],
-          details: { messageId: message.id },
+          details: { messageId: result.value.id },
        };
      } catch (err) {
        const errorMessage = err instanceof Error ? err.message : String(err);
@@ -2132,7 +2150,15 @@ export function createResearchTools(options: ResearchToolsOptions): ToolDefiniti
  return [runTool, listTool, getTool, cancelTool];
 }

-export function createPostRoomMessageTool(chatStore: ChatStore, fromAgentId: string): ToolDefinition {
+export function createPostRoomMessageTool(
+  chatStore: ChatStore,
+  fromAgentId: string,
+  options?: { autoRecovery?: ProjectSettings["autoRecovery"]; runAudit?: RunAuditor },
+): ToolDefinition {
+  const deliveryHandler = new MessageDeliveryAutoRecoveryHandler({
+    runAudit: options?.runAudit ?? { database: async () => {}, git: async () => {}, filesystem: async () => {} },
+  });
+
  return {
    name: "fn_post_room_message",
    label: "Post Room Message",
@@ -2173,17 +2199,28 @@ export function createPostRoomMessageTool(chatStore: ChatStore, fromAgentId: str
          };
        }

-        const message = chatStore.addRoomMessage(params.roomId, {
-          role: "assistant",
-          senderAgentId: fromAgentId,
-          content,
-          mentions: params.mentions ?? [],
-          ...(replyToMessageId ? { metadata: { replyToMessageId } } : {}),
-        });
+        const result = await deliveryHandler.runWithBoundedRetry({
+          run: async () => Promise.resolve(chatStore.addRoomMessage(params.roomId, {
+            role: "assistant",
+            senderAgentId: fromAgentId,
+            content,
+            mentions: params.mentions ?? [],
+            ...(replyToMessageId ? { metadata: { replyToMessageId } } : {}),
+          })),
+          correlation: { kind: "room", fromAgentId, roomId: params.roomId },
+        }, options?.autoRecovery ?? { mode: "deterministic-only", maxRetries: 3 });
+
+        if (result.outcome === "parked") {
+          return {
+            content: [{ type: "text" as const, text: `ERROR: Failed to post room message: ${result.error.message}` }],
+            details: {},
+            isError: true,
+          };
+        }

        return {
-          content: [{ type: "text" as const, text: `Room message posted to ${params.roomId} (ID: ${message.id})` }],
-          details: { messageId: message.id },
+          content: [{ type: "text" as const, text: `Room message posted to ${params.roomId} (ID: ${result.value.id})` }],
+          details: { messageId: result.value.id },
        };
      } catch (err) {
        const errorMessage = err instanceof Error ? err.message : String(err);
--- a/packages/engine/src/auto-recovery-handlers/message-delivery.ts
+++ b/packages/engine/src/auto-recovery-handlers/message-delivery.ts
@@ -0,0 +1,91 @@
+import type { AutoRecoverySettings } from "@fusion/core";
+import { createLogger, type Logger } from "../logger.js";
+import type { RunAuditor } from "../run-audit.js";
+
+const baseLog = createLogger("auto-recovery:message-delivery");
+
+export interface MessageDeliveryRecoveryDeps {
+  runAudit: RunAuditor;
+  logger?: Logger;
+  now?: () => number;
+  sleep?: (ms: number) => Promise<void>;
+  onRetryBurn?: (attempt: number) => Promise<void>;
+}
+
+export interface MessageDeliveryAttempt<T> {
+  run: () => Promise<T>;
+  correlation: { kind: "direct" | "room"; fromAgentId: string; toId?: string; roomId?: string };
+  runId?: string;
+}
+
+export class MessageDeliveryAutoRecoveryHandler {
+  constructor(private readonly deps: MessageDeliveryRecoveryDeps) {}
+
+  private get logger(): Logger {
+    return this.deps.logger ?? baseLog;
+  }
+
+  private isRetryMode(settings: AutoRecoverySettings): boolean {
+    return settings.mode === "programmatic" || settings.mode === "ai-assisted";
+  }
+
+  private isTransientError(error: Error): boolean {
+    const code = (error as Error & { code?: string }).code;
+    const message = error.message.toLowerCase();
+    return code === "SQLITE_BUSY"
+      || message.includes("sqlite_busy")
+      || message.includes("timeout")
+      || message.includes("econnreset")
+      || message.includes("eai_again");
+  }
+
+  async runWithBoundedRetry<T>(
+    attempt: MessageDeliveryAttempt<T>,
+    settings: AutoRecoverySettings,
+  ): Promise<{ outcome: "delivered"; value: T } | { outcome: "parked"; error: Error; attempts: number }> {
+    const maxAttempts = Math.max(1, settings.maxRetries ?? 3);
+    const shouldRetry = this.isRetryMode(settings);
+    const backoffs = [50, 200, 800];
+    let attempts = 0;
+
+    while (attempts < maxAttempts) {
+      attempts += 1;
+      try {
+        const value = await attempt.run();
+        if (attempts > 1) {
+          await this.deps.runAudit.database({
+            type: "message-delivery:retry-issued",
+            target: attempt.correlation.fromAgentId,
+            metadata: { correlation: attempt.correlation, attempt: attempts, mode: settings.mode },
+          });
+          if (this.deps.onRetryBurn) {
+            await this.deps.onRetryBurn(attempts);
+          }
+        }
+        return { outcome: "delivered", value };
+      } catch (error) {
+        const normalized = error instanceof Error ? error : new Error(String(error));
+        const transient = this.isTransientError(normalized);
+        if (!transient || !shouldRetry || attempts >= maxAttempts) {
+          await this.deps.runAudit.database({
+            type: "message-delivery:park",
+            target: attempt.correlation.fromAgentId,
+            metadata: {
+              correlation: attempt.correlation,
+              attempts,
+              errorMessage: normalized.message,
+              mode: settings.mode,
+            },
+          });
+          return { outcome: "parked", error: normalized, attempts };
+        }
+        const delayMs = backoffs[Math.min(attempts - 1, backoffs.length - 1)];
+        this.logger.warn(`message-delivery retrying ${attempt.correlation.kind} message for ${attempt.correlation.fromAgentId} attempt=${attempts + 1}`);
+        await (this.deps.sleep ? this.deps.sleep(delayMs) : new Promise((resolve) => setTimeout(resolve, delayMs)));
+      }
+    }
+
+    const exhausted = new Error("message delivery retry loop exhausted");
+    return { outcome: "parked", error: exhausted, attempts: maxAttempts };
+  }
+}
--- a/packages/engine/src/executor.ts
+++ b/packages/engine/src/executor.ts
@@ -3366,7 +3366,7 @@ export class TaskExecutor {
        ] : []),
        // Messaging tools — allows executor agents to send and receive messages.
        ...(this.options.messageStore && assignedAgentId ? [
-          createSendMessageTool(this.options.messageStore, assignedAgentId),
+          createSendMessageTool(this.options.messageStore, assignedAgentId, { autoRecovery: settings.autoRecovery, runAudit: audit }),
          createReadMessagesTool(this.options.messageStore, assignedAgentId),
        ] : []),
        // Add plugin tools from PluginRunner
--- a/packages/engine/src/step-session-executor.ts
+++ b/packages/engine/src/step-session-executor.ts
@@ -931,7 +931,7 @@ export class StepSessionExecutor {
          const messagingTools =
            this.options.messageStore && taskDetail.assignedAgentId
              ? [
-                  createSendMessageTool(this.options.messageStore, taskDetail.assignedAgentId),
+                  createSendMessageTool(this.options.messageStore, taskDetail.assignedAgentId, { autoRecovery: settings.autoRecovery }),
                  createReadMessagesTool(this.options.messageStore, taskDetail.assignedAgentId),
                ]
              : [];