From b747f51f8b2e1e6bc3006b2e0782aa82b63da3a0 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 16 Jul 2026 20:50:18 -0700 Subject: [PATCH] FN-8185: use async persistence for reflection rework reads Resolve workflow rework metrics from the persisted production run. - Resolve workflow selections to definition-backed run IDs - Prefer asynchronous step-instance reads with legacy-store compatibility - Cover PostgreSQL, fallback, and degraded reflection metrics Files changed: .../engine/src/__tests__/agent-reflection.test.ts | 51 +++++++++++++++++++--- packages/engine/src/agent-reflection.ts | 50 ++++++++++++++------- 2 files changed, 80 insertions(+), 21 deletions(-) Fusion-Task-Id: FN-8185 Fusion-Task-Lineage: 2554d5f1-62b7-47a6-bad0-c00a7d649a2d Co-authored-by: Fusion (runfusion.ai) --- .../src/__tests__/agent-reflection.test.ts | 51 +++++++++++++++++-- packages/engine/src/agent-reflection.ts | 50 ++++++++++++------ 2 files changed, 80 insertions(+), 21 deletions(-) diff --git a/packages/engine/src/__tests__/agent-reflection.test.ts b/packages/engine/src/__tests__/agent-reflection.test.ts index bb7de76985..f50d09123d 100644 --- a/packages/engine/src/__tests__/agent-reflection.test.ts +++ b/packages/engine/src/__tests__/agent-reflection.test.ts @@ -588,28 +588,69 @@ describe("AgentReflectionService", () => { expect(reflection?.metrics.verificationScopeReason).toBeUndefined(); }); - it("aggregates workflow step rework cycles (RETHINK/rework) alongside recoveryRetryCount into retryReworkCount and durationDrivers", async () => { + it("aggregates backend-persisted workflow rework under the resolved definition run id", async () => { const { agentStore, taskStore, reflectionStore } = createMockDeps(); taskStore.getTask.mockResolvedValue(makeTask({ id: "FN-7528-rework", column: "done", recoveryRetryCount: 1, })); - taskStore.loadWorkflowRunStepInstances = vi.fn().mockReturnValue([ - { taskId: "FN-7528-rework", runId: "FN-7528-rework:run", foreachNodeId: "n1", stepIndex: 0, reworkCount: 2 }, - { taskId: "FN-7528-rework", runId: "FN-7528-rework:run", foreachNodeId: "n1", stepIndex: 1, reworkCount: 1 }, + taskStore.getTaskWorkflowSelectionAsync = vi.fn().mockResolvedValue({ + workflowId: "selected-workflow", + stepIds: [], + }); + taskStore.getWorkflowDefinition = vi.fn().mockResolvedValue({ id: "resolved-definition" }); + taskStore.loadWorkflowRunStepInstancesAsync = vi.fn().mockResolvedValue([ + { taskId: "FN-7528-rework", runId: "FN-7528-rework:resolved-definition", foreachNodeId: "n1", stepIndex: 0, reworkCount: 2 }, + { taskId: "FN-7528-rework", runId: "FN-7528-rework:resolved-definition", foreachNodeId: "n1", stepIndex: 1, reworkCount: 1 }, ]); const service = new AgentReflectionService({ agentStore, taskStore, reflectionStore, rootDir: tempRoot }); const reflection = await service.captureTaskPerformance("agent-1", "FN-7528-rework"); - expect(taskStore.loadWorkflowRunStepInstances).toHaveBeenCalledWith("FN-7528-rework", "FN-7528-rework:run"); + expect(taskStore.getWorkflowDefinition).toHaveBeenCalledWith("selected-workflow"); + expect(taskStore.loadWorkflowRunStepInstancesAsync).toHaveBeenCalledWith("FN-7528-rework", "FN-7528-rework:resolved-definition"); + expect(taskStore.loadWorkflowRunStepInstancesAsync).not.toHaveBeenCalledWith("FN-7528-rework", "FN-7528-rework:selected-workflow"); + expect(taskStore.loadWorkflowRunStepInstancesAsync).not.toHaveBeenCalledWith("FN-7528-rework", "FN-7528-rework:run"); // recoveryRetryCount(1) + workflowReworkCount(2+1=3) = 4 expect(reflection?.metrics.retryReworkCount).toBe(4); expect(reflection?.metrics.durationDrivers).toContain("retries:1"); expect(reflection?.metrics.durationDrivers).toContain("rework:3"); }); + it("uses the synchronous step-instance fallback for legacy stores under the resolved run id", async () => { + const { agentStore, taskStore, reflectionStore } = createMockDeps(); + taskStore.getTask.mockResolvedValue(makeTask({ id: "FN-7528-sync", column: "done" })); + taskStore.getTaskWorkflowSelection = vi.fn().mockReturnValue({ workflowId: "builtin:coding", stepIds: [] }); + taskStore.loadWorkflowRunStepInstances = vi.fn().mockReturnValue([ + { taskId: "FN-7528-sync", runId: "FN-7528-sync:builtin:coding", foreachNodeId: "n1", stepIndex: 0, reworkCount: 2 }, + ]); + + const service = new AgentReflectionService({ agentStore, taskStore, reflectionStore, rootDir: tempRoot }); + const reflection = await service.captureTaskPerformance("agent-1", "FN-7528-sync"); + + expect(taskStore.loadWorkflowRunStepInstances).toHaveBeenCalledWith("FN-7528-sync", "FN-7528-sync:builtin:coding"); + expect(reflection?.metrics.retryReworkCount).toBe(2); + expect(reflection?.metrics.durationDrivers).toContain("rework:2"); + }); + + it("silently omits workflow rework when no selection or step-instance reader is available", async () => { + const { agentStore, taskStore, reflectionStore } = createMockDeps(); + taskStore.getTask.mockResolvedValue(makeTask({ + id: "FN-7528-degraded", + column: "done", + recoveryRetryCount: 1, + })); + + const service = new AgentReflectionService({ agentStore, taskStore, reflectionStore, rootDir: tempRoot }); + const reflection = await service.captureTaskPerformance("agent-1", "FN-7528-degraded"); + + expect(reflection).not.toBeNull(); + expect(reflection?.metrics.retryReworkCount).toBe(1); + expect(reflection?.metrics.durationDrivers).toContain("retries:1"); + expect(reflection?.metrics.durationDrivers).not.toContain("rework:0"); + }); + it("classifies a broad/whole-suite verification command as not file-scoped, with a reason", async () => { const { agentStore, taskStore, reflectionStore } = createMockDeps(); taskStore.getTask.mockResolvedValue(makeTask({ diff --git a/packages/engine/src/agent-reflection.ts b/packages/engine/src/agent-reflection.ts index 12521e90f2..23930f0d54 100644 --- a/packages/engine/src/agent-reflection.ts +++ b/packages/engine/src/agent-reflection.ts @@ -220,7 +220,7 @@ export class AgentReflectionService { return null; } - const metrics = this.buildCapturedMetrics(taskId, task, outcome); + const metrics = await this.buildCapturedMetrics(taskId, task, outcome); const reflection = await this.reflectionStore.createReflection({ agentId, @@ -259,17 +259,19 @@ export class AgentReflectionService { * FNXC:AgentReflection 2026-07-04-00:00: * Code review (FN-7528) flagged that `retryReworkCount` only reflected `Task.recoveryRetryCount`, * silently dropping workflow step RETHINK/rework cycles tracked per-step-instance - * (`WorkflowRunStepInstance.reworkCount`, keyed by taskId+runId). `captureTaskPerformance` has no - * real runId threaded through, so we probe the same `${taskId}:run` fallback literal the executor - * itself falls back to when no runId is threaded (see executor.ts loadWorkflowRunStepInstances call - * sites) and sum reworkCount across every persisted instance row for that task. `retryReworkCount` - * is now `recoveryRetryCount + workflowReworkCount`; either driver is surfaced individually in - * `durationDrivers` (`retries:N` / `rework:N`) so the two causes stay distinguishable. + * (`WorkflowRunStepInstance.reworkCount`, keyed by taskId+runId). + * + * FNXC:AgentReflection 2026-07-16-00:00: + * The synchronous step-instance read returns no rows in PostgreSQL backend mode. Prefer its async + * sibling and mirror the executor's production `${taskId}:${definitionId}` run id: resolve the + * selection's workflowId to its definition id through awaited `getWorkflowDefinition`, never the + * legacy `${taskId}:run` literal. This keeps persisted RETHINK/rework cycles visible while missing + * selection, definition, or store capabilities still degrade to an unfabricated zero. */ - private buildCapturedMetrics(taskId: string, task: Task, outcome: "completed" | "failed"): ReflectionMetrics { + private async buildCapturedMetrics(taskId: string, task: Task, outcome: "completed" | "failed"): Promise { const durationMs = this.calculateDurationMs(task); const recoveryRetryCount = task.recoveryRetryCount ?? 0; - const workflowReworkCount = this.sumWorkflowStepReworkCount(taskId); + const workflowReworkCount = await this.sumWorkflowStepReworkCount(taskId); const retryReworkCount = recoveryRetryCount + workflowReworkCount; const touchedFiles = task.mergeDetails?.landedFiles ?? task.modifiedFiles; @@ -312,18 +314,34 @@ export class AgentReflectionService { } /** - * Sum `reworkCount` across every persisted `WorkflowRunStepInstance` row for this task (KTD-6), - * under the same `${taskId}:run` fallback runId literal used elsewhere when no real runId is - * threaded. Returns 0 (never fabricated) when the store lacks the method or the table/rows don't - * exist — additive bookkeeping, degrades silently like its call sites in executor.ts. + * Sum `reworkCount` for the executor's resolved production run. The async persistence read is + * required for PostgreSQL backend mode; synchronous stores retain the compatibility fallback. + * Missing capability or an unresolvable selection/definition returns 0 without inventing metrics. */ - private sumWorkflowStepReworkCount(taskId: string): number { + private async sumWorkflowStepReworkCount(taskId: string): Promise { const store = this.taskStore as unknown as { + getTaskWorkflowSelectionAsync?: (taskId: string) => Promise<{ workflowId: string; stepIds: string[] } | undefined>; + getTaskWorkflowSelection?: (taskId: string) => { workflowId: string; stepIds: string[] } | undefined; + getWorkflowDefinition?: (workflowId: string) => Promise<{ id: string } | undefined>; + loadWorkflowRunStepInstancesAsync?: (taskId: string, runId: string) => Promise>; loadWorkflowRunStepInstances?: (taskId: string, runId: string) => Array<{ reworkCount?: number }>; }; - if (typeof store.loadWorkflowRunStepInstances !== "function") return 0; + try { - const rows = store.loadWorkflowRunStepInstances(taskId, `${taskId}:run`); + const selection = typeof store.getTaskWorkflowSelectionAsync === "function" + ? await store.getTaskWorkflowSelectionAsync(taskId) + : store.getTaskWorkflowSelection?.(taskId); + if (!selection) return 0; + + const definition = selection.workflowId === "builtin:coding" + ? { id: "builtin:coding" } + : await store.getWorkflowDefinition?.(selection.workflowId); + if (!definition) return 0; + + const runId = `${taskId}:${definition.id}`; + const rows = typeof store.loadWorkflowRunStepInstancesAsync === "function" + ? await store.loadWorkflowRunStepInstancesAsync(taskId, runId) + : store.loadWorkflowRunStepInstances?.(taskId, runId); if (!Array.isArray(rows) || rows.length === 0) return 0; return rows.reduce((sum, row) => sum + (row.reworkCount ?? 0), 0); } catch {