diff --git a/changelog.d/1472-workflow-dispose-release-all.fixed.md b/changelog.d/1472-workflow-dispose-release-all.fixed.md new file mode 100644 index 000000000..cf539ed72 --- /dev/null +++ b/changelog.d/1472-workflow-dispose-release-all.fixed.md @@ -0,0 +1 @@ +Workflow: unloading the plugin now releases the run Bots of every live run in a workflow, not only the last one, so a workflow with two concurrent runs no longer leaves hidden Bots that no saved record names ([#1472](https://github.com/ThinkFlowLab/FlyBot/issues/1472)). diff --git a/docs/subsystems/workflow.md b/docs/subsystems/workflow.md index 909979abe..616b1980e 100644 --- a/docs/subsystems/workflow.md +++ b/docs/subsystems/workflow.md @@ -28,7 +28,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 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/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: it is retried twice with a short backoff, and a step still unanswered after `STEP_DEADLINE_MS` (10 minutes) fails the run and releases its Bots (#1530). | +| `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, 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 `STEP_DEADLINE_MS` (10 minutes) fails the run and releases its Bots (#1530). | | `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. | diff --git a/plugins/workflow/src/service.js b/plugins/workflow/src/service.js index 31f0c2650..6085a323c 100644 --- a/plugins/workflow/src/service.js +++ b/plugins/workflow/src/service.js @@ -1240,14 +1240,18 @@ export function createWorkflowService({ const runs = [] let endedLive = false for (const [workflowId, lane] of Object.entries(state.lanes ?? {})) { - for (const run of [...(lane.active ?? [])]) { + // Collect before archiving: archiveCutoffRun removes the run from + // `active` and makes it `last`, so a collection after the loop names + // only the run archived last and leaks every earlier run's Bots (#1472). + const live = [...(lane.active ?? [])] + runs.push(...[...live, lane.last].filter(Boolean)) + for (const run of live) { if (run?.live === true && run.status === 'running') { endRun(workflow(workflowId), run) archiveCutoffRun(workflowId, run) endedLive = true } } - runs.push(...[...(lane.active ?? []), lane.last].filter(Boolean)) } for (const run of runs) ids.push(...Object.values(run?.ephemeral ?? {})) // Release first, then forget the ids (#1387). End live runs before the diff --git a/plugins/workflow/test/dispose.test.js b/plugins/workflow/test/dispose.test.js index c7f945d95..a3974dbbd 100644 --- a/plugins/workflow/test/dispose.test.js +++ b/plugins/workflow/test/dispose.test.js @@ -263,3 +263,47 @@ test('#1387: a released runner name goes back into the pool', async (t) => { // it six times and fall back to the 3-byte name. assert.deepEqual(names, ['Planner aaaa', 'Planner aaaa']) }) + +test('#1472: dispose() releases the Bots of every concurrent live run, not just the last one', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-dispose-1472-')) + const released = [] + const spawned = [] + let seq = 0 + try { + // One lane at the default maxConcurrent of 2: two assigns start two live + // runs, each holding one instance per node. + const svc = createWorkflowService({ + dir, + persist: true, + now: () => CLOCK, + bots: () => roster, + spawn: async (spec) => { + seq += 1 + const id = `eph-${spec.nodeId}-${seq}` + spawned.push(id) + return id + }, + release: async (id) => { released.push(id) }, + dispatch: async () => ({ ok: true }), + }) + const made = pair('Lane', svc) + await svc.assign({ workflowId: made.id, taskId: '#1472a', title: 'First task' }) + await svc.assign({ workflowId: made.id, taskId: '#1472b', title: 'Second task' }) + assert.equal(svc.state().workflows.find(item => item.id === made.id).activeCount, 2) + assert.equal(spawned.length, 4) + assert.equal(new Set(spawned).size, 4) + + await svc.dispose() + + // Every instance the lane held is released, not only the run archived last. + assert.deepEqual([...released].sort(), [...spawned].sort()) + assert.equal(released.length, 4) + const lane = readStore(dir).lanes[made.id] + assert.equal(lane.active.length, 0) + assert.deepEqual(lane.history.map(row => row.status), ['failed', 'failed']) + // #1387's invariant: no id is dropped while its Bot may still be alive. + for (const id of spawned) assert.equal(storeText(dir).includes(id), false, id) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +})