root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(FN-4545): wire retry-burn logging and complete handler coverage

Browse Source 
Fusion-Task-Id: FN-4545
Fusion-Task-Lineage: 95108429-618e-4a6d-9fa9-7ac2596665a2
This commit is contained in:
Fusion
2026-05-14 21:07:03 -07:00
committed by gsxdsm
parent 05ecf1fb41
commit 2d782569d2
7 changed files with 84 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
packages/engine/src/__tests__/auto-recovery-contamination.test.ts
View File

@@ -24,6 +24,24 @@ describe("ContaminationAutoRecoveryHandler", () => {
expect(runAudit.database).toHaveBeenCalledWith(expect.objectContaining({ type: "contamination:retry-issued" }));
});
it("emits irreducible pause and skips retry for destructive ambiguity", async () => {
const taskStore = { moveTask: vi.fn(), updateTask: vi.fn() } as any;
const runAudit = { database: vi.fn(), git: vi.fn(), filesystem: vi.fn() } as any;
const handler = new ContaminationAutoRecoveryHandler({ taskStore, runAudit, repoDir: process.cwd() });
await handler.issueRetry({ class: "branch-cross-contamination", taskId: "FN-1", pausedReason: "branch-cross-contamination", evidence: { ownCommits: 1, foreignAttributedCommits: 1 } }, { action: "retry", rationale: "mode-programmatic", auditMetadata: {}, legacyPausedReason: "x" }, { task: { ...baseTask } as Task, retryCount: 1, settings: { mode: "programmatic", maxRetries: 3 } });
expect(taskStore.moveTask).not.toHaveBeenCalled();
expect(runAudit.database).toHaveBeenCalledWith(expect.objectContaining({ type: "contamination:irreducible-pause" }));
});
it("emits irreducible pause and skips retry when retry budget exhausted", async () => {
const taskStore = { moveTask: vi.fn(), updateTask: vi.fn() } as any;
const runAudit = { database: vi.fn(), git: vi.fn(), filesystem: vi.fn() } as any;
const handler = new ContaminationAutoRecoveryHandler({ taskStore, runAudit, repoDir: process.cwd() });
await handler.issueRetry({ class: "branch-cross-contamination", taskId: "FN-1", pausedReason: "branch-cross-contamination", evidence: { ownCommits: 0, foreignAttributedCommits: 2 } }, { action: "retry", rationale: "mode-programmatic", auditMetadata: {}, legacyPausedReason: "x" }, { task: { ...baseTask } as Task, retryCount: 3, settings: { mode: "programmatic", maxRetries: 3 } });
expect(taskStore.moveTask).not.toHaveBeenCalled();
expect(runAudit.database).toHaveBeenCalledWith(expect.objectContaining({ type: "contamination:irreducible-pause" }));
});
it("mode off does not call handler", async () => {
const issueRetry = vi.fn();
const dispatcher = new AutoRecoveryDispatcher({ taskStore: {} as any, auditEmitter: { database: vi.fn(), git: vi.fn(), filesystem: vi.fn() }, handlers: { issueRetry } });

31
packages/engine/src/__tests__/auto-recovery-message-delivery.test.ts
View File

@@ -24,6 +24,16 @@ describe("MessageDeliveryAutoRecoveryHandler", () => {
expect(runAudit.database).toHaveBeenCalledWith(expect.objectContaining({ type: "message-delivery:retry-issued" }));
});
it("parks after transient retries are exhausted", async () => {
const runAudit = { database: vi.fn(), git: vi.fn(), filesystem: vi.fn() } as any;
const run = vi.fn().mockRejectedValue(Object.assign(new Error("SQLITE_BUSY"), { code: "SQLITE_BUSY" }));
const handler = new MessageDeliveryAutoRecoveryHandler({ runAudit, sleep: vi.fn(async () => {}) });
const result = await handler.runWithBoundedRetry({ run, correlation: { kind: "room", fromAgentId: "a1", roomId: "r1" } }, { mode: "programmatic", maxRetries: 3 });
expect(result.outcome).toBe("parked");
expect(run).toHaveBeenCalledTimes(3);
expect(runAudit.database).toHaveBeenCalledWith(expect.objectContaining({ type: "message-delivery:park", metadata: expect.objectContaining({ attempts: 3 }) }));
});
it("parks permanent errors without retry", async () => {
const runAudit = { database: vi.fn(), git: vi.fn(), filesystem: vi.fn() } as any;
const run = vi.fn().mockRejectedValue(new Error("Room membership required"));
@@ -33,4 +43,25 @@ describe("MessageDeliveryAutoRecoveryHandler", () => {
expect(run).toHaveBeenCalledTimes(1);
expect(runAudit.database).toHaveBeenCalledWith(expect.objectContaining({ type: "message-delivery:park" }));
});
it("mode off does not retry transient failures", async () => {
const runAudit = { database: vi.fn(), git: vi.fn(), filesystem: vi.fn() } as any;
const run = vi.fn().mockRejectedValue(Object.assign(new Error("SQLITE_BUSY"), { code: "SQLITE_BUSY" }));
const handler = new MessageDeliveryAutoRecoveryHandler({ runAudit, sleep: vi.fn(async () => {}) });
const result = await handler.runWithBoundedRetry({ run, correlation: { kind: "direct", fromAgentId: "a1", toId: "a2" } }, { mode: "off", maxRetries: 3 });
expect(result.outcome).toBe("parked");
expect(run).toHaveBeenCalledTimes(1);
expect(runAudit.database).toHaveBeenCalledWith(expect.objectContaining({ type: "message-delivery:park", metadata: expect.objectContaining({ mode: "off" }) }));
});
it("records direct and room correlation metadata", async () => {
const runAudit = { database: vi.fn(), git: vi.fn(), filesystem: vi.fn() } as any;
const handler = new MessageDeliveryAutoRecoveryHandler({ runAudit, sleep: vi.fn(async () => {}) });
await handler.runWithBoundedRetry({ run: vi.fn().mockRejectedValue(new Error("recipient missing")), correlation: { kind: "direct", fromAgentId: "a1", toId: "a2" } }, { mode: "programmatic", maxRetries: 3 });
await handler.runWithBoundedRetry({ run: vi.fn().mockRejectedValue(new Error("room missing")), correlation: { kind: "room", fromAgentId: "a1", roomId: "r1" } }, { mode: "programmatic", maxRetries: 3 });
expect(runAudit.database).toHaveBeenCalledWith(expect.objectContaining({ metadata: expect.objectContaining({ correlation: expect.objectContaining({ kind: "direct", toId: "a2" }) }) }));
expect(runAudit.database).toHaveBeenCalledWith(expect.objectContaining({ metadata: expect.objectContaining({ correlation: expect.objectContaining({ kind: "room", roomId: "r1" }) }) }));
});
});

31
packages/engine/src/agent-tools.ts
View File

@@ -24,6 +24,7 @@ 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";
import { recordRetry } from "./retry-burned-logger.js";
// ── Tool parameter schemas (canonical definitions) ────────────────────────
@@ -1834,7 +1835,7 @@ export function createDelegateTaskTool(
export function createSendMessageTool(
messageStore: MessageStore,
fromAgentId: string,
options?: { autoRecovery?: ProjectSettings["autoRecovery"]; runAudit?: RunAuditor },
options?: { autoRecovery?: ProjectSettings["autoRecovery"]; runAudit?: RunAuditor; taskStore?: TaskStore; settings?: Settings },
): ToolDefinition {
const deliveryHandler = new MessageDeliveryAutoRecoveryHandler({
runAudit: options?.runAudit ?? { database: async () => {}, git: async () => {}, filesystem: async () => {} },
@@ -1892,7 +1893,17 @@ export function createSendMessageTool(
...(replyToMessageId ? { metadata: { replyTo: { messageId: replyToMessageId } } } : {}),
})),
correlation: { kind: "direct", fromAgentId, toId: recipient.id },
}, options?.autoRecovery ?? { mode: "deterministic-only", maxRetries: 3 });
}, options?.autoRecovery ?? { mode: "deterministic-only", maxRetries: 3 }, async () => {
const taskId = _ctx?.taskId as string | undefined;
if (!taskId || !options?.taskStore || !options.settings) {
return;
}
const task = await options.taskStore.getTask(taskId);
if (!task) {
return;
}
await recordRetry({ store: options.taskStore, settings: options.settings, task, category: "messageDelivery", role: "executor", agentId: fromAgentId });
});
if (result.outcome === "parked") {
return {
@@ -2153,7 +2164,7 @@ export function createResearchTools(options: ResearchToolsOptions): ToolDefiniti
export function createPostRoomMessageTool(
chatStore: ChatStore,
fromAgentId: string,
options?: { autoRecovery?: ProjectSettings["autoRecovery"]; runAudit?: RunAuditor },
options?: { autoRecovery?: ProjectSettings["autoRecovery"]; runAudit?: RunAuditor; taskStore?: TaskStore; settings?: Settings },
): ToolDefinition {
const deliveryHandler = new MessageDeliveryAutoRecoveryHandler({
runAudit: options?.runAudit ?? { database: async () => {}, git: async () => {}, filesystem: async () => {} },
@@ -2166,7 +2177,7 @@ export function createPostRoomMessageTool(
"Post a message to a room you are a member of. Room membership is enforced before posting, " +
"so only reply when the room content is relevant to your role or identity.",
parameters: postRoomMessageParams,
execute: async (_id: string, params: Static<typeof postRoomMessageParams>) => {
execute: async (_id: string, params: Static<typeof postRoomMessageParams>, _signal?: any, _onUpdate?: any, _ctx?: any) => {
const content = params.content.trim();
if (content.length === 0) {
return {
@@ -2208,7 +2219,17 @@ export function createPostRoomMessageTool(
...(replyToMessageId ? { metadata: { replyToMessageId } } : {}),
})),
correlation: { kind: "room", fromAgentId, roomId: params.roomId },
}, options?.autoRecovery ?? { mode: "deterministic-only", maxRetries: 3 });
}, options?.autoRecovery ?? { mode: "deterministic-only", maxRetries: 3 }, async () => {
const taskId = _ctx?.taskId as string | undefined;
if (!taskId || !options?.taskStore || !options.settings) {
return;
}
const task = await options.taskStore.getTask(taskId);
if (!task) {
return;
}
await recordRetry({ store: options.taskStore, settings: options.settings, task, category: "messageDelivery", role: "executor", agentId: fromAgentId });
});
if (result.outcome === "parked") {
return {

5
packages/engine/src/auto-recovery-handlers/message-delivery.ts
View File

@@ -42,6 +42,7 @@ export class MessageDeliveryAutoRecoveryHandler {
async runWithBoundedRetry<T>(
attempt: MessageDeliveryAttempt<T>,
settings: AutoRecoverySettings,
onRetryBurn?: () => Promise<void>,
): Promise<{ outcome: "delivered"; value: T } | { outcome: "parked"; error: Error; attempts: number }> {
const maxAttempts = Math.max(1, settings.maxRetries ?? 3);
const shouldRetry = this.isRetryMode(settings);
@@ -58,7 +59,9 @@ export class MessageDeliveryAutoRecoveryHandler {
target: attempt.correlation.fromAgentId,
metadata: { correlation: attempt.correlation, attempt: attempts, mode: settings.mode },
});
if (this.deps.onRetryBurn) {
if (onRetryBurn) {
await onRetryBurn();
} else if (this.deps.onRetryBurn) {
await this.deps.onRetryBurn(attempts);
}
}

2
packages/engine/src/executor.ts
View File

@@ -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, { autoRecovery: settings.autoRecovery, runAudit: audit }),
createSendMessageTool(this.options.messageStore, assignedAgentId, { autoRecovery: settings.autoRecovery, runAudit: audit, taskStore: this.store, settings }),
createReadMessagesTool(this.options.messageStore, assignedAgentId),
] : []),
// Add plugin tools from PluginRunner

4
packages/engine/src/retry-burned-logger.ts
View File

@@ -19,7 +19,8 @@ type RetryCategory =
| "workflowStep"
| "verification"
| "postReviewFix"
| "mergeConflict";
| "mergeConflict"
| "messageDelivery";
const CATEGORY_COLUMN: Record<RetryCategory, keyof TaskDetail> = {
branchConflict: "branchConflictRecoveryCount",
@@ -32,6 +33,7 @@ const CATEGORY_COLUMN: Record<RetryCategory, keyof TaskDetail> = {
verification: "verificationFailureCount",
postReviewFix: "postReviewFixCount",
mergeConflict: "mergeConflictBounceCount",
messageDelivery: "recoveryRetryCount",
};
const CATEGORY_CAP = (category: RetryCategory, settings: Settings): number | undefined => {

2
packages/engine/src/step-session-executor.ts
View File

@@ -931,7 +931,7 @@ export class StepSessionExecutor {
const messagingTools =
this.options.messageStore && taskDetail.assignedAgentId
? [
createSendMessageTool(this.options.messageStore, taskDetail.assignedAgentId, { autoRecovery: settings.autoRecovery }),
createSendMessageTool(this.options.messageStore, taskDetail.assignedAgentId, { autoRecovery: settings.autoRecovery, taskStore: this.options.store!, settings }),
createReadMessagesTool(this.options.messageStore, taskDetail.assignedAgentId),
]
: [];