diff --git a/changelog.d/1526.added.md b/changelog.d/1526.added.md index e6471f564..417005a17 100644 --- a/changelog.d/1526.added.md +++ b/changelog.d/1526.added.md @@ -1 +1 @@ -**Workflow failure recovery:** A node can now recover from failure by returning to an upstream node and retrying, instead of stopping the run. Define a failure recovery edge with `onFail: true` pointing from the failing node back to the recovery target, with a configurable `max` attempts bound (1–8, default 3). Example: `{ from: 'review', to: 'draft', onFail: true, max: 3 }` — when Review fails, the workflow resets to Draft and runs through Review again, up to 3 times total. After attempts are exhausted, the run fails cleanly. Does not interfere with existing `failOnce` retry or success-path `loop` edges. +**Workflow failure recovery:** A node can now recover from failure by returning to an upstream node and retrying, instead of stopping the run. Define a failure recovery edge with `onFail: true` pointing from the failing node back to the recovery target, with a configurable `max` bound on recoveries (1–8, default 3). Example: `{ from: 'review', to: 'draft', onFail: true, max: 3 }` — when Review fails, the workflow resets the body from Draft through Review and runs it again, up to three recoveries. After the recoveries are used, the run fails cleanly. Does not interfere with existing `failOnce` retry or success-path `loop` edges. diff --git a/changelog.d/1541.fixed.md b/changelog.d/1541.fixed.md new file mode 100644 index 000000000..9460a061b --- /dev/null +++ b/changelog.d/1541.fixed.md @@ -0,0 +1 @@ +The failure recovery edge added by #1529 reset only the recovery target and the failing node, so a node on the way back stayed `done` and the retry was fed the output it had just rejected. Recovery now resets every node from the target through the failing node and re-runs them in order. ([#1541](https://github.com/ThinkFlowLab/FlyBot/issues/1541)) diff --git a/changelog.d/1542.fixed.md b/changelog.d/1542.fixed.md new file mode 100644 index 000000000..d196fa364 --- /dev/null +++ b/changelog.d/1542.fixed.md @@ -0,0 +1 @@ +The failure recovery added by #1529 accepted a `max` from 1 to 8 but never recovered on `max: 1`: the first failure already read as the limit, so the run failed at once and the attempt counter stayed empty. `max` now counts recoveries, so `max: 1` recovers once and `max: 8` still means at most eight. ([#1542](https://github.com/ThinkFlowLab/FlyBot/issues/1542)) diff --git a/docs/subsystems/workflow.md b/docs/subsystems/workflow.md index 7a1c04729..3987f5148 100644 --- a/docs/subsystems/workflow.md +++ b/docs/subsystems/workflow.md @@ -27,7 +27,7 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher, | File | Role | | --- | --- | | `plugins/workflow/index.js` | Plugin entry (`flybot-workflow`). Endpoints, `run_workflow`, `define_workflow`, the board listener, and the reply middleware that ingests a step result and denies delivery to persistent Bot chats. | -| `plugins/workflow/src/graph.js` | Nodes, edges, positions, and named inputs and outputs. A node starts when its forward predecessors are settled and a taken edge reaches it. Same-rank nodes run together. A forward cycle is refused. A loop edge is the only cycle for success paths. A `when` edge chooses a branch; the other is skipped. A required input with no upstream provider is refused before a run starts. An `onFail` edge provides failure recovery: when a node fails with recovery attempts remaining, the workflow resets to the upstream target and re-runs through the failing node (tracked by `failAttempts` per edge, with a configurable `max` bound, 1–8, default 3). After attempts are exhausted, the node fails permanently. | +| `plugins/workflow/src/graph.js` | Nodes, edges, positions, and named inputs and outputs. A node starts when its forward predecessors are settled and a taken edge reaches it. Same-rank nodes run together. A forward cycle is refused. A loop edge is the only cycle for success paths. A `when` edge chooses a branch; the other is skipped. A required input with no upstream provider is refused before a run starts. An `onFail` edge provides failure recovery: when a node fails with recoveries remaining, the workflow resets the body from the upstream target through the failing node and re-runs it in order (tracked by `failAttempts` per edge, with `max` counting recoveries, 1–8, default 3). After the recoveries are used, the node fails permanently. | | `plugins/workflow/src/service.js` | The store, lanes (active runs, FIFO queue, terminal history), `maxConcurrent`, `rename`, `delete`, `saveGraph`, `define`, `runNamed`, `ingest`, `stepResult`. `rename` updates a workflow's name and optional description; `delete` removes the workflow, releases ephemeral Bots, and cleans logs/runs/routes. Factory template ids (e.g. `delivery`) are normal saved rows and may be deleted. A spawn hook makes one run Bot per node. Disposing the plugin ends every live run still marked `running` (nodes in flight are failed, the run is archived) and then releases every run Bot and drops those ids from the saved record only once the release has finished, handing the release back as a promise the host awaits, so a shutdown cannot persist a run as `running` with no ephemeral map and cannot leave hidden Bots that no state names (#1387, #1414). A released run Bot also gives its runner name back to the allocator instead of keeping it for the lifetime of the process. A dispatch that does not land is not marked handed. | | `plugins/workflow/src/ephemeral.js` | Builds each node's run Bot from its prompt, role, model and tools, and releases the ids the host returns. The `tools` field is **always omitted** from `fb.bot.create` so every node Bot inherits all registered tools (GitHub, Composio, file/shell tools, and team tools limited by `EPHEMERAL_DENIED_TOOLS` at the guard layer). `node.tools` is stored as documentation only in `fields.tools` but does not restrict the grant. All node Bots get `danger-full-access` permission. | | `plugins/workflow/src/match.js` | Which open card matches a Workflow or a local Bot. | @@ -50,7 +50,7 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher, 4. Each workflow keeps a FIFO queue and up to `maxConcurrent` live runs (default 2, an integer from 1 to 8 on the record, via `workflow/concurrency`, or in Settings → Workflow graphs). A task that does not fit is enqueued. When a run finishes, every workflow with a free slot is drained, oldest enqueue first. A new assign on a workflow that already has a queue waits behind it. A task does not replace a running run or release that run's Bots. Migrating an old store keeps only runs that were still `running` with ephemeral Bot ids; other legacy scripted rows are dropped from active state (finished tasks may remain in lane history). History records a finished or failed live run. Lowering the cap does not stop a run that has already started. Assigning `{ botId }` claims the card for that local Bot and does not enqueue. A remote Bot is refused. `activeCount` is `runningOf(lane)` — live runs with `live === true` and `status === 'running'`. The stage aggregate chip and the Runs tab (运行记录) must agree: every such run appears as its own running row and progress card, and the chip count does not exceed that list. 5. A node declares up to eight inputs and eight outputs. A required input is an output of the same name on an ancestor, or `task`, `title`, or `brief` on a root. A start with no provider is refused, and Bots spawned for that attempt are released. On a Bot-backed run a successor stays queued until its required inputs are present. If every forward predecessor is done, failed or skipped and an input is still missing, that node fails and the run stops. The dispatch text carries `Input : `. The Bot reports `OUTPUT : `; a single declared output may be the reply body. Nodes start in edge order. 6. Each node is a real ephemeral Bot for that run, then it is removed. A permanent sidebar Bot (`botId`) is optional and is not required for the step to run. **Every node Bot gets the same full permissions a normal sidebar Bot gets:** the `tools` field is always omitted from `fb.bot.create`, letting the Bot inherit all registered tools — file/shell tools, GitHub (`mcp__github__*`: pull requests, reviews, comments), Composio, and team tools — limited only by `EPHEMERAL_DENIED_TOOLS` at the guard/approval layer (#1466, 29 team-admin tools). `node.tools` is stored as documentation in `fields.tools` but does not restrict the grant. All node Bots get `danger-full-access` permission. A hidden run Bot cannot show an approval card, so a granted tool is allowed and a connector it did not name is refused. The host stores this on the run Bot as `stepTools` and `permission`. -7. A loop edge (`loop: true`, `max` from 1 to 8, default 3, optional `until: { output, eq }`) is the only cycle for success paths. The body runs again until `until` matches or `max` runs have finished, and the forward edges wait until then. Restarting the body drops the branch choice made inside it and queues a sibling that the previous pass skipped, so the next pass can take the other side. A `when: { output, eq }` edge is taken only when that output equals `eq`. `when: { else: true }` is taken only when no other `when` matched. A node whose forward edges are all closed is `skipped` and does not block a join. A run cannot spin: every loop has a max. An `onFail` edge (`onFail: true`, `max` from 1 to 8, default 3) provides failure recovery: when a node fails, if an `onFail` edge points from that node back to an upstream target and recovery attempts remain (tracked in `failAttempts` per edge), the workflow resets both nodes to their pre-failure states (the target to `running`, the failing node to `queued`) and re-runs from the target through the failing node. After `max` attempts are exhausted, the node fails permanently and the run stops. An `onFail` edge does not interfere with `failOnce` retry (which restarts the same node once) or success-path `loop` edges. Settings → Workflow graphs can add a link, a condition, a loop, or a failure recovery edge. +7. A loop edge (`loop: true`, `max` from 1 to 8, default 3, optional `until: { output, eq }`) is the only cycle for success paths. The body runs again until `until` matches or `max` runs have finished, and the forward edges wait until then. Restarting the body drops the branch choice made inside it and queues a sibling that the previous pass skipped, so the next pass can take the other side. A `when: { output, eq }` edge is taken only when that output equals `eq`. `when: { else: true }` is taken only when no other `when` matched. A node whose forward edges are all closed is `skipped` and does not block a join. A run cannot spin: every loop has a max. An `onFail` edge (`onFail: true`, `max` recoveries from 1 to 8, default 3) provides failure recovery: when a node fails, if an `onFail` edge points from that node back to an upstream target and recoveries remain (tracked in `failAttempts` per edge), the workflow resets every node from the recovery target through the failing node (the target to `running`, the rest to `queued`) and re-runs them in order. After `max` recoveries are used, the node fails permanently and the run stops. An `onFail` edge does not interfere with `failOnce` retry (which restarts the same node once) or success-path `loop` edges. Settings → Workflow graphs can add a link, a condition, or a loop; a failure recovery edge is added through `define_workflow` (`onFail: true`) and has no Settings control yet. 8. Auto-assign considers an `open` card only. A saved Bot claim beats a workflow trigger. Otherwise first hit wins: a `workflow:` token, then `workflowId` / `workflow`, then an assignee that names a workflow, then a `bot:` token, then an assignee that names a local Bot, then the longest trigger contained in the title, note or criteria. A miss returns no route. A second ingest of the same task and target is a no-op, and a live run for that task id is not restarted. 9. The browser draws text only. Names and messages are React text children. The unit renderer asserts a hostile name is not markup. 10. `enabled: false` registers no endpoint, tool or listener and writes nothing. diff --git a/plugins/workflow/README.md b/plugins/workflow/README.md index fbfa4f6c7..36abff64f 100644 --- a/plugins/workflow/README.md +++ b/plugins/workflow/README.md @@ -42,9 +42,9 @@ An edge may be a condition, a loop, or a failure recovery. A condition is `when: A loop is `loop: true` on the edge that returns to the body, with `max` from 1 to 8 (default 3) and an optional `until: { output, eq }`. The body runs again while `until` does not match and the run count is still under `max`. A branch the previous pass skipped can be chosen the next time. The forward edges wait until the loop stops. A loop is the only cycle the graph allows for success paths. -A failure recovery is `onFail: true` on an edge pointing from the failing node back to an upstream node, with `max` from 1 to 8 (default 3). When the node fails and recovery attempts remain, the workflow resets to the recovery target and runs through the failing node again. After `max` recovery attempts, the node fails permanently and the run stops. Failure recovery edges do not interfere with success-path loops or `failOnce` retry behavior. +A failure recovery is `onFail: true` on an edge pointing from the failing node back to an upstream node, with `max` from 1 to 8 recoveries (default 3). When the node fails and fewer than `max` recoveries have run, the workflow resets the body from the recovery target through the failing node and runs it again. After `max` recoveries, the node fails permanently and the run stops. Failure recovery edges do not interfere with success-path loops or `failOnce` retry behavior. -Settings → Workflow graphs can add a link, a condition, a loop, or a failure recovery. After a live run finishes, the stage keeps showing that run until a new one starts. +Settings → Workflow graphs can add a link, a condition, or a loop. A failure recovery edge is added through `define_workflow` (`onFail: true`); the Settings canvas has no control for it yet. After a live run finishes, the stage keeps showing that run until a new one starts. ## When a step does not land @@ -57,7 +57,7 @@ A reply that says the model could not answer (`could not answer your message` or A local Bot uses the same store: - `run_workflow` — `workflow` (id or exact name) and `brief` (the task). Optional `task` id. This joins the same queue a manual assign uses. -- `define_workflow` — `name`, `nodes` (`{ id, name, prompt?, role?, model?, tools?, botId?, inputs?, outputs?, x?, y? }`), `edges` (`{ from, to, when?, loop?, max?, until? }`). `prompt` is that node's Bot system prompt. `tools` equips the ephemeral Bot (`github`, `bash`, `read`, `edit`). `botId` is optional. `inputs` and `outputs` are `{ name, type?, required? }`. `when` chooses an edge; `loop: true` with `max` is the only cycle. Pass `workflow` to replace one. A friend's Bot is refused. This writes the same graph Settings and the flowchart edit. +- `define_workflow` — `name`, `nodes` (`{ id, name, prompt?, role?, model?, tools?, botId?, inputs?, outputs?, x?, y? }`), `edges` (`{ from, to, when?, loop?, max?, until?, onFail? }`). `prompt` is that node's Bot system prompt. `tools` equips the ephemeral Bot (`github`, `bash`, `read`, `edit`). `botId` is optional. `inputs` and `outputs` are `{ name, type?, required? }`. `when` chooses an edge; `loop: true` with `max` is the only cycle. Pass `workflow` to replace one. A friend's Bot is refused. This writes the same graph Settings and the flowchart edit. ## Install diff --git a/plugins/workflow/src/graph.js b/plugins/workflow/src/graph.js index a118fc0da..0deab29ff 100644 --- a/plugins/workflow/src/graph.js +++ b/plugins/workflow/src/graph.js @@ -542,13 +542,14 @@ function loopDecision(graph, nodeId, outputs, loops) { return null } -// Check if a failed node has an onFail edge and attempts remaining. +// Check if a failed node has an onFail edge and recoveries left. `max` counts +// recoveries, unlike a loop edge's max which counts how many times its body runs. function failRecoveryDecision(graph, nodeId, failAttempts) { const edges = graph.edges.filter(edge => edge.onFail === true && edge.from === nodeId) for (const edge of edges) { const ran = (Number(failAttempts?.[edgeKey(edge)]) || 0) + 1 const max = Number.isInteger(edge.max) ? edge.max : 3 - if (ran >= max) continue + if (ran > max) continue return { edge, ran } } return null @@ -697,7 +698,7 @@ export function failNode(graph, status, nodeId, ctx = {}) { if (recovery) { // Failure recovery: reset to the upstream target node failAttempts[edgeKey(recovery.edge)] = recovery.ran - const body = loopBody(graph, recovery.edge.to, nodeId) + const body = loopBody(graph, nodeId, recovery.edge.to) const next = copyStatus(status) const reset = [] diff --git a/plugins/workflow/test/failure-recovery-attempts.test.js b/plugins/workflow/test/failure-recovery-attempts.test.js new file mode 100644 index 000000000..6c14ba008 --- /dev/null +++ b/plugins/workflow/test/failure-recovery-attempts.test.js @@ -0,0 +1,64 @@ +// #1542: an onFail edge's `max` counts recoveries. max: 1 must recover once; +// the run fails permanently only after those recoveries are used. + +import { describe, it } from 'node:test' +import assert from 'node:assert/strict' +import { failNode, normalizeGraph } from '../src/graph.js' + +function chain(max) { + return normalizeGraph({ + nodes: [ + { id: 'draft', name: 'Draft' }, + { id: 'copy', name: 'Copy' }, + { id: 'review', name: 'Review' }, + ], + edges: [ + { from: 'draft', to: 'copy' }, + { from: 'copy', to: 'review' }, + { from: 'review', to: 'draft', onFail: true, max }, + ], + }) +} + +// The state just before Review fails on a three-node chain. It is rebuilt for +// each attempt so this test does not depend on how the recovered body is reset. +function reviewRunning() { + return { draft: 'done', copy: 'done', review: 'running' } +} + +describe('Failure recovery attempts', () => { + it('max: 1 recovers on the first failure and records one attempt', () => { + const graph = chain(1) + const result = failNode(graph, reviewRunning(), 'review', { failAttempts: {} }) + + assert.equal(result.recovered, true) + assert.equal(result.failAttempts['review>draft'], 1) + assert.equal(result.status.draft, 'running') + assert.equal(result.status.review, 'queued') + }) + + it('max: 1 fails permanently on the next failure and keeps the count at max', () => { + const graph = chain(1) + const result = failNode(graph, reviewRunning(), 'review', { failAttempts: { 'review>draft': 1 } }) + + assert.equal(result.recovered, undefined) + assert.equal(result.status.review, 'failed') + assert.equal(result.failAttempts['review>draft'], 1) + }) + + it('max: 2 recovers twice and then fails permanently', () => { + const graph = chain(2) + const first = failNode(graph, reviewRunning(), 'review', { failAttempts: {} }) + assert.equal(first.recovered, true) + assert.equal(first.failAttempts['review>draft'], 1) + + const second = failNode(graph, reviewRunning(), 'review', { failAttempts: first.failAttempts }) + assert.equal(second.recovered, true) + assert.equal(second.failAttempts['review>draft'], 2) + + const third = failNode(graph, reviewRunning(), 'review', { failAttempts: second.failAttempts }) + assert.equal(third.recovered, undefined) + assert.equal(third.status.review, 'failed') + assert.equal(third.failAttempts['review>draft'], 2) + }) +}) diff --git a/plugins/workflow/test/failure-recovery-body.test.js b/plugins/workflow/test/failure-recovery-body.test.js new file mode 100644 index 000000000..5d7c7ff8b --- /dev/null +++ b/plugins/workflow/test/failure-recovery-body.test.js @@ -0,0 +1,109 @@ +// #1541: a failure recovery resets every node from the recovery target through +// the failing node. A chain longer than the onFail edge's two ends must re-run +// its middle nodes too, or the retry feeds the failing node stale output. + +import { describe, it } from 'node:test' +import assert from 'node:assert/strict' +import { completeNode, failNode, freshStatus, normalizeGraph, startFrontier } from '../src/graph.js' +import { createWorkflowService } from '../src/service.js' + +function chain() { + return normalizeGraph({ + nodes: [ + { id: 'draft', name: 'Draft' }, + { id: 'copy', name: 'Copy' }, + { id: 'review', name: 'Review' }, + ], + edges: [ + { from: 'draft', to: 'copy' }, + { from: 'copy', to: 'review' }, + { from: 'review', to: 'draft', onFail: true, max: 3 }, + ], + }) +} + +describe('Failure recovery body', () => { + it('a recovered run resets the nodes between the target and the failing node', () => { + const graph = chain() + let status = freshStatus(graph) + status = startFrontier(graph, status).status + status = completeNode(graph, status, 'draft').status + status = completeNode(graph, status, 'copy').status + assert.equal(status.review, 'running') + + const recovered = failNode(graph, status, 'review', { failAttempts: {} }) + + assert.equal(recovered.recovered, true) + assert.deepEqual(recovered.reset.slice().sort(), ['copy', 'draft', 'review']) + assert.equal(recovered.status.draft, 'running') + assert.equal(recovered.status.copy, 'queued') + assert.equal(recovered.status.review, 'queued') + + // The second pass walks draft -> copy -> review again. A done copy would + // skip straight back to review with the output the rejection was about. + const second = completeNode(graph, recovered.status, 'draft') + assert.equal(second.status.copy, 'running') + assert.deepEqual(second.started, ['copy']) + }) + + it('a recovered live run re-runs the middle node instead of feeding stale output', async () => { + const spawned = [] + const released = [] + const dispatched = [] + let seq = 0 + const roster = [{ id: 'main', name: 'Chief', role: 'lead', isMain: true, remote: false }] + const svc = createWorkflowService({ + persist: false, + now: () => 1_700_000_000_000, + bots: () => roster, + spawn: async (spec) => { + const id = `eph-${++seq}` + spawned.push({ id, spec }) + return id + }, + release: async (id) => { released.push(id) }, + dispatch: async (step) => { + dispatched.push({ nodeId: step.nodeId, botId: step.botId }) + return { ok: true } + }, + }) + const made = svc.define({ + name: 'Recovery chain', + nodes: [ + { id: 'draft', name: 'Draft', prompt: 'Write it.' }, + { id: 'copy', name: 'Copy', prompt: 'Polish it.' }, + { id: 'review', name: 'Review', prompt: 'Judge it.' }, + ], + edges: [ + { from: 'draft', to: 'copy' }, + { from: 'copy', to: 'review' }, + { from: 'review', to: 'draft', onFail: true, max: 3 }, + ], + }).workflows.find(item => item.name === 'Recovery chain') + const runOf = (view) => view.workflows.find(item => item.id === made.id).run + + const started = await svc.runNamed({ workflow: 'Recovery chain', brief: 'Ship the note' }) + assert.equal(runOf(started).nodeStatus.draft, 'running') + const mid = await svc.advance({ workflowId: made.id, nodeId: 'draft', outcome: 'done' }) + assert.equal(runOf(mid).nodeStatus.copy, 'running') + const late = await svc.advance({ workflowId: made.id, nodeId: 'copy', outcome: 'done' }) + assert.equal(runOf(late).nodeStatus.review, 'running') + + const failed = await svc.advance({ workflowId: made.id, nodeId: 'review', outcome: 'failed' }) + const parked = runOf(failed) + assert.equal(parked.status, 'running') + assert.equal(parked.nodeStatus.draft, 'running') + assert.equal(parked.nodeStatus.copy, 'queued') + assert.equal(parked.nodeStatus.review, 'queued') + assert.equal(dispatched.filter(step => step.nodeId === 'copy').length, 1) + + const redraft = await svc.advance({ workflowId: made.id, nodeId: 'draft', outcome: 'done' }) + assert.equal(runOf(redraft).nodeStatus.copy, 'running') + assert.equal(dispatched.filter(step => step.nodeId === 'copy').length, 2) + + const recopy = await svc.advance({ workflowId: made.id, nodeId: 'copy', outcome: 'done' }) + assert.equal(runOf(recopy).nodeStatus.review, 'running') + assert.equal(dispatched.filter(step => step.nodeId === 'review').length, 2) + assert.deepEqual(released, []) + }) +}) diff --git a/plugins/workflow/test/failure-recovery.test.js b/plugins/workflow/test/failure-recovery.test.js index 3435f278f..798a8c315 100644 --- a/plugins/workflow/test/failure-recovery.test.js +++ b/plugins/workflow/test/failure-recovery.test.js @@ -89,12 +89,21 @@ describe('Failure recovery', () => { failAttempts = result.failAttempts assert.equal(failAttempts['review>draft'], 2) - // Third attempt: draft → review (fail) → STOP (max 3 exhausted) + // Third attempt: draft → review (fail) → draft (recovery 3) status = completeNode(graph, status, 'draft').status result = failNode(graph, status, 'review', { failAttempts }) - + assert.equal(result.recovered, true) + status = result.status + failAttempts = result.failAttempts + assert.equal(failAttempts['review>draft'], 3) + + // Fourth attempt: draft → review (fail) → STOP (max 3 recoveries used) + status = completeNode(graph, status, 'draft').status + result = failNode(graph, status, 'review', { failAttempts }) + assert.equal(result.recovered, undefined) assert.equal(result.status.review, 'failed') + assert.equal(result.failAttempts['review>draft'], 3) }) it('does not break existing failOnce behavior', () => {