fix(workflow): release run Bots before dispose() drops their ids (#1387)

changelog.d/1387-workflow-dispose-order.fixed.md

@@ -0,0 +1 @@
+Disposing the Workflow plugin now releases every run Bot before the saved record drops its id, and hands that release back as a promise the host waits on, so a slow or interrupted shutdown can no longer strand hidden Bots that no state can find or remove. A released run Bot also gives its name back instead of keeping it for the lifetime of the process ([#1387](https://github.com/ThinkFlowLab/FlyBot/issues/1387)).

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 reports a step result. |
 | `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. A `when` edge chooses a branch; the other is skipped. A required input with no upstream provider is refused before a run starts. |
-| `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. Both refuse factory presets. A spawn hook makes one run Bot per node. A dispatch that does not land is not marked handed. |
+| `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. Both refuse factory presets. A spawn hook makes one run Bot per node. Disposing the plugin releases every run Bot first 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 that is cut short cannot leave hidden Bots that no state names (#1387). 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. `permissionFor` picks `danger-full-access` when the node names a shell, otherwise `workspace-write`. |
 | `plugins/workflow/src/match.js` | Which open card matches a Workflow or a local Bot. |
 | `plugins/workflow/src/template.js` | 发布交付流水线 (a parallel branch) and the weekly report, each node with the inputs and outputs that edge carries. |
@@ -59,4 +59,4 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher,
 
 ## Tests
 
-`plugins/workflow/test/`: graph advance and join, graph edit, assign and persistence, the FIFO queue and concurrency cap, Bot-reply advance, a dispatch that fails without being marked handed, assign-to-bot versus assign-to-workflow, inputs and outputs along edges, a loop that breaks and an if/else branch, templates, `define` / `runNamed`, the matcher, ephemeral spawn and cleanup, the reply middleware, the client canvas, the settings page (including a condition and a loop), rename and delete (factory refuse, cleanup on delete, #1434), and a sidebar click that leaves the Workflows section in place. Run `npm test` in `plugins/workflow`. The host seam that applies `ownerBotId` is `core/test/task-route.test.js`. The host omission of run Bots, and the grant of `stepTools` plus the hidden-approval answer, is `core/test/flybot.test.js`. Which connectors a run Bot may call is `core/test/connector-access.test.js`.
+`plugins/workflow/test/`: graph advance and join, graph edit, assign and persistence, the FIFO queue and concurrency cap, Bot-reply advance, a dispatch that fails without being marked handed, disposing releases the Bots before the record drops their ids and a released runner name is reusable again, assign-to-bot versus assign-to-workflow, inputs and outputs along edges, a loop that breaks and an if/else branch, templates, `define` / `runNamed`, the matcher, ephemeral spawn and cleanup, the reply middleware, the client canvas, the settings page (including a condition and a loop), rename and delete (factory refuse, cleanup on delete, #1434), and a sidebar click that leaves the Workflows section in place. Run `npm test` in `plugins/workflow`. The host seam that applies `ownerBotId` is `core/test/task-route.test.js`. The host omission of run Bots, and the grant of `stepTools` plus the hidden-approval answer, is `core/test/flybot.test.js`. Which connectors a run Bot may call is `core/test/connector-access.test.js`.

plugins/workflow/index.js

@@ -25,7 +25,10 @@ export function apply(ctx, config = {}) {
   const fb = ctx.flybot.scope('flybot-workflow')
   const dshHome = process.env.DSH_HOME ?? join(process.env.HOME ?? '.', '.dsh')
   const persist = config.persist !== false
-  const liveNames = new Set()
+  // The runner names held by the ephemeral Bots this plugin spawned, keyed by
+  // Bot id, so a release gives its name back instead of keeping it for the
+  // lifetime of the process (#1387).
+  const liveNames = new Map()
   const lifecycle = typeof fb.bot?.create === 'function' ? {
     async spawn(spec) {
       const tools = Array.isArray(spec.tools) ? spec.tools : []
@@ -35,7 +38,7 @@ export function apply(ctx, config = {}) {
       const prompt = [spec.prompt, ...notes].filter(Boolean).join('\n\n')
       let last
       for (let n = 0; n < 4; n += 1) {
-        const taken = new Set([...(fb.bot.list?.() ?? []).map(bot => bot.name), ...liveNames])
+        const taken = new Set([...(fb.bot.list?.() ?? []).map(bot => bot.name), ...liveNames.keys()])
         const name = runnerName(spec.name, taken)
         try {
           const id = await fb.bot.create({
@@ -50,7 +53,7 @@ export function apply(ctx, config = {}) {
             permission: permissionFor(tools),
             fields: { nodeId: spec.nodeId, tools },
           })
-          liveNames.add(name)
+          liveNames.set(name, id)
           return id
         } catch (error) {
           last = error
@@ -60,6 +63,9 @@ export function apply(ctx, config = {}) {
       throw last
     },
     release(id) {
+      // The instance is going away, so its name is free again. Without this the
+      // allocator treats every name this process ever handed out as taken.
+      for (const [name, owner] of liveNames) if (owner === id) liveNames.delete(name)
       if (typeof fb.bot.remove !== 'function') return undefined
       return fb.bot.remove(id)
     },
@@ -150,7 +156,13 @@ export function apply(ctx, config = {}) {
       return null
     }))
   }
-  const dispose = () => { service.dispose(); for (const off of offs.splice(0)) off() }
+  // The hooks come off at once; the Bot release is behind a promise, so hand it
+  // back instead of dropping it (#1387).
+  const dispose = () => {
+    const released = service.dispose()
+    for (const off of offs.splice(0)) off()
+    return released
+  }
   if (typeof ctx.effect === 'function') ctx.effect(() => dispose, 'flybot-workflow')
   else ctx.on?.('dispose', dispose)
   return { service, enabled: true, dispose }

plugins/workflow/src/service.js

@@ -265,7 +265,7 @@
       const missing = unmetRequired(item, id, run.produced, run.task)
       if (missing.length === 0) { started.push(id); continue }
       const preds = item.edges.filter(edge => edge.to === id && edge.loop !== true).map(edge => edge.from)
       const stuck = preds.length === 0 || preds.every(pred => status[pred] === 'done' || status[pred] === 'failed' || status[pred] === 'skipped')
       if (stuck) {
         status[id] = 'failed'
         blocked.push({ id, missing })
@@ -278,7 +278,7 @@
   function applyOp(item, run, op) {
     let working = run
     if (op.type === 'complete' && op.outputs && typeof op.outputs === 'object') {
       working = { ...run, produced: { ...(run.produced ?? {}), [op.nodeId]: op.outputs } }
     }
     let result
     if (op.type === 'start') result = { ...startFrontier(item, working.nodeStatus, working.fired), joined: false, layerSize: 0 }
@@ -377,7 +377,7 @@
         push(item.id, [{ kind: 'system', tone: 'bad', text: `${node.name} 没有接到这一步（${code}）。` }], now())
         continue
       }
       current = { ...current, handed: { ...current.handed, [nodeId]: true }, inputs: { ...(current.inputs ?? {}), [nodeId]: body.inputs } }
       storeRun(item.id, current)
     }
     write()
@@ -999,12 +999,19 @@
     dispose() {
       const runs = Object.values(state.lanes).flatMap(lane => [...(lane.active ?? []), lane.last].filter(Boolean))
       const ids = runs.flatMap(run => Object.values(run?.ephemeral ?? {}))
-      for (const run of runs) {
-        delete run.ephemeral
-        delete run.handed
-      }
-      if (ids.length > 0) write()
-      void releaseAll(ids, release, warn)
+      // Release first, then forget the ids (#1387). A record that drops an id
+      // before its Bot is gone leaves a hidden Bot that no state names, so
+      // nothing can ever find it again. releaseAll reports a failed id through
+      // `warn` instead of rejecting, so one bad release cannot skip the write.
+      // The caller awaits this: a disposer that returns before the Bots are
+      // gone is not a cleanup.
+      return releaseAll(ids, release, warn).then(() => {
+        for (const run of runs) {
+          delete run.ephemeral
+          delete run.handed
+        }
+        if (ids.length > 0) write()
+      })
     },
   }
 }

plugins/workflow/test/dispose.test.js

@@ -0,0 +1,216 @@
+// #1387: dispose() releases the run Bots *before* the record drops their ids,
+// and hands the host a promise it can await. A state file that forgets an id
+// while its Bot is still alive leaves a hidden Bot nobody can see or remove.
+import assert from 'node:assert/strict'
+import crypto from 'node:crypto'
+import { mkdtempSync, readFileSync, rmSync } from 'node:fs'
+import { syncBuiltinESMExports } from 'node:module'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import { apply } from '../index.js'
+import { createWorkflowService } from '../src/service.js'
+
+const roster = [
+  { id: 'main', name: 'Chief', role: 'lead', isMain: true, remote: false },
+]
+
+const CLOCK = 1_700_000_000_000
+
+function readStore(dir) {
+  return JSON.parse(readFileSync(join(dir, 'workflows.json'), 'utf8'))
+}
+
+function storeText(dir) {
+  return readFileSync(join(dir, 'workflows.json'), 'utf8')
+}
+
+function activeRun(store, workflowId, runId) {
+  return (store.lanes?.[workflowId]?.active ?? []).find(run => run.id === runId)
+}
+
+function deferred() {
+  let resolve
+  const promise = new Promise(done => { resolve = done })
+  return { promise, resolve }
+}
+
+function pair(name, svc) {
+  return svc.define({
+    name,
+    nodes: [
+      { id: 'plan', name: 'Planner', role: 'plan', prompt: 'Write the plan.' },
+      { id: 'code', name: 'Coder', role: 'code', prompt: 'Implement it.' },
+    ],
+    edges: [{ from: 'plan', to: 'code' }],
+  }).workflows.find(item => item.name === name)
+}
+
+test('#1387: dispose() releases the Bots before the record drops their ids', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-dispose-'))
+  const released = []
+  const pending = deferred()
+  let seq = 0
+  try {
+    // Two runs, two nodes each: four instances, so the release really is a set.
+    const svc = createWorkflowService({
+      dir,
+      persist: true,
+      now: () => CLOCK,
+      bots: () => roster,
+      spawn: async (spec) => {
+        seq += 1
+        return `eph-${spec.nodeId}-${seq}`
+      },
+      // A slow release: nothing resolves until the test says so.
+      release: (id) => {
+        released.push(id)
+        return pending.promise
+      },
+      dispatch: async () => ({ ok: true }),
+    })
+    const first = pair('First', svc)
+    const second = pair('Second', svc)
+    const a = await svc.runNamed({ workflow: 'First', brief: 'Ship the note', task: '#1387a' })
+    const b = await svc.runNamed({ workflow: 'Second', brief: 'Ship the other', task: '#1387b' })
+    assert.equal(a.ok, true)
+    assert.equal(b.ok, true)
+    const runA = a.workflows.find(item => item.id === first.id).run
+    const runB = b.workflows.find(item => item.id === second.id).run
+    const ids = [...Object.values(runA.ephemeral), ...Object.values(runB.ephemeral)]
+    assert.equal(ids.length, 4)
+    assert.equal(new Set(ids).size, 4)
+
+    const before = storeText(dir)
+    const stored = readStore(dir)
+    const rowA = activeRun(stored, first.id, runA.id)
+    const rowB = activeRun(stored, second.id, runB.id)
+    assert.deepEqual(Object.values(rowA.ephemeral).sort(), Object.values(runA.ephemeral).sort())
+    assert.deepEqual(Object.values(rowB.ephemeral).sort(), Object.values(runB.ephemeral).sort())
+    assert.equal(rowA.handed.plan, true)
+    assert.equal(rowB.handed.plan, true)
+
+    const disposing = svc.dispose()
+
+    // The release is still in flight. The record must still name every instance,
+    // or an interrupted release strands Bots that no state can ever find again.
+    const during = readStore(dir)
+    for (const [workflowId, run] of [[first.id, runA], [second.id, runB]]) {
+      const row = activeRun(during, workflowId, run.id)
+      assert.deepEqual(row.ephemeral, run.ephemeral)
+      assert.equal(row.handed.plan, true)
+    }
+    assert.equal(storeText(dir) === before, true, 'dispose() rewrote the record before the release finished')
+    assert.equal(released.length > 0, true, 'dispose() must start the release')
+
+    // The host can await the release: dispose() is not fire-and-forget.
+    assert.equal(typeof disposing?.then, 'function')
+
+    pending.resolve()
+    await disposing
+
+    const after = readStore(dir)
+    for (const [workflowId, run] of [[first.id, runA], [second.id, runB]]) {
+      const row = activeRun(after, workflowId, run.id)
+      assert.equal(row.ephemeral, undefined)
+      assert.equal('handed' in row, false)
+    }
+    assert.deepEqual([...released].sort(), [...ids].sort())
+    assert.equal(released.length, 4)
+    assert.equal(new Set(released).size, 4)
+  } finally {
+    pending.resolve()
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('#1387: dispose() still returns a promise and still writes when nothing can be released', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-dispose-bare-'))
+  try {
+    // The host spawned Bots but has no release hook at all: the record still drops
+    // the ids, and the caller can still await the attempt.
+    const svc = createWorkflowService({
+      dir,
+      persist: true,
+      now: () => CLOCK,
+      bots: () => roster,
+      spawn: async (spec) => `eph-${spec.nodeId}`,
+      dispatch: async () => ({ ok: true }),
+    })
+    const made = pair('Bare', svc)
+    await svc.runNamed({ workflow: 'Bare', brief: 'Ship it' })
+    assert.deepEqual(Object.values(readStore(dir).lanes[made.id].active[0].ephemeral).sort(), ['eph-code', 'eph-plan'])
+
+    const disposed = svc.dispose()
+    assert.equal(typeof disposed?.then, 'function')
+    await disposed
+    const row = readStore(dir).lanes[made.id].active[0]
+    assert.equal(row.ephemeral, undefined)
+    assert.equal('handed' in row, false)
+
+    // No spawn hook either: there are no ids, nothing is written, but dispose()
+    // still resolves so every caller may await it the same way.
+    const offDir = join(dir, 'off')
+    const off = createWorkflowService({ dir: offDir, persist: true, now: () => CLOCK, bots: () => roster })
+    const idle = off.define({ name: 'Idle', nodes: [{ id: 'only', name: 'Planner', prompt: 'Wait.' }], edges: [] })
+      .workflows.find(item => item.name === 'Idle')
+    const text = storeText(offDir)
+    const quiet = off.dispose()
+    assert.equal(typeof quiet?.then, 'function')
+    await quiet
+    assert.equal(storeText(offDir), text)
+    assert.equal(off.state().workflows.find(item => item.id === idle.id).run, null)
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('#1387: a released runner name goes back into the pool', async (t) => {
+  // The plugin names a runner `<stem> <4 hex>` and skips any name a live hidden
+  // Bot still holds in this process. Hex is random, so a leaked name can only be
+  // told apart from a fresh one by fixing the bytes: 2 bytes -> `<stem> aaaa`,
+  // the 3-byte fallback after six collisions -> `<stem> bbbbbb`.
+  const bytes = crypto.randomBytes
+  crypto.randomBytes = size => Buffer.alloc(size, size === 2 ? 0xaa : 0xbb)
+  syncBuiltinESMExports()
+  t.after(() => {
+    crypto.randomBytes = bytes
+    syncBuiltinESMExports()
+  })
+  const names = []
+  const removed = []
+  const fb = {
+    bot: {
+      list: () => roster,
+      async create(spec) {
+        names.push(spec.name)
+        return `bot-${names.length}`
+      },
+      async remove(id) { removed.push(id); return true },
+      async send() { return { action: 'allow' } },
+    },
+    readOnly: () => false,
+    api: { handle() { return () => {} } },
+    tools: { register() { return () => {} } },
+    warn() {},
+  }
+  const started = apply({ flybot: { scope: () => fb }, effect(fn) { fn() } }, { persist: false })
+  const made = started.service.define({
+    name: 'One',
+    nodes: [{ id: 'plan', name: 'Planner', role: 'plan', prompt: 'Plan once.' }],
+    edges: [],
+  }).workflows.find(item => item.name === 'One')
+  const first = await started.service.runNamed({ workflow: 'One', brief: 'First' })
+  assert.equal(first.ok, true)
+  assert.deepEqual(names, ['Planner aaaa'])
+  const run = first.workflows.find(item => item.id === made.id).run
+
+  await started.service.advance({ workflowId: made.id, nodeId: 'plan', outcome: 'done' })
+  assert.deepEqual(removed, [run.ephemeral.plan])
+
+  const second = await started.service.runNamed({ workflow: 'One', brief: 'Second', task: '#1387b' })
+  assert.equal(second.ok, true)
+  // The name the released Bot held is reusable. A leak makes the allocator skip
+  // it six times and fall back to the 3-byte name.
+  assert.deepEqual(names, ['Planner aaaa', 'Planner aaaa'])
+})