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

FN-8185: use async persistence for reflection rework reads

Browse Source 
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) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-07-16 20:50:18 -07:00
parent 30aaab209c
commit b747f51f8b
2 changed files with 80 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
packages/engine/src/__tests__/agent-reflection.test.ts
View File

@@ -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({

50
packages/engine/src/agent-reflection.ts
View File

@@ -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<ReflectionMetrics> {
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<number> {
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<Array<{ reworkCount?: number }>>;
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 {