From c7dfaa914c4b0d5da5cd7349b7253c706f9c2b57 Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Thu, 29 Oct 2026 12:37:48 +0800 Subject: [PATCH] fix(workflow): drop a deleted workflow's task claims delete() judged every claim by the route map it had just emptied, so the condition never held and the claim stayed on disk. ingest() then trusted the dangling claim and asked assign() for a workflow that no longer existed, leaving the task unmatched forever, across reloads. Judge a claim on its own record, kind included, so a Bot claim that shares the id is kept. The load path also drops a workflow claim whose workflow is gone, so a store written before this fix heals on the next start. Fixes #1632 --- changelog.d/1632.fixed.md | 1 + docs/subsystems/workflow.md | 4 +- plugins/workflow/src/service.js | 12 +- plugins/workflow/test/delete-claims.test.js | 129 ++++++++++++++++++++ 4 files changed, 143 insertions(+), 3 deletions(-) create mode 100644 changelog.d/1632.fixed.md create mode 100644 plugins/workflow/test/delete-claims.test.js diff --git a/changelog.d/1632.fixed.md b/changelog.d/1632.fixed.md new file mode 100644 index 000000000..2fde59ce8 --- /dev/null +++ b/changelog.d/1632.fixed.md @@ -0,0 +1 @@ +Deleting a Workflow no longer leaves its task claim behind, so the task matches again on the next board write instead of failing with `No such workflow`; a store that already holds a claim for a deleted Workflow drops it when the plugin loads. ([#1632](https://github.com/ThinkFlowLab/FlyBot/issues/1632)) diff --git a/docs/subsystems/workflow.md b/docs/subsystems/workflow.md index 15e03c2e6..23e9c8705 100644 --- a/docs/subsystems/workflow.md +++ b/docs/subsystems/workflow.md @@ -30,7 +30,7 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher, | --- | --- | | `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 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; a run Bot whose release failed is kept in the store's own `orphans` list — not in a variable — and the load-time `reap` retries it, once per id, capped, so deleting a workflow whose Bot will not go does not strand that Bot under a name nothing can reach (the #1616 contract). 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, for every live run in a lane rather than only the one archived last (#1387, #1414, #1472). 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: it is retried twice with a short backoff, and a step still unanswered after its workflow's `stepTimeoutMinutes` (45 minutes by default) fails the run and releases its Bots (#1530, #1580). | +| `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 and the task claims that named it — a claim is judged on its own record (`kind === 'workflow'` and that id), not by the route map, which its own cleanup has already emptied (#1632); a run Bot whose release failed is kept in the store's own `orphans` list — not in a variable — and the load-time `reap` retries it, once per id, capped, so deleting a workflow whose Bot will not go does not strand that Bot under a name nothing can reach (the #1616 contract). 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, for every live run in a lane rather than only the one archived last (#1387, #1414, #1472). 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: it is retried twice with a short backoff, and a step still unanswered after its workflow's `stepTimeoutMinutes` (45 minutes by default) fails the run and releases its Bots (#1530, #1580). | | `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. `node.tools` names the connectors that node may call: `spawn` passes the cleaned list to `fb.bot.create`, the host records it as `stepTools`, and `connector-access.js` allows only the connectors it names (`github` covers `mcp__github__*`, `composio` covers `mcp__composio__*`) — a node that names none reaches none. File and shell rights do not come from that list: every node Bot gets the `danger-full-access` preset a normal sidebar Bot has, and `EPHEMERAL_DENIED_TOOLS` still refuses the team tools at the guard and approval layers. The list is also stored in `fields.tools` for the UI. | | `plugins/workflow/src/match.js` | Which open card matches a Workflow or a local Bot. | | `plugins/workflow/src/template.js` | Factory delivery/weekly graphs plus `EARLY_END_GUIDANCE` for Claim/Verify/Triage early-end markers; 发布交付流水线 (a parallel branch) and the weekly report, each node with the inputs and outputs that edge carries. | @@ -40,7 +40,7 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher, ## Entry points - Endpoint `workflow/state` and `workflow/templates` are reads. -- `workflow/create`, `workflow/rename`, `workflow/delete`, `workflow/assign`, `workflow/concurrency`, `workflow/graph`, `workflow/advance`, `workflow/reset`, `workflow/ingest` write. `workflow/concurrency` sets that workflow's `maxConcurrent` (an integer from 1 to 8); the same control is in Settings → Workflow graphs. `workflow/step-timeout` sets that workflow's `stepTimeoutMinutes` (an integer from 1 to 120, or `null` to go back to the default), and `workflow/create` accepts the same field. `workflow/rename` updates a workflow's name and optional description; read-only hosts are refused. `workflow/delete` removes a workflow from state, releases its ephemeral Bots if a run exists, and cleans logs/runs/routes. +- `workflow/create`, `workflow/rename`, `workflow/delete`, `workflow/assign`, `workflow/concurrency`, `workflow/graph`, `workflow/advance`, `workflow/reset`, `workflow/ingest` write. `workflow/concurrency` sets that workflow's `maxConcurrent` (an integer from 1 to 8); the same control is in Settings → Workflow graphs. `workflow/step-timeout` sets that workflow's `stepTimeoutMinutes` (an integer from 1 to 120, or `null` to go back to the default), and `workflow/create` accepts the same field. `workflow/rename` updates a workflow's name and optional description; read-only hosts are refused. `workflow/delete` removes a workflow from state, releases its ephemeral Bots if a run exists, and cleans logs/runs/routes and every task claim that named it, so the next `ingest` of that task matches again instead of asking `assign` for a workflow that no longer exists; a load still drops a workflow claim whose workflow is gone, so a store written before this fix heals (#1632). - Tool `run_workflow` starts a named run with a brief. Tool `define_workflow` writes the same graph the canvas edits. - `flybot.tasks.onWrite` calls `ingest`. A new Bot match returns `{ ownerBotId }`. A Workflow match starts the run inside the plugin and leaves the card's owner to the host. A host that offers no `tasks.onWrite` warns once and reports `tasksWired: false` in `workflow/state`, and the Workflows panel says that a board card cannot start a run there (#1462). diff --git a/plugins/workflow/src/service.js b/plugins/workflow/src/service.js index bb1a291d7..22b9c9a86 100644 --- a/plugins/workflow/src/service.js +++ b/plugins/workflow/src/service.js @@ -961,8 +961,11 @@ export function createWorkflowService({ Object.keys(state.routes).forEach(taskId => { if (state.routes[taskId]?.id === workflowId) delete state.routes[taskId] }) + // #1632: a claim is judged on its own record. The routes above are gone + // by now, and a claim and a route do not hold the same keys anyway. Object.keys(state.claims ?? {}).forEach(taskId => { - if (state.routes[taskId]?.id === workflowId) delete state.claims[taskId] + const claim = state.claims[taskId] + if (claim?.kind === 'workflow' && claim.id === workflowId) delete state.claims[taskId] }) write() return present() @@ -1711,6 +1714,13 @@ function migrateRuns(state) { // the store's only record of those Bots, so an upgrade keeps them (deduped and // capped) and reap() retries them afterwards. noteOrphans(state, []) + // #1632: a store written while delete() left its claims behind holds claims + // for workflows that no longer exist, and ingest() would trust them forever. + for (const [taskId, claim] of Object.entries(state.claims)) { + if (claim?.kind === 'workflow' && !(state.workflows ?? []).some(item => item?.id === claim.id)) { + delete state.claims[taskId] + } + } if (state.laneFormat === true) return for (const [id, run] of Object.entries(state.runs ?? {})) { if (!run || typeof run !== 'object' || state.lanes[id]) continue diff --git a/plugins/workflow/test/delete-claims.test.js b/plugins/workflow/test/delete-claims.test.js new file mode 100644 index 000000000..8941886e2 --- /dev/null +++ b/plugins/workflow/test/delete-claims.test.js @@ -0,0 +1,129 @@ +// #1632: deleting a workflow must take its task claims with it. The claim is +// what survives on disk after the run is gone, and `ingest` trusts any claim, so +// a dangling one hands the deleted workflow's id to `assign()` on every retry — +// forever, across reloads. +import assert from 'node:assert/strict' +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createWorkflowService } from '../src/service.js' + +const CLOCK = 1_700_000_000_000 +const roster = [ + { id: 'main', name: 'Chief', role: 'lead', isMain: true, remote: false }, +] + +function readStore(dir) { + return JSON.parse(readFileSync(join(dir, 'workflows.json'), 'utf8')) +} + +function serviceAt(dir, bots = () => roster) { + let seq = 0 + return createWorkflowService({ + dir, + persist: true, + now: () => CLOCK, + bots, + spawn: async () => `eph-${++seq}`, + release: async () => {}, + dispatch: async () => ({ ok: true }), + // The step deadline is a timer, not part of what this case asserts. + setTimer: () => 0, + clearTimer: () => {}, + }) +} + +function leaky(svc) { + const created = svc.create({ + name: 'Leaky', + nodes: [{ id: 'only', name: 'Only', prompt: 'Do it.' }], + edges: [], + }) + assert.equal(created.ok, true) + return created.workflows.find(item => item.name === 'Leaky') +} + +test('#1632: delete takes the task claim with it, so the next ingest matches again', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-claim-delete-')) + try { + const svc = serviceAt(dir) + const workflow = leaky(svc) + const first = await svc.ingest({ id: 'task-9', status: 'open', title: 'Ship it', workflowId: workflow.id }) + assert.equal(first.ok, true) + assert.equal(first.matched, true) + assert.deepEqual(readStore(dir).claims['task-9'], { kind: 'workflow', id: workflow.id }, 'the run recorded its claim') + + await svc.delete({ workflowId: workflow.id }) + const store = readStore(dir) + assert.equal(store.claims['task-9'], undefined, 'a deleted workflow must not keep a claim on disk') + assert.equal(store.routes['task-9'], undefined, 'its route goes with it') + + const again = await svc.ingest({ id: 'task-9', status: 'open', title: 'Ship it' }) + assert.notEqual(again.error, 'No such workflow', 'the stale claim must not name a deleted workflow') + assert.deepEqual({ ok: again.ok, matched: again.matched }, { ok: true, matched: false }, 'it is matched again, not stuck') + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('#1632: the healed task still matches after a reload', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-claim-reload-')) + try { + const first = serviceAt(dir) + const workflow = leaky(first) + await first.ingest({ id: 'task-9', status: 'open', title: 'Ship it', workflowId: workflow.id }) + await first.delete({ workflowId: workflow.id }) + assert.equal(readStore(dir).claims['task-9'], undefined, 'nothing dangling reached the disk') + + const reloaded = serviceAt(dir) + const again = await reloaded.ingest({ id: 'task-9', status: 'open', title: 'Ship it' }) + assert.notEqual(again.error, 'No such workflow', 'a reload must not resurrect the deleted workflow id') + assert.deepEqual({ ok: again.ok, matched: again.matched }, { ok: true, matched: false }) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('#1632: a store an older delete already poisoned drops the dangling claim on load', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-claim-poison-')) + try { + const svc = serviceAt(dir) + const workflow = leaky(svc) + await svc.ingest({ id: 'task-9', status: 'open', title: 'Ship it', workflowId: workflow.id }) + // The store an older version wrote: the workflow row is gone, its claim stayed. + const store = readStore(dir) + store.workflows = store.workflows.filter(item => item.id !== workflow.id) + writeFileSync(join(dir, 'workflows.json'), JSON.stringify(store)) + + const reloaded = serviceAt(dir) + const again = await reloaded.ingest({ id: 'task-9', status: 'open', title: 'Ship it' }) + assert.notEqual(again.error, 'No such workflow', 'loading must drop a claim whose workflow is gone') + assert.deepEqual({ ok: again.ok, matched: again.matched }, { ok: true, matched: false }) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('#1632: a Bot claim that shares the id is not the delete\'s business', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-claim-kind-')) + try { + const bots = [...roster] + const svc = serviceAt(dir, () => bots) + const workflow = leaky(svc) + // A Bot whose id is the same string: judging only by id would drop this one. + bots.push({ id: workflow.id, name: 'Twin', role: 'Twin', remote: false }) + const routed = await svc.assign({ botId: workflow.id, taskId: 'task-bot', title: 'Owned by a Bot' }) + assert.equal(routed.ok, true) + assert.deepEqual(readStore(dir).claims['task-bot'], { kind: 'bot', id: workflow.id }) + + await svc.delete({ workflowId: workflow.id }) + assert.deepEqual( + readStore(dir).claims['task-bot'], + { kind: 'bot', id: workflow.id }, + 'only claims that name the workflow are removed', + ) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +})