From 0897b2a0bf64d46f45e6cccb452ed179c0db5e57 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 12 Jun 2026 11:10:12 -0700 Subject: [PATCH] FN-6281: add bounded graph resume retries Add a narrow persisted retry path for transient workflow-graph failures during resume after restart or unpause. - Persist graphResumeRetryCount with schema migration, store round-tripping, manual retry reset, and task serialization coverage. - Retry only no-progress execute-seam graph failures immediately following resume markers, then clear transient status/error and re-run within a capped budget. - Preserve terminal graph failure handling for explicit reasons, durable task failures, completed progress, non-resume cases, and exhausted retry budgets. - Document the resume-limbo retry contract and add a patch changeset for @runfusion/fusion. Files changed: .changeset/fn-6281-graph-resume-retry.md | 5 + docs/architecture.md | 1 + packages/core/src/__tests__/db-migrate.test.ts | 30 +-- packages/core/src/__tests__/db.test.ts | 44 ++--- packages/core/src/__tests__/goals-schema.test.ts | 2 +- packages/core/src/__tests__/insight-store.test.ts | 10 +- .../core/src/__tests__/manual-retry-reset.test.ts | 1 + .../src/__tests__/merge-request-record.test.ts | 2 +- packages/core/src/__tests__/mission-store.test.ts | 2 +- packages/core/src/__tests__/run-audit.test.ts | 4 +- .../core/src/__tests__/store-merge-queue.test.ts | 2 +- .../core/src/__tests__/store-persistence.test.ts | 22 +++ packages/core/src/__tests__/task-documents.test.ts | 2 +- packages/core/src/db.ts | 10 +- packages/core/src/manual-retry-reset.ts | 1 + packages/core/src/store.ts | 14 +- packages/core/src/types.ts | 5 + .../engine/src/__tests__/executor-recovery.test.ts | 201 +++++++++++++++++++++ packages/engine/src/executor.ts | 54 ++++++ 19 files changed, 359 insertions(+), 53 deletions(-) Fusion-Task-Id: FN-6281 Fusion-Task-Lineage: f9061445-8624-465a-9255-c3d02eac1bdb --- .changeset/fn-6281-graph-resume-retry.md | 5 + docs/architecture.md | 1 + .../core/src/__tests__/db-migrate.test.ts | 30 +-- packages/core/src/__tests__/db.test.ts | 44 ++-- .../core/src/__tests__/goals-schema.test.ts | 2 +- .../core/src/__tests__/insight-store.test.ts | 10 +- .../src/__tests__/manual-retry-reset.test.ts | 1 + .../__tests__/merge-request-record.test.ts | 2 +- .../core/src/__tests__/mission-store.test.ts | 2 +- packages/core/src/__tests__/run-audit.test.ts | 4 +- .../src/__tests__/store-merge-queue.test.ts | 2 +- .../src/__tests__/store-persistence.test.ts | 22 ++ .../core/src/__tests__/task-documents.test.ts | 2 +- packages/core/src/db.ts | 10 +- packages/core/src/manual-retry-reset.ts | 1 + packages/core/src/store.ts | 14 +- packages/core/src/types.ts | 5 + .../src/__tests__/executor-recovery.test.ts | 201 ++++++++++++++++++ packages/engine/src/executor.ts | 54 +++++ 19 files changed, 359 insertions(+), 53 deletions(-) create mode 100644 .changeset/fn-6281-graph-resume-retry.md diff --git a/.changeset/fn-6281-graph-resume-retry.md b/.changeset/fn-6281-graph-resume-retry.md new file mode 100644 index 0000000000..6af8e408c0 --- /dev/null +++ b/.changeset/fn-6281-graph-resume-retry.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Add a bounded persisted auto-retry for transient workflow-graph resume failures after engine restart or unpause, while preserving terminal failures for genuine graph errors. diff --git a/docs/architecture.md b/docs/architecture.md index e7badf70a9..85efeae024 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1269,6 +1269,7 @@ The columns/traits track moved *board* policy (transitions, capacity, hold, merg - A `parse-steps` node reads a workflow-declared **artifact** (PROMPT.md is just the default workflow's declared `step-source` artifact) and runs a registry **parser** (`step-headings`, `json-steps`, or a plugin-contributed parser) to write `Task.steps[]`. It is the only graph-side step-list writer and must dominate any `foreach`. Parsers fail closed to a routable `outcome:parse-error`. - A `foreach(source:"task-steps")` node instantiates an inline template subgraph once per planned step, with `mode` (sequential/parallel) and `isolation` (shared/worktree) as explicit axes and per-instance run-state pinned + persisted for crash-safe resume. +- Resume-limbo graph failures are retried only through a narrow persisted counter (`Task.graphResumeRetryCount`, max 2). The executor classifies a failure as transient only when it happens immediately after the engine restart/unpause resume log marker, reports no graph `reason`, has no completed step progress, and the task has no durable `lastError`/`failureReason`; it clears transient `status`/`error`, logs the auto-retry, and schedules one more graph execution. Any explicit graph reason, completed step progress, durable task error, missing resume marker, or exhausted counter remains a genuine `status:"failed"` disposition and goes to review handoff, preserving the FN-5704 anti-loop contract. - A `step-review` node surfaces reviewer verdicts (APPROVE/REVISE/RETHINK/UNAVAILABLE) as outcome edges; `rework` edges (the only legal graph cycles, bounded per instance) route REVISE/RETHINK back to `step-execute`, with RETHINK traversal triggering the reset seam. - A `code` node runs sandboxed TypeScript (esbuild + child process, clamped timeout, no store handle) for arbitrary computed routing/field logic — the same trust tier as project-local script steps. diff --git a/packages/core/src/__tests__/db-migrate.test.ts b/packages/core/src/__tests__/db-migrate.test.ts index b2f321eae4..9b77725ae8 100644 --- a/packages/core/src/__tests__/db-migrate.test.ts +++ b/packages/core/src/__tests__/db-migrate.test.ts @@ -715,7 +715,7 @@ describe("schema migration", () => { const row = db.prepare("SELECT deletedAt FROM tasks WHERE id = 'FN-legacy'").get() as { deletedAt: string | null }; expect(row.deletedAt).toBeNull(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -748,7 +748,7 @@ describe("schema migration", () => { { id: "WS-001", mode: "prompt", gateMode: "advisory" }, { id: "WS-002", mode: "script", gateMode: "advisory" }, ]); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -798,7 +798,7 @@ describe("schema migration", () => { reviewerContextRetryCount: 0, reviewerFallbackRetryCount: 0, }); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -827,7 +827,7 @@ describe("schema migration", () => { const columns = db.prepare("PRAGMA table_info(milestones)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("acceptanceCriteria"); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -868,7 +868,7 @@ describe("schema migration", () => { const missionColumns = db.prepare("PRAGMA table_info(missions)").all() as Array<{ name: string }>; expect(missionColumns.map((column) => column.name)).toContain("autoMerge"); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -902,7 +902,7 @@ describe("schema migration", () => { { id: "WS-002", mode: "script", enabled: 1, gateMode: "advisory" }, { id: "WS-003", mode: "prompt", enabled: 0, gateMode: "advisory" }, ]); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -939,7 +939,7 @@ describe("schema migration", () => { const indexes = db.prepare("PRAGMA index_list(mission_goals)").all() as Array<{ name: string }>; expect(indexes.some((index) => index.name === "idxMissionGoalsGoalId")).toBe(true); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -1000,7 +1000,7 @@ describe("schema migration", () => { expect(customFieldsColumn).toBeDefined(); expect(customFieldsColumn?.dflt_value).toBe("'{}'"); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -1038,7 +1038,7 @@ describe("schema migration", () => { const indexes = db.prepare("PRAGMA index_list(workflow_settings)").all() as Array<{ name: string }>; expect(indexes.some((index) => index.name === "idx_workflow_settings_project")).toBe(true); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -1120,7 +1120,7 @@ describe("schema migration", () => { expect(indexNames).toContain("idx_cli_sessions_chatSessionId"); expect(indexNames).toContain("idx_cli_sessions_project_state"); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -1152,7 +1152,7 @@ describe("schema migration", () => { .all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("cliExecutorAdapterId"); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -1162,7 +1162,7 @@ describe("schema migration", () => { const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table'").all() as Array<{ name: string }>; expect(tables.map((row) => row.name)).toContain("cli_sessions"); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -1219,20 +1219,20 @@ describe("schema migration", () => { .get() as { migrated_fragment_id: string | null }; expect(stepRow.migrated_fragment_id).toBeNull(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); it("migration 109 is idempotent on re-init", () => { const db = new Database(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); // Re-open the same on-disk DB: already at 109, the 109 block must be a no-op. const reopened = new Database(fusionDir); reopened.init(); - expect(reopened.getSchemaVersion()).toBe(115); + expect(reopened.getSchemaVersion()).toBe(116); const workflowColumns = reopened.prepare("PRAGMA table_info(workflows)").all() as Array<{ name: string }>; expect(workflowColumns.filter((c) => c.name === "kind")).toHaveLength(1); const stepColumns = reopened.prepare("PRAGMA table_info(workflow_steps)").all() as Array<{ name: string }>; diff --git a/packages/core/src/__tests__/db.test.ts b/packages/core/src/__tests__/db.test.ts index a3fa334ad3..fb30e4fcbf 100644 --- a/packages/core/src/__tests__/db.test.ts +++ b/packages/core/src/__tests__/db.test.ts @@ -334,7 +334,7 @@ describe("Database", () => { }); it("seeds schema version", () => { - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); }); it("includes tokenUsageCacheWriteTokens on freshly initialized tasks table", () => { @@ -393,7 +393,7 @@ describe("Database", () => { it("is idempotent - calling init() twice does not fail", () => { expect(() => db.init()).not.toThrow(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); }); it("does not overwrite existing config on re-init", () => { // Update the config @@ -1463,7 +1463,7 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 29 (includes v1→v2 through v26→v29) - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1488,15 +1488,15 @@ describe("schema migrations", () => { const db = new Database(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); // Re-init should not fail db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); // Re-init should not fail db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); db.close(); }); @@ -1531,7 +1531,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; expect(cols.map((col) => col.name)).toContain("priority"); @@ -1572,7 +1572,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const colNames = cols.map((col) => col.name); @@ -1644,7 +1644,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const colNames = cols.map((col) => col.name); @@ -1884,7 +1884,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); const cols = db.prepare("PRAGMA table_info(chat_messages)").all() as Array<{ name: string }>; expect(cols.map((col) => col.name)).toContain("attachments"); @@ -1958,7 +1958,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'agentRatings'").all() as Array<{ name: string }>; expect(tables).toEqual([{ name: "agentRatings" }]); @@ -1982,7 +1982,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'mission_events'").all() as Array<{ name: string }>; expect(tables).toEqual([{ name: "mission_events" }]); @@ -2086,7 +2086,7 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 29 - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -2305,7 +2305,7 @@ describe("schema migrations", () => { localDb.init(); - expect(localDb.getSchemaVersion()).toBe(115); + expect(localDb.getSchemaVersion()).toBe(116); const columns = localDb.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("tokenUsageCacheWriteTokens"); @@ -2616,7 +2616,7 @@ describe("createDatabase factory", () => { const db = createDatabase(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); expect(db.getLastModified()).toBeGreaterThan(0); db.close(); @@ -2770,7 +2770,7 @@ describe("migration v77 task token budget columns", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(115); + expect(migrated.getSchemaVersion()).toBe(116); const rows = migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const names = new Set(rows.map((row) => row.name)); expect(names.has("tokenBudgetSoftAlertedAt")).toBe(true); @@ -2801,7 +2801,7 @@ describe("migration v106 adds tasks.transitionPending (FN-1417)", () => { const fresh = new Database(fusion); try { fresh.init(); - expect(fresh.getSchemaVersion()).toBe(115); + expect(fresh.getSchemaVersion()).toBe(116); const names = new Set( (fresh.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>).map((r) => r.name), ); @@ -2829,7 +2829,7 @@ describe("migration v106 adds tasks.transitionPending (FN-1417)", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(115); + expect(migrated.getSchemaVersion()).toBe(116); const names = new Set( (migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>).map((r) => r.name), ); @@ -2855,7 +2855,7 @@ describe("migration v107 adds workflow_run_branches + index (FN-1417)", () => { const fresh = new Database(fusion); try { fresh.init(); - expect(fresh.getSchemaVersion()).toBe(115); + expect(fresh.getSchemaVersion()).toBe(116); const table = fresh .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'workflow_run_branches'") .get() as { name: string } | undefined; @@ -2889,7 +2889,7 @@ describe("migration v107 adds workflow_run_branches + index (FN-1417)", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(115); + expect(migrated.getSchemaVersion()).toBe(116); const table = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'workflow_run_branches'") .get() as { name: string } | undefined; @@ -2930,7 +2930,7 @@ describe("migration v67 drops orphan project auth tables", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(115); + expect(migrated.getSchemaVersion()).toBe(116); const tables = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; @@ -2957,7 +2957,7 @@ describe("migration v67 drops orphan project auth tables", () => { try { fresh.init(); - expect(fresh.getSchemaVersion()).toBe(115); + expect(fresh.getSchemaVersion()).toBe(116); const tables = fresh .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; diff --git a/packages/core/src/__tests__/goals-schema.test.ts b/packages/core/src/__tests__/goals-schema.test.ts index 5ad25567d1..0f55f5d80f 100644 --- a/packages/core/src/__tests__/goals-schema.test.ts +++ b/packages/core/src/__tests__/goals-schema.test.ts @@ -91,6 +91,6 @@ describe("goals schema", () => { }); it("reports schema version 101", () => { - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); }); }); diff --git a/packages/core/src/__tests__/insight-store.test.ts b/packages/core/src/__tests__/insight-store.test.ts index cd834d81a9..467dfa3f79 100644 --- a/packages/core/src/__tests__/insight-store.test.ts +++ b/packages/core/src/__tests__/insight-store.test.ts @@ -1000,7 +1000,7 @@ describe("Migration: pre-33 DB upgrade", () => { // Step 1: Create a fresh database at v33 (runs all migrations up to 33) const db1 = createDatabase(legacyDir); db1.init(); - expect(db1.getSchemaVersion()).toBe(115); + expect(db1.getSchemaVersion()).toBe(116); db1.close(); // Step 2: Manually downgrade to version 32 and drop insight tables @@ -1035,7 +1035,7 @@ describe("Migration: pre-33 DB upgrade", () => { expect(tableNamesBefore).not.toContain("project_insight_runs"); // Now run init — this triggers the v32→v33 migration db3.init(); - expect(db3.getSchemaVersion()).toBe(115); + expect(db3.getSchemaVersion()).toBe(116); // Step 4: Verify insight tables exist after migration const tablesAfter = db3.prepare( @@ -1066,12 +1066,12 @@ describe("Migration: pre-33 DB upgrade", () => { try { const db1 = createDatabase(testDir); db1.init(); - expect(db1.getSchemaVersion()).toBe(115); + expect(db1.getSchemaVersion()).toBe(116); db1.close(); const db2 = createDatabase(testDir); expect(() => db2.init()).not.toThrow(); - expect(db2.getSchemaVersion()).toBe(115); + expect(db2.getSchemaVersion()).toBe(116); db2.close(); } finally { rmSync(testDir, { recursive: true, force: true }); @@ -1085,7 +1085,7 @@ describe("Migration: pre-33 DB upgrade", () => { // Step 1: Create a fresh DB and run migrations const db1 = createDatabase(compatDir); db1.init(); - expect(db1.getSchemaVersion()).toBe(115); + expect(db1.getSchemaVersion()).toBe(116); // Step 2: Strip lifecycle and cancelledAt columns by recreating the // table without them. This simulates a DB that was created before the diff --git a/packages/core/src/__tests__/manual-retry-reset.test.ts b/packages/core/src/__tests__/manual-retry-reset.test.ts index 5b6a249fb1..3594e32b10 100644 --- a/packages/core/src/__tests__/manual-retry-reset.test.ts +++ b/packages/core/src/__tests__/manual-retry-reset.test.ts @@ -53,6 +53,7 @@ describe("buildManualRetryResetPatch", () => { for (const key of MANUAL_RETRY_RESET_COUNTER_KEYS) { expect(patch[key]).toBe(0); } + expect(patch.graphResumeRetryCount).toBe(0); }); it("includes all retry-summary counters in the reset key list", () => { diff --git a/packages/core/src/__tests__/merge-request-record.test.ts b/packages/core/src/__tests__/merge-request-record.test.ts index 19dea9f5f3..dd9089dfd3 100644 --- a/packages/core/src/__tests__/merge-request-record.test.ts +++ b/packages/core/src/__tests__/merge-request-record.test.ts @@ -38,7 +38,7 @@ describe("TaskStore merge request record + completion handoff marker", () => { .all() as Array<{ name: string }>; expect(tableRows).toEqual([{ name: "completion_handoff_markers" }, { name: "merge_requests" }]); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); }); it("upserts merge request records", async () => { diff --git a/packages/core/src/__tests__/mission-store.test.ts b/packages/core/src/__tests__/mission-store.test.ts index 0f3d2b09c5..62e62ac8c8 100644 --- a/packages/core/src/__tests__/mission-store.test.ts +++ b/packages/core/src/__tests__/mission-store.test.ts @@ -3746,7 +3746,7 @@ describe("MissionStore", () => { describe("Loop State & Validator Run Schema (v31)", () => { it("schema version is 101 after migration", () => { - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); }); it("mission_features table has loop state columns", () => { diff --git a/packages/core/src/__tests__/run-audit.test.ts b/packages/core/src/__tests__/run-audit.test.ts index 6e15da28c5..73edb60e08 100644 --- a/packages/core/src/__tests__/run-audit.test.ts +++ b/packages/core/src/__tests__/run-audit.test.ts @@ -583,8 +583,8 @@ describe("Run Audit", () => { expect(indexNames).toContain("idxRunAuditEventsTimestamp"); }); - it("schema version is bumped to 115", () => { - expect(db.getSchemaVersion()).toBe(115); + it("schema version is bumped to 116", () => { + expect(db.getSchemaVersion()).toBe(116); }); }); }); diff --git a/packages/core/src/__tests__/store-merge-queue.test.ts b/packages/core/src/__tests__/store-merge-queue.test.ts index 2db85dc1d2..2c77538a71 100644 --- a/packages/core/src/__tests__/store-merge-queue.test.ts +++ b/packages/core/src/__tests__/store-merge-queue.test.ts @@ -60,7 +60,7 @@ describe("TaskStore merge queue", () => { expect.arrayContaining(["idx_mergeQueue_lease_ready", "idx_mergeQueue_leaseExpiresAt"]), ); - expect(store.getDatabase().getSchemaVersion()).toBe(115); + expect(store.getDatabase().getSchemaVersion()).toBe(116); }); it("migrates a legacy v88 database and preserves task rows", async () => { diff --git a/packages/core/src/__tests__/store-persistence.test.ts b/packages/core/src/__tests__/store-persistence.test.ts index 41c5fa579c..ef62ce6f3c 100644 --- a/packages/core/src/__tests__/store-persistence.test.ts +++ b/packages/core/src/__tests__/store-persistence.test.ts @@ -59,6 +59,28 @@ describe("TaskStore", () => { }); }); + describe("graphResumeRetryCount persistence", () => { + it("defaults to zero and round-trips updateTask values", async () => { + const task = await harness.store().createTask({ description: "Graph retry counter task" }); + + expect((await harness.store().getTask(task.id)).graphResumeRetryCount).toBe(0); + + const updated = await harness.store().updateTask(task.id, { graphResumeRetryCount: 2 }); + expect(updated.graphResumeRetryCount).toBe(2); + expect((await harness.store().getTask(task.id)).graphResumeRetryCount).toBe(2); + }); + + it("clears graphResumeRetryCount with null", async () => { + const task = await harness.store().createTask({ description: "Graph retry clear task" }); + await harness.store().updateTask(task.id, { graphResumeRetryCount: 1 }); + + const cleared = await harness.store().updateTask(task.id, { graphResumeRetryCount: null }); + + expect(cleared.graphResumeRetryCount).toBeNull(); + expect((await harness.store().getTask(task.id)).graphResumeRetryCount).toBeUndefined(); + }); + }); + describe("agent taskId sync on reassignment", () => { it("reassignment clears the old agent taskId and sets the new agent taskId", async () => { harness.store().close(); diff --git a/packages/core/src/__tests__/task-documents.test.ts b/packages/core/src/__tests__/task-documents.test.ts index af7c770510..b79d8b63a3 100644 --- a/packages/core/src/__tests__/task-documents.test.ts +++ b/packages/core/src/__tests__/task-documents.test.ts @@ -51,7 +51,7 @@ describe("TaskStore task documents", () => { expect(tableNames.has("task_documents")).toBe(true); expect(tableNames.has("task_document_revisions")).toBe(true); - expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(116); const index = db .prepare( diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 25b640cabe..72b20e8b18 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -162,7 +162,7 @@ export function isFts5CorruptionError(error: unknown): boolean { // ── Schema Definition ──────────────────────────────────────────────── -const SCHEMA_VERSION = 115; +const SCHEMA_VERSION = 116; const TASKS_FTS_AUTOMERGE = 8; const TASKS_FTS_CRISISMERGE = 16; @@ -262,6 +262,7 @@ CREATE TABLE IF NOT EXISTS tasks ( mergeRetries INTEGER, workflowStepRetries INTEGER, resumeLimboCount INTEGER DEFAULT 0, + graphResumeRetryCount INTEGER DEFAULT 0, resumeLimboTipSha TEXT, resumeLimboStepSignature TEXT, recoveryRetryCount INTEGER, @@ -4689,6 +4690,13 @@ export class Database { }); } + // Migration 116: Bounded transient resume-after-restart graph retries. + if (version < 116) { + this.applyMigration(116, () => { + this.addColumnIfMissing("tasks", "graphResumeRetryCount", "INTEGER DEFAULT 0"); + }); + } + } /** diff --git a/packages/core/src/manual-retry-reset.ts b/packages/core/src/manual-retry-reset.ts index 938177ca26..903b89131f 100644 --- a/packages/core/src/manual-retry-reset.ts +++ b/packages/core/src/manual-retry-reset.ts @@ -5,6 +5,7 @@ export const IN_REVIEW_STALL_DEADLOCK_PAUSE_REASON = "in-review-stall-deadlock"; export const MANUAL_RETRY_RESET_COUNTER_KEYS = [ "stuckKillCount", "resumeLimboCount", + "graphResumeRetryCount", "recoveryRetryCount", "taskDoneRetryCount", "worktreeSessionRetryCount", diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 30f898c534..728d8d2a96 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -203,6 +203,7 @@ interface TaskRow { workflowStepRetries: number | null; stuckKillCount: number | null; resumeLimboCount: number | null; + graphResumeRetryCount: number | null; resumeLimboTipSha: string | null; resumeLimboStepSignature: string | null; postReviewFixCount: number | null; @@ -348,6 +349,7 @@ const TASK_COLUMN_DESCRIPTORS: TaskColumnDescriptor[] = [ defineTaskColumn("workflowStepRetries", (task) => task.workflowStepRetries ?? null), defineTaskColumn("stuckKillCount", (task) => task.stuckKillCount ?? 0), defineTaskColumn("resumeLimboCount", (task) => task.resumeLimboCount ?? 0), + defineTaskColumn("graphResumeRetryCount", (task) => task.graphResumeRetryCount === undefined ? 0 : task.graphResumeRetryCount), defineTaskColumn("resumeLimboTipSha", (task) => task.resumeLimboTipSha ?? null), defineTaskColumn("resumeLimboStepSignature", (task) => task.resumeLimboStepSignature ?? null), defineTaskColumn("postReviewFixCount", (task) => task.postReviewFixCount ?? 0), @@ -1933,6 +1935,7 @@ export class TaskStore extends EventEmitter { workflowStepRetries: row.workflowStepRetries ?? undefined, stuckKillCount: row.stuckKillCount ?? undefined, resumeLimboCount: row.resumeLimboCount ?? undefined, + graphResumeRetryCount: row.graphResumeRetryCount ?? undefined, resumeLimboTipSha: row.resumeLimboTipSha || undefined, resumeLimboStepSignature: row.resumeLimboStepSignature || undefined, postReviewFixCount: row.postReviewFixCount ?? undefined, @@ -2447,7 +2450,7 @@ export class TaskStore extends EventEmitter { "modelPresetId", "modelProvider", "modelId", "validatorModelProvider", "validatorModelId", "planningModelProvider", "planningModelId", - "mergeRetries", "workflowStepRetries", "stuckKillCount", "resumeLimboCount", "resumeLimboTipSha", "resumeLimboStepSignature", "postReviewFixCount", "recoveryRetryCount", "taskDoneRetryCount", "worktreeSessionRetryCount", "completionHandoffLimboRecoveryCount", "verificationFailureCount", "mergeConflictBounceCount", "mergeAuditBounceCount", "mergeTransientRetryCount", "branchConflictRecoveryCount", "reviewerContextRetryCount", "reviewerFallbackRetryCount", "nextRecoveryAt", + "mergeRetries", "workflowStepRetries", "stuckKillCount", "resumeLimboCount", "graphResumeRetryCount", "resumeLimboTipSha", "resumeLimboStepSignature", "postReviewFixCount", "recoveryRetryCount", "taskDoneRetryCount", "worktreeSessionRetryCount", "completionHandoffLimboRecoveryCount", "verificationFailureCount", "mergeConflictBounceCount", "mergeAuditBounceCount", "mergeTransientRetryCount", "branchConflictRecoveryCount", "reviewerContextRetryCount", "reviewerFallbackRetryCount", "nextRecoveryAt", "error", "summary", "thinkingLevel", "executionMode", "tokenUsageInputTokens", "tokenUsageOutputTokens", "tokenUsageCachedTokens", "tokenUsageCacheWriteTokens", "tokenUsageTotalTokens", "tokenUsageFirstUsedAt", "tokenUsageLastUsedAt", "tokenBudgetSoftAlertedAt", "tokenBudgetHardAlertedAt", "tokenBudgetOverride", "createdAt", "updatedAt", "columnMovedAt", "firstExecutionAt", "cumulativeActiveMs", "executionStartedAt", "executionCompletedAt", @@ -2496,7 +2499,7 @@ export class TaskStore extends EventEmitter { "modelPresetId", "modelProvider", "modelId", "validatorModelProvider", "validatorModelId", "planningModelProvider", "planningModelId", - "mergeRetries", "workflowStepRetries", "stuckKillCount", "resumeLimboCount", "resumeLimboTipSha", "resumeLimboStepSignature", "postReviewFixCount", "recoveryRetryCount", "taskDoneRetryCount", "worktreeSessionRetryCount", "completionHandoffLimboRecoveryCount", "verificationFailureCount", "mergeConflictBounceCount", "mergeAuditBounceCount", "mergeTransientRetryCount", "branchConflictRecoveryCount", "reviewerContextRetryCount", "reviewerFallbackRetryCount", "nextRecoveryAt", + "mergeRetries", "workflowStepRetries", "stuckKillCount", "resumeLimboCount", "graphResumeRetryCount", "resumeLimboTipSha", "resumeLimboStepSignature", "postReviewFixCount", "recoveryRetryCount", "taskDoneRetryCount", "worktreeSessionRetryCount", "completionHandoffLimboRecoveryCount", "verificationFailureCount", "mergeConflictBounceCount", "mergeAuditBounceCount", "mergeTransientRetryCount", "branchConflictRecoveryCount", "reviewerContextRetryCount", "reviewerFallbackRetryCount", "nextRecoveryAt", "error", "summary", "thinkingLevel", "executionMode", "tokenUsageInputTokens", "tokenUsageOutputTokens", "tokenUsageCachedTokens", "tokenUsageCacheWriteTokens", "tokenUsageTotalTokens", "tokenUsageFirstUsedAt", "tokenUsageLastUsedAt", "tokenBudgetSoftAlertedAt", "tokenBudgetHardAlertedAt", "tokenBudgetOverride", "createdAt", "updatedAt", "columnMovedAt", "firstExecutionAt", "cumulativeActiveMs", "executionStartedAt", "executionCompletedAt", @@ -7498,7 +7501,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} async updateTask( id: string, - updates: { title?: string; description?: string; priority?: TaskPriority | null; prompt?: string; worktree?: string | null; status?: string | null; dependencies?: string[]; steps?: import("./types.js").TaskStep[]; customFields?: Record; currentStep?: number; blockedBy?: string | null; overlapBlockedBy?: string | null; assignedAgentId?: string | null; pausedByAgentId?: string | null; pausedReason?: string | null; tokenBudgetSoftAlertedAt?: string | null; worktrunkFallbackAlertedAt?: string | null; worktrunkFailure?: import("./types.js").Task["worktrunkFailure"] | null; tokenBudgetHardAlertedAt?: string | null; tokenBudgetOverride?: import("./types.js").TaskTokenBudgetOverride | null; dispatchStormCount?: number | null; lastDispatchAt?: string | null; assigneeUserId?: string | null; scopeOverride?: boolean | null; scopeOverrideReason?: string | null; scopeAutoWiden?: string[] | null; nodeId?: string | null; effectiveNodeId?: string | null; effectiveNodeSource?: string | null; checkedOutBy?: string | null; checkedOutAt?: string | null; checkoutNodeId?: string | null; checkoutRunId?: string | null; checkoutLeaseRenewedAt?: string | null; checkoutLeaseEpoch?: number | null; paused?: boolean; baseBranch?: string | null; autoMerge?: boolean | null; branch?: string | null; executionStartBranch?: string | null; baseCommitSha?: string | null; size?: "S" | "M" | "L"; reviewLevel?: number; executionMode?: import("./types.js").ExecutionMode | null; mergeRetries?: number; workflowStepRetries?: number; stuckKillCount?: number | null; resumeLimboCount?: number | null; resumeLimboTipSha?: string | null; resumeLimboStepSignature?: string | null; postReviewFixCount?: number | null; recoveryRetryCount?: number | null; taskDoneRetryCount?: number | null; worktreeSessionRetryCount?: number | null; completionHandoffLimboRecoveryCount?: number | null; verificationFailureCount?: number | null; mergeConflictBounceCount?: number | null; mergeAuditBounceCount?: number | null; mergeTransientRetryCount?: number | null; branchConflictRecoveryCount?: number | null; reviewerContextRetryCount?: number | null; reviewerFallbackRetryCount?: number | null; nextRecoveryAt?: string | null; enabledWorkflowSteps?: string[]; noCommitsExpected?: boolean | null; modelProvider?: string | null; modelId?: string | null; validatorModelProvider?: string | null; validatorModelId?: string | null; planningModelProvider?: string | null; planningModelId?: string | null; thinkingLevel?: string | null; error?: string | null; summary?: string | null; sessionFile?: string | null; firstExecutionAt?: string | null; cumulativeActiveMs?: number | null; executionStartedAt?: string | null; executionCompletedAt?: string | null; review?: import("./types.js").TaskReview | null; reviewState?: import("./types.js").TaskReviewState | null; workflowStepResults?: import("./types.js").WorkflowStepResult[] | null; mergeDetails?: import("./types.js").MergeDetails | null; sourceIssue?: import("./types.js").TaskSourceIssue | null; sourceMetadataPatch?: Record | null; githubTracking?: import("./types.js").TaskGithubTracking | null; tokenUsage?: import("./types.js").TaskTokenUsage | null; modifiedFiles?: string[] | null; missionId?: string | null; sliceId?: string | null }, + updates: { title?: string; description?: string; priority?: TaskPriority | null; prompt?: string; worktree?: string | null; status?: string | null; dependencies?: string[]; steps?: import("./types.js").TaskStep[]; customFields?: Record; currentStep?: number; blockedBy?: string | null; overlapBlockedBy?: string | null; assignedAgentId?: string | null; pausedByAgentId?: string | null; pausedReason?: string | null; tokenBudgetSoftAlertedAt?: string | null; worktrunkFallbackAlertedAt?: string | null; worktrunkFailure?: import("./types.js").Task["worktrunkFailure"] | null; tokenBudgetHardAlertedAt?: string | null; tokenBudgetOverride?: import("./types.js").TaskTokenBudgetOverride | null; dispatchStormCount?: number | null; lastDispatchAt?: string | null; assigneeUserId?: string | null; scopeOverride?: boolean | null; scopeOverrideReason?: string | null; scopeAutoWiden?: string[] | null; nodeId?: string | null; effectiveNodeId?: string | null; effectiveNodeSource?: string | null; checkedOutBy?: string | null; checkedOutAt?: string | null; checkoutNodeId?: string | null; checkoutRunId?: string | null; checkoutLeaseRenewedAt?: string | null; checkoutLeaseEpoch?: number | null; paused?: boolean; baseBranch?: string | null; autoMerge?: boolean | null; branch?: string | null; executionStartBranch?: string | null; baseCommitSha?: string | null; size?: "S" | "M" | "L"; reviewLevel?: number; executionMode?: import("./types.js").ExecutionMode | null; mergeRetries?: number; workflowStepRetries?: number; stuckKillCount?: number | null; resumeLimboCount?: number | null; graphResumeRetryCount?: number | null; resumeLimboTipSha?: string | null; resumeLimboStepSignature?: string | null; postReviewFixCount?: number | null; recoveryRetryCount?: number | null; taskDoneRetryCount?: number | null; worktreeSessionRetryCount?: number | null; completionHandoffLimboRecoveryCount?: number | null; verificationFailureCount?: number | null; mergeConflictBounceCount?: number | null; mergeAuditBounceCount?: number | null; mergeTransientRetryCount?: number | null; branchConflictRecoveryCount?: number | null; reviewerContextRetryCount?: number | null; reviewerFallbackRetryCount?: number | null; nextRecoveryAt?: string | null; enabledWorkflowSteps?: string[]; noCommitsExpected?: boolean | null; modelProvider?: string | null; modelId?: string | null; validatorModelProvider?: string | null; validatorModelId?: string | null; planningModelProvider?: string | null; planningModelId?: string | null; thinkingLevel?: string | null; error?: string | null; summary?: string | null; sessionFile?: string | null; firstExecutionAt?: string | null; cumulativeActiveMs?: number | null; executionStartedAt?: string | null; executionCompletedAt?: string | null; review?: import("./types.js").TaskReview | null; reviewState?: import("./types.js").TaskReviewState | null; workflowStepResults?: import("./types.js").WorkflowStepResult[] | null; mergeDetails?: import("./types.js").MergeDetails | null; sourceIssue?: import("./types.js").TaskSourceIssue | null; sourceMetadataPatch?: Record | null; githubTracking?: import("./types.js").TaskGithubTracking | null; tokenUsage?: import("./types.js").TaskTokenUsage | null; modifiedFiles?: string[] | null; missionId?: string | null; sliceId?: string | null }, runContext?: RunMutationContext, ): Promise { return this.withTaskLock(id, () => this.updateTaskUnlocked(id, updates, runContext)); @@ -8087,6 +8090,11 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} } else if (updates.resumeLimboCount !== undefined) { task.resumeLimboCount = updates.resumeLimboCount; } + if (updates.graphResumeRetryCount === null) { + task.graphResumeRetryCount = null; + } else if (updates.graphResumeRetryCount !== undefined) { + task.graphResumeRetryCount = updates.graphResumeRetryCount; + } if (updates.resumeLimboTipSha === null) { task.resumeLimboTipSha = undefined; } else if (updates.resumeLimboTipSha !== undefined) { diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 12bad910f7..966f0003f7 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -2265,6 +2265,11 @@ export interface Task { * Incremented by self-healing for resume-limbo detection and reset when * progress is observed or recovery escalates to a fresh todo dispatch. */ resumeLimboCount?: number; + /** Bounded auto-retry attempts for transient workflow-graph failures observed + * immediately after engine-restart or unpause resume. Reset by manual retry + * and by successful forward progress; capped by the executor before terminal + * `status:"failed"` is recorded to preserve the FN-5704 anti-loop exemption. */ + graphResumeRetryCount?: number | null; /** Branch tip SHA snapshot captured at the last reclaim/unpause attempt used * by resume-limbo detection to determine whether commits advanced. */ resumeLimboTipSha?: string; diff --git a/packages/engine/src/__tests__/executor-recovery.test.ts b/packages/engine/src/__tests__/executor-recovery.test.ts index 112ca63678..056cb1cc39 100644 --- a/packages/engine/src/__tests__/executor-recovery.test.ts +++ b/packages/engine/src/__tests__/executor-recovery.test.ts @@ -698,6 +698,207 @@ describe("TaskExecutor bounded recovery retries", () => { expect(store.handoffToReview).not.toHaveBeenCalled(); }); + it("auto-retries a bounded transient resume-after-restart graph failure instead of parking", async () => { + const store = createMockStore(); + const task = { + id: "FN-001", + title: "Test", + description: "Test", + column: "in-progress", + status: undefined, + dependencies: [], + steps: [{ name: "Step 1", status: "pending" }], + currentStep: 0, + log: [{ timestamp: new Date().toISOString(), action: "Resumed after engine restart" }], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + graphResumeRetryCount: 0, + } as Task; + store.getTask.mockResolvedValue({ ...task, paused: false, error: null }); + const executor = new TaskExecutor(store, "/tmp/test", {}); + const executeSpy = vi.spyOn(executor as any, "execute").mockResolvedValue(undefined); + + await (executor as any).handleGraphFailure(task, { + disposition: "failed", + outcome: "failure", + visitedNodeIds: ["execute"], + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(store.updateTask).toHaveBeenCalledWith( + "FN-001", + { graphResumeRetryCount: 1, status: null, error: null }, + undefined, + ); + expect(store.updateTask).not.toHaveBeenCalledWith( + "FN-001", + expect.objectContaining({ status: "failed" }), + expect.anything(), + ); + expect(store.handoffToReview).not.toHaveBeenCalled(); + expect(executeSpy).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-001" })); + }); + + it("auto-retries a bounded transient graph failure after unpause resume instead of parking", async () => { + const store = createMockStore(); + const task = { + id: "FN-001", + title: "Test", + description: "Test", + column: "in-progress", + status: undefined, + dependencies: [], + steps: [{ name: "Step 1", status: "pending" }], + currentStep: 0, + log: [{ timestamp: new Date().toISOString(), action: "Resuming execution after unpause" }], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + graphResumeRetryCount: 0, + } as Task; + store.getTask.mockResolvedValue({ ...task, paused: false, error: null }); + const executor = new TaskExecutor(store, "/tmp/test", {}); + const executeSpy = vi.spyOn(executor as any, "execute").mockResolvedValue(undefined); + + await (executor as any).handleGraphFailure(task, { + disposition: "failed", + outcome: "failure", + visitedNodeIds: ["execute"], + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(store.updateTask).toHaveBeenCalledWith( + "FN-001", + { graphResumeRetryCount: 1, status: null, error: null }, + undefined, + ); + expect(store.handoffToReview).not.toHaveBeenCalled(); + expect(executeSpy).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-001" })); + }); + + it("parks a transient resume graph failure once the retry budget is exhausted", async () => { + const store = createMockStore(); + const task = { + id: "FN-001", + title: "Test", + description: "Test", + column: "in-progress", + status: undefined, + dependencies: [], + steps: [{ name: "Step 1", status: "pending" }], + currentStep: 0, + log: [{ timestamp: new Date().toISOString(), action: "Resumed after engine restart" }], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + graphResumeRetryCount: 2, + } as Task; + store.getTask.mockResolvedValue({ ...task, paused: false, error: null }); + const warnSpy = vi.spyOn(executorLog, "warn").mockImplementation(() => undefined); + const executor = new TaskExecutor(store, "/tmp/test", {}); + const executeSpy = vi.spyOn(executor as any, "execute").mockResolvedValue(undefined); + + await (executor as any).handleGraphFailure(task, { + disposition: "failed", + outcome: "failure", + visitedNodeIds: ["execute"], + }); + + const message = "Workflow graph terminated with failure at node 'execute'"; + expect(store.updateTask).toHaveBeenCalledWith("FN-001", { error: message, status: "failed" }, undefined); + expect(store.handoffToReview).toHaveBeenCalledWith( + "FN-001", + expect.objectContaining({ evidence: expect.objectContaining({ reason: "workflow-graph-failed" }) }), + ); + expect(executeSpy).not.toHaveBeenCalled(); + warnSpy.mockRestore(); + }); + + it.each([ + ["non-empty execute-seam reason", { result: { reason: "interpreter-error: boom", visitedNodeIds: ["execute"] } }], + ["settings/workflow-selection reason before node progress", { result: { reason: "settings-load-failed: boom", visitedNodeIds: [] } }], + ["completed step progress", { task: { steps: [{ name: "Step 1", status: "done" }] }, result: { visitedNodeIds: ["execute"] } }], + ["lastError", { task: { lastError: "boom" }, result: { visitedNodeIds: ["execute"] } }], + ["failureReason", { task: { failureReason: "boom" }, result: { visitedNodeIds: ["execute"] } }], + ])("preserves terminal failed handling for genuine graph failure: %s", async (_name, fixture) => { + const store = createMockStore(); + const task = { + id: "FN-001", + title: "Test", + description: "Test", + column: "in-progress", + status: undefined, + dependencies: [], + steps: [{ name: "Step 1", status: "pending" }], + currentStep: 0, + log: [{ timestamp: new Date().toISOString(), action: "Resumed after engine restart" }], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + graphResumeRetryCount: 0, + ...(fixture.task ?? {}), + } as Task; + store.getTask.mockResolvedValue({ ...task, paused: false, error: null }); + const warnSpy = vi.spyOn(executorLog, "warn").mockImplementation(() => undefined); + const executor = new TaskExecutor(store, "/tmp/test", {}); + const executeSpy = vi.spyOn(executor as any, "execute").mockResolvedValue(undefined); + + await (executor as any).handleGraphFailure(task, { + disposition: "failed", + outcome: "failure", + ...fixture.result, + }); + + const failedNode = fixture.result.visitedNodeIds.at(-1) ?? "unknown"; + const message = `Workflow graph terminated with failure at node '${failedNode}'`; + expect(store.updateTask).toHaveBeenCalledWith("FN-001", { error: message, status: "failed" }, undefined); + expect(store.handoffToReview).toHaveBeenCalledWith( + "FN-001", + expect.objectContaining({ evidence: expect.objectContaining({ reason: "workflow-graph-failed" }) }), + ); + expect(executeSpy).not.toHaveBeenCalled(); + warnSpy.mockRestore(); + }); + + describe("transient resume-after-restart graph failure classifier", () => { + const makeClassifierTask = (overrides: Partial = {}) => ({ + id: "FN-001", + title: "Test", + description: "Test", + column: "in-progress", + status: undefined, + dependencies: [], + steps: [ + { name: "Step 1", status: "pending" }, + { name: "Step 2", status: "pending" }, + ], + currentStep: 0, + log: [{ timestamp: new Date().toISOString(), action: "Resumed after engine restart" }], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + ...overrides, + }) as Task; + + const isTransient = (task: Task, result: any) => { + const executor = new TaskExecutor(createMockStore(), "/tmp/test", {}); + return (executor as any).isTransientResumeAfterRestartGraphFailure(task, result); + }; + + it("accepts only the exact no-progress execute-seam post-resume signature", () => { + expect(isTransient(makeClassifierTask(), { visitedNodeIds: ["execute"] })).toBe(true); + expect(isTransient(makeClassifierTask({ log: [{ timestamp: new Date().toISOString(), action: "Resuming execution after unpause" }] }), { visitedNodeIds: ["execute"] })).toBe(true); + expect(isTransient(makeClassifierTask(), { visitedNodeIds: [] })).toBe(true); + }); + + it.each([ + ["non-empty reason", makeClassifierTask(), { visitedNodeIds: ["execute"], reason: "settings-load-failed: boom" }], + ["non-execute failed node", makeClassifierTask(), { visitedNodeIds: ["planning"] }], + ["completed step progress", makeClassifierTask({ steps: [{ name: "Step 1", status: "done" }] }), { visitedNodeIds: ["execute"] }], + ["lastError", makeClassifierTask({ lastError: "boom" } as any), { visitedNodeIds: ["execute"] }], + ["failureReason", makeClassifierTask({ failureReason: "boom" } as any), { visitedNodeIds: ["execute"] }], + ["missing resume log", makeClassifierTask({ log: [{ timestamp: new Date().toISOString(), action: "Started execution" }] }), { visitedNodeIds: ["execute"] }], + ])("rejects %s as genuine/non-transient", (_name, task, result) => { + expect(isTransient(task as Task, result)).toBe(false); + }); + }); + it("preserves genuine in-progress graph failure handling", async () => { const store = createMockStore(); const task = { diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 7ce9123200..a3dea6f43d 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -275,6 +275,13 @@ const MAX_WORKFLOW_STEP_RETRIES = 3; const MAX_TASK_DONE_SESSION_RETRIES = 3; /** Maximum todo requeues after exhausting in-session fn_task_done retries. */ const MAX_TASK_DONE_REQUEUE_RETRIES = 3; +/** + * Maximum bounded retries for the narrow resume-after-restart graph transient. + * Budget exhaustion falls through to terminal status:"failed" so FN-5704's + * self-healing anti-loop exemption remains intact for genuine graph failures. + */ +const MAX_TRANSIENT_GRAPH_RESUME_RETRIES = 2; +const TRANSIENT_GRAPH_RESUME_RETRY_BACKOFF_MS = process.env.VITEST || process.env.NODE_ENV === "test" ? 0 : 1_000; /** How long to wait before recovering a completed task still stuck in in-progress. */ const COMPLETED_TASK_WATCHDOG_MS = 60_000; /** How long to wait before retrying a workflow rerun handoff that never reached in-progress. */ @@ -3833,6 +3840,11 @@ export class TaskExecutor { } if (result.disposition === "failed") { await this.handleGraphFailure(task, result); + } else if (result.disposition === "completed") { + const live = await this.store.getTask(task.id).catch(() => task); + if ((live.graphResumeRetryCount ?? 0) !== 0) { + await this.store.updateTask(task.id, { graphResumeRetryCount: 0 }, this.getRunContextFor(task.id)); + } } return true; } finally { @@ -5970,6 +5982,22 @@ export class TaskExecutor { } } + private isTransientResumeAfterRestartGraphFailure(live: Task, result: WorkflowGraphTaskRunResult): boolean { + if ((result.reason ?? "").trim().length > 0) return false; + + const failedNode = result.visitedNodeIds[result.visitedNodeIds.length - 1]; + if (failedNode !== undefined && failedNode !== "execute") return false; + + if (live.steps.some((step) => step.status === "done")) return false; + + const failureState = live as Task & { lastError?: unknown; failureReason?: unknown }; + if (failureState.lastError != null || failureState.failureReason != null) return false; + + const latestAction = live.log.at(-1)?.action; + return latestAction === "Resumed after engine restart" + || latestAction === "Resuming execution after unpause"; + } + /** Terminal failure of a graph run: record the error and park the task in * review so a human can act — never leave it invisible in in-progress. */ private async handleGraphFailure(task: Task, result: WorkflowGraphTaskRunResult): Promise { @@ -5992,6 +6020,32 @@ export class TaskExecutor { return; } const failedNode = result.visitedNodeIds[result.visitedNodeIds.length - 1]; + if (this.isTransientResumeAfterRestartGraphFailure(live, result)) { + const priorRetries = live.graphResumeRetryCount ?? 0; + if (priorRetries < MAX_TRANSIENT_GRAPH_RESUME_RETRIES) { + const nextRetries = priorRetries + 1; + const benignMessage = `Transient resume-after-restart graph failure — auto-retrying (${nextRetries}/${MAX_TRANSIENT_GRAPH_RESUME_RETRIES}) instead of parking`; + executorLog.warn(`${task.id}: ${benignMessage}`); + await this.store.logEntry(task.id, benignMessage, undefined, this.getRunContextFor(task.id)); + await this.store.updateTask(task.id, { + graphResumeRetryCount: nextRetries, + status: null, + error: null, + }, this.getRunContextFor(task.id)); + const scheduleRetry = () => { + this.execute(live).catch((err) => + executorLog.error(`Failed transient graph resume retry for ${task.id}:`, err), + ); + }; + if (TRANSIENT_GRAPH_RESUME_RETRY_BACKOFF_MS > 0) { + const handle = setTimeout(scheduleRetry, TRANSIENT_GRAPH_RESUME_RETRY_BACKOFF_MS); + handle.unref?.(); + } else { + setTimeout(scheduleRetry, 0).unref?.(); + } + return; + } + } const message = `Workflow graph terminated with failure at node '${failedNode ?? "unknown"}'`; executorLog.warn(`${task.id}: ${message}`); await this.store.logEntry(task.id, message, undefined, this.getRunContextFor(task.id));