End workflow zombie runs on dispose and reap

changelog.d/1414-workflow-dispose-zombie.fixed.md

@@ -0,0 +1 @@
+Plugin unload no longer persists workflow runs as `running` after their ephemeral Bot map is cleared; the next load also ends live runs that were already left in that zombie shape ([#1414](https://github.com/ThinkFlowLab/FlyBot/issues/1414)).

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. 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. 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 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/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. `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. |
@@ -46,7 +46,7 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher,
 
 1. Friends' Bots are not nodes. A `remote` Bot is refused by create, define and the matcher.
 2. The canvas and the run share one graph: nodes, edges and positions. A drag commits `workflow/graph`. Status highlighting reads that same graph. Replacing edges does not recompute who is already running.
-3. **A live run advances only when its ephemeral Bot replies, following the graph's edges in order.** That reply advances the run and may appear on the workflow stage log; it is not delivered into a persistent sidebar Bot conversation (for example Chief). The `workflow-step` send middleware denies those replies after `stepResult` runs. Nodes start only when their forward predecessors finish and edges reach them. There is no scripted or seeded run path: `workflow/state` exposes only `live === true` runs. A run does not create a schedule. `scheduleLabel` is display text. When the host can create Bots, starting a live run (`assign`, `run_workflow`, `ingest`) spawns one ephemeral Bot per node from that node's prompt, role, model and tools, sends it the step from the Main Bot, and removes every instance when the run finishes, fails for good, or is reset. A `failOnce` retry keeps them. They are hidden and left out of the sidebar. Loading the plugin does not spawn Bots for the default workflows. Loading is also the one moment a run the host left behind is noticed: a run still marked `running` is abandoned, so loading releases its instances and records the run as `failed`. Nothing resumes it. A run that already ended keeps its own status (#1388). A dispatch that does not land is not marked handed. The chat names the code, and the next hand-off tries that node again. It is not sent twice in the same hand-off. A reply that says `could not answer your message` or `model call failed` fails that node and does not start its successor. A 429 or rate-limit mention in any other reply does not.
+3. **A live run advances only when its ephemeral Bot replies, following the graph's edges in order.** That reply advances the run and may appear on the workflow stage log; it is not delivered into a persistent sidebar Bot conversation (for example Chief). The `workflow-step` send middleware denies those replies after `stepResult` runs. Nodes start only when their forward predecessors finish and edges reach them. There is no scripted or seeded run path: `workflow/state` exposes only `live === true` runs. A run does not create a schedule. `scheduleLabel` is display text. When the host can create Bots, starting a live run (`assign`, `run_workflow`, `ingest`) spawns one ephemeral Bot per node from that node's prompt, role, model and tools, sends it the step from the Main Bot, and removes every instance when the run finishes, fails for good, or is reset. A `failOnce` retry keeps them. They are hidden and left out of the sidebar. Loading the plugin does not spawn Bots for the default workflows. Plugin unload (`dispose`) and the load-time `reap` both end a live run still marked `running`: in-flight nodes are failed, the run is archived as `failed`, and any ephemeral Bot ids are released. A run already written as `running` with no ephemeral map (an older unload) is healed the same way on the next load (#1414). Loading is also the one moment a run the host left behind is noticed when its ephemeral map is still present: `reap` releases those instances and records the run as `failed`. Nothing resumes it. A run that already ended keeps its own status (#1388). A dispatch that does not land is not marked handed. The chat names the code, and the next hand-off tries that node again. It is not sent twice in the same hand-off. A reply that says `could not answer your message` or `model call failed` fails that node and does not start its successor. A 429 or rate-limit mention in any other reply does not.
 4. Each workflow keeps a FIFO queue and up to `maxConcurrent` live runs (default 2, an integer from 1 to 8 on the record or via `workflow/concurrency`). The host runs at most 6 live runs at once. A task that does not fit is enqueued. When a run finishes, every workflow with a free slot is drained, oldest enqueue first, until the host cap. 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.
 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 <name>: <value>`. The Bot reports `OUTPUT <name>: <value>`; 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. `tools` on the node is what that Bot may use: `github` grants `mcp__github__*` (pull requests, reviews, comments); `bash` or `shell` sets the permission to `danger-full-access` and allows a full-access shell; any other step gets `workspace-write`. 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`.

plugins/workflow/src/service.js

@@ -266,7 +266,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 })
@@ -279,7 +279,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 }
@@ -378,7 +378,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()
@@ -399,6 +399,28 @@
     write()
   }
 
+  // A live run the host cut off: fail in-flight nodes and mark the run failed (#1414,
+  // #1424). Ephemeral ids are left on the record until release strips them.
+  function failCutoffRun(item, run) {
+    if (run.status === 'done' || run.status === 'failed') return
+    for (const nodeId of item ? orderedIds(item, run.nodeStatus, 'running') : []) {
+      try {
+        run.nodeStatus = failNode(item, run.nodeStatus, nodeId).status
+      } catch (error) {
+        warn(`flybot-workflow: node ${nodeId} kept its status: ${error?.message ?? error}`)
+      }
+    }
+    run.status = 'failed'
+    run.updatedAt = now()
+  }
+
+  function archiveCutoffRun(workflowId, run) {
+    const lane = laneOf(workflowId)
+    lane.active = lane.active.filter(item => item.id !== run.id)
+    lane.last = run
+    lane.history = [...lane.history, { task: run.task, status: run.status, at: run.updatedAt }].slice(-20)
+  }
+
   // The oldest queued task whose own workflow still has a slot, or null.
   function takeNext() {
     let best = null
@@ -964,52 +986,80 @@
       if (typeof release !== 'function') return Promise.resolve()
       const stale = []
       const abandoned = []
+      let wrote = false
       for (const [workflowId, lane] of Object.entries(state.lanes ?? {})) {
         const seen = new Set()
         for (const run of [...(lane.active ?? []), lane.last].filter(Boolean)) {
-          if (!run?.ephemeral || seen.has(run)) continue
+          if (seen.has(run)) continue
           seen.add(run)
+          const hasEphemeral = run?.ephemeral && Object.keys(run.ephemeral).length > 0
+          const zombie = run?.live === true && run.status === 'running' && !hasEphemeral
+          if (!hasEphemeral && !zombie) continue
           // A run still marked running here was cut off with the host that ran it,
           // and nothing resumes it: its instances are abandoned, not in progress.
           // Release them and end the run, or they stay hidden with no way to remove
-          // them. A run that already ended keeps its own status (#1388).
+          // them. A run that already ended keeps its own status (#1388). A zombie
+          // is a run dispose() already stripped: heal it on load (#1414).
           if (run.status === 'running') {
             abandoned.push({ workflowId, run })
             continue
           }
-          stale.push(...Object.values(run.ephemeral))
+          if (hasEphemeral) {
+            stale.push(...Object.values(run.ephemeral))
+            delete run.ephemeral
+            delete run.handed
+            wrote = true
+          }
+        }
+      }
+      for (const { workflowId, run } of abandoned) {
+        try {
+          if (run.ephemeral) stale.push(...Object.values(run.ephemeral))
+          failCutoffRun(workflow(workflowId), run)
           delete run.ephemeral
           delete run.handed
+          archiveCutoffRun(workflowId, run)
+          wrote = true
+        } catch (error) {
+          warn(`flybot-workflow: could not end ${run.id}: ${error?.message ?? error}`)
         }
       }
-      for (const { workflowId, run } of abandoned) {
-        stale.push(...Object.values(run.ephemeral))
-        const ended = { ...run, status: 'failed', updatedAt: now() }
-        delete ended.ephemeral
-        delete ended.handed
-        const lane = laneOf(workflowId)
-        lane.active = lane.active.filter(item => item.id !== run.id)
-        lane.last = ended
-        lane.history = [...lane.history, { task: ended.task, status: ended.status, at: ended.updatedAt }].slice(-20)
+      if (stale.length === 0 && !wrote) return Promise.resolve()
+      if (stale.length === 0) {
+        write()
+        return Promise.resolve()
       }
-      if (stale.length === 0 && abandoned.length === 0) return Promise.resolve()
       return releaseAll(stale, release, warn).then(() => { write() })
     },
     dispose() {
-      const runs = Object.values(state.lanes).flatMap(lane => [...(lane.active ?? []), lane.last].filter(Boolean))
-      const ids = runs.flatMap(run => Object.values(run?.ephemeral ?? {}))
-      // 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.
+      const ids = []
+      const runs = []
+      let endedLive = false
+      for (const [workflowId, lane] of Object.entries(state.lanes ?? {})) {
+        for (const run of [...(lane.active ?? [])]) {
+          if (run?.live === true && run.status === 'running') {
+            failCutoffRun(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
+      // record drops ephemeral so nothing is persisted as running without a map
+      // (#1414). 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.
+      const shouldWrite = ids.length > 0 || endedLive
       return releaseAll(ids, release, warn).then(() => {
         for (const run of runs) {
           delete run.ephemeral
           delete run.handed
         }
-        if (ids.length > 0) write()
+        if (shouldWrite) write()
       })
     },
   }

plugins/workflow/test/dispose.test.js

@@ -25,8 +25,10 @@ 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 storedRun(store, workflowId, runId) {
+  const lane = store.lanes?.[workflowId]
+  if (!lane) return undefined
+  return (lane.active ?? []).find(run => run.id === runId) ?? (lane.last?.id === runId ? lane.last : undefined)
 }
 
 function deferred() {
@@ -83,8 +85,8 @@ test('#1387: dispose() releases the Bots before the record drops their ids', asy
 
     const before = storeText(dir)
     const stored = readStore(dir)
-    const rowA = activeRun(stored, first.id, runA.id)
-    const rowB = activeRun(stored, second.id, runB.id)
+    const rowA = storedRun(stored, first.id, runA.id)
+    const rowB = storedRun(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)
@@ -96,7 +98,7 @@ test('#1387: dispose() releases the Bots before the record drops their ids', asy
     // 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)
+      const row = storedRun(during, workflowId, run.id)
       assert.deepEqual(row.ephemeral, run.ephemeral)
       assert.equal(row.handed.plan, true)
     }
@@ -111,9 +113,11 @@ test('#1387: dispose() releases the Bots before the record drops their ids', asy
 
     const after = readStore(dir)
     for (const [workflowId, run] of [[first.id, runA], [second.id, runB]]) {
-      const row = activeRun(after, workflowId, run.id)
+      const row = storedRun(after, workflowId, run.id)
+      assert.equal(row.status, 'failed')
       assert.equal(row.ephemeral, undefined)
       assert.equal('handed' in row, false)
+      assert.equal((after.lanes?.[workflowId]?.active ?? []).length, 0)
     }
     assert.deepEqual([...released].sort(), [...ids].sort())
     assert.equal(released.length, 4)
@@ -144,7 +148,11 @@ test('#1387: dispose() still returns a promise and still writes when nothing can
     const disposed = svc.dispose()
     assert.equal(typeof disposed?.then, 'function')
     await disposed
-    const row = readStore(dir).lanes[made.id].active[0]
+    const lane = readStore(dir).lanes[made.id]
+    assert.equal(lane.active.length, 0)
+    const row = lane.last
+    assert.equal(row.status, 'failed')
+    assert.equal(row.nodeStatus.plan, 'failed')
     assert.equal(row.ephemeral, undefined)
     assert.equal('handed' in row, false)
 
@@ -165,6 +173,47 @@ test('#1387: dispose() still returns a promise and still writes when nothing can
   }
 })
 
+test('#1414: dispose() ends live runs before the record drops ephemeral', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-dispose-1414-'))
+  const released = []
+  try {
+    const svc = createWorkflowService({
+      dir,
+      persist: true,
+      now: () => CLOCK,
+      bots: () => roster,
+      spawn: async (spec) => `eph-${spec.nodeId}`,
+      release: async (id) => { released.push(id) },
+      dispatch: async () => ({ ok: true }),
+    })
+    const made = pair('Lane', svc)
+    await svc.runNamed({ workflow: 'Lane', brief: 'Ship it', task: '#1414' })
+    await svc.dispose()
+    const lane = readStore(dir).lanes[made.id]
+    assert.equal(lane.active.length, 0)
+    assert.equal(lane.last.status, 'failed')
+    assert.equal(lane.last.nodeStatus.plan, 'failed')
+    assert.equal(lane.last.ephemeral, undefined)
+    assert.deepEqual(released.sort(), ['eph-code', 'eph-plan'])
+
+    const after = createWorkflowService({
+      dir,
+      persist: true,
+      now: () => CLOCK + 1,
+      bots: () => roster,
+      spawn: async (spec) => `eph2-${spec.nodeId}`,
+      release: async () => {},
+      dispatch: async () => ({ ok: true }),
+    })
+    assert.equal(after.state().workflows.find(item => item.id === made.id).activeCount, 0)
+    released.length = 0
+    await after.reap()
+    assert.deepEqual(released, [])
+  } 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