fix(workflow): settle a run's node map when the run is ended (#1424)

changelog.d/1424-workflow-reap-node-status.fixed.md

@@ -0,0 +1 @@
+A run ended by a restart, by a step that threw, or by stopping it kept the nodes it had been running: the run said `failed` while the flow strip still painted "Running now" for that node. Ending a run now marks those nodes `failed` with it, from one helper every path goes through ([#1424](https://github.com/ThinkFlowLab/FlyBot/issues/1424)).

docs/subsystems/workflow.md

@@ -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`, but only while **Edit** is on — a factory preset is copied first (#1378). 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. 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.
+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). `workflow/reset` stopping one live run ends it the same way before its Bots are released: nodes still `running` become `failed`, so the strip cannot keep painting a run that is over (#1424). 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. `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 <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. **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`.
@@ -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, 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 (including factory-id rows such as `delivery`), 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, a stopped run with its settled node map (#1424), the reply middleware, the client canvas, the settings page (including a condition and a loop), rename and delete (including factory-id rows such as `delivery`), 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/src/service.js

@@ -267,7 +267,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 })
@@ -280,7 +280,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 }
@@ -379,7 +379,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()
@@ -400,10 +400,14 @@
     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
+  // End a run the host gave up on: fail the nodes still in flight, then mark the
+  // run failed. Every path that abandons a run comes through here — reap() at load,
+  // dispose(), the catch around a step that threw, and reset() stopping one run —
+  // so run.status and run.nodeStatus cannot drift apart and a client stops painting
+  // "Running now" (#1424). A run that already ended keeps its own ending (#1388);
+  // ephemeral ids stay on the record until release strips them.
+  function endRun(item, run) {
+    if (run.status === 'done' || run.status === 'failed') return run
     for (const nodeId of item ? orderedIds(item, run.nodeStatus, 'running') : []) {
       try {
         run.nodeStatus = failNode(item, run.nodeStatus, nodeId).status
@@ -413,6 +417,7 @@
     }
     run.status = 'failed'
     run.updatedAt = now()
+    return run
   }
 
   function archiveCutoffRun(workflowId, run) {
@@ -540,7 +545,8 @@
       try {
         run = await handOff(item, run)
       } catch (error) {
-        await releaseRun(item.id, { ...run, status: 'failed', updatedAt: now() })
+        endRun(item, run)
+        await releaseRun(item.id, run)
         return refuse(error)
       }
     }
@@ -869,7 +875,9 @@
           const run = lane.active.find(itemRun => itemRun.id === runId)
           if (!run || run.live !== true || run.status !== 'running') return { ok: false, error: 'No live run to stop' }
           const ids = Object.values(run.ephemeral ?? {})
-          const ended = { ...run, status: 'failed', updatedAt: now() }
+          // Stopping one run ends it the same way a reap does: the node map settles
+          // with the run, or the strip keeps painting the node it was on (#1424).
+          const ended = { ...endRun(item, run) }
           delete ended.ephemeral
           delete ended.handed
           lane.active = lane.active.filter(itemRun => itemRun.id !== runId)
@@ -1059,7 +1067,7 @@
       for (const { workflowId, run } of abandoned) {
         try {
           if (run.ephemeral) stale.push(...Object.values(run.ephemeral))
-          failCutoffRun(workflow(workflowId), run)
+          endRun(workflow(workflowId), run)
           delete run.ephemeral
           delete run.handed
           archiveCutoffRun(workflowId, run)
@@ -1080,9 +1088,9 @@
       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)
+            endRun(workflow(workflowId), run)
             archiveCutoffRun(workflowId, run)
             endedLive = true
           }

plugins/workflow/test/ephemeral.test.js

@@ -586,3 +586,125 @@ test('the plugin load itself reaps a run the restart left running (#1388)', asyn
     rmSync(dir, { recursive: true, force: true })
   }
 })
+
+// #1424: ending a run has to settle the node map with it. `publicRun` exposes
+// `running` from `nodeStatus`, and the strip paints "Running now" from that, so a
+// run that says `failed` while a node still says `running` is a ghost on screen.
+test('stopping a live run settles its node map: no node stays running (#1424)', async () => {
+  const { svc, spawned, released } = harness()
+  const made = lane(svc).workflows.find(item => item.name === 'Lane')
+  const started = await svc.runNamed({ workflow: 'Lane', brief: 'Ship the note' })
+  assert.equal(started.ok, true)
+  const live = started.workflows.find(item => item.id === made.id).run
+  assert.equal(live.status, 'running')
+  assert.deepEqual(live.running, ['plan'])
+
+  const stopped = await svc.reset({ workflowId: made.id, runId: live.id })
+  assert.equal(stopped.ok, true)
+  const run = stopped.workflows.find(item => item.id === made.id).run
+  assert.equal(run.status, 'failed')
+  assert.deepEqual(run.running, [])
+  assert.equal(run.nodeStatus.plan, 'failed')
+  assert.equal(run.nodeStatus.code, 'queued')
+  assert.deepEqual(released.sort(), spawned.map(entry => entry.id).sort())
+
+  // The ending reaches the list the stage renders, not just the run record.
+  const history = stopped.workflows.find(item => item.id === made.id).history
+  assert.equal(history.at(-1).status, 'failed')
+  assert.equal(history.at(-1).at, run.updatedAt)
+})
+
+test('a reaped run whose node map names a node the graph no longer has settles the rest (#1424)', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-reap-1424-ghost-'))
+  const released = []
+  const clock = () => 1_700_000_000_000
+  try {
+    const svc = persistent(dir, released, clock)
+    const made = svc.define({
+      name: 'Lane',
+      nodes: [{ id: 'plan', name: 'Planner', role: 'plan', prompt: 'Write the plan.' }],
+      edges: [],
+    }).workflows.find(item => item.name === 'Lane')
+    const store = readStore(dir)
+    const lane = store.lanes[made.id] ?? { active: [], queue: [], last: null, history: [] }
+    lane.active = [...(lane.active ?? []), {
+      id: 'run-ghost',
+      workflowId: made.id,
+      task: null,
+      live: true,
+      status: 'running',
+      nodeStatus: { plan: 'running', gone: 'running' },
+      steps: 0,
+      stepBudget: 2,
+      failedOnce: {},
+      history: [],
+      startedAt: 4,
+      updatedAt: 5,
+      ephemeral: { plan: 'eph-plan', gone: 'eph-gone' },
+    }]
+    store.lanes[made.id] = lane
+    writeFileSync(join(dir, 'workflows.json'), JSON.stringify(store))
+
+    const after = persistent(dir, released, clock)
+    await after.reap()
+    assert.deepEqual(released.sort(), ['eph-gone', 'eph-plan'])
+    const run = after.state().workflows.find(item => item.id === made.id).run
+    assert.equal(run.status, 'failed')
+    assert.equal(run.nodeStatus.plan, 'failed')
+    // Nothing to move: the graph no longer has that node, so it keeps its record
+    // and the strip, which walks the graph, does not paint it.
+    assert.equal(run.nodeStatus.gone, 'running')
+    assert.deepEqual(run.running, [])
+    assert.equal(run.ephemeral, undefined)
+    assert.equal(run.handed, undefined)
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('an ended run keeps its ending when reap drops the records it still holds (#1424)', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-reap-1424-done-'))
+  const released = []
+  const clock = () => 1_700_000_000_000
+  try {
+    const svc = persistent(dir, released, clock)
+    const made = svc.define({
+      name: 'Lane',
+      nodes: [{ id: 'only', name: 'Planner', role: 'plan', prompt: 'Say it once.' }],
+      edges: [],
+    }).workflows.find(item => item.name === 'Lane')
+    const store = readStore(dir)
+    const lane = store.lanes[made.id] ?? { active: [], queue: [], last: null, history: [] }
+    lane.last = {
+      id: 'run-done',
+      workflowId: made.id,
+      task: null,
+      live: true,
+      status: 'done',
+      nodeStatus: { only: 'done' },
+      steps: 1,
+      stepBudget: 2,
+      failedOnce: {},
+      history: [{ task: null, status: 'done', at: 5 }],
+      startedAt: 4,
+      updatedAt: 6,
+      ephemeral: { only: 'eph-only' },
+    }
+    store.lanes[made.id] = lane
+    writeFileSync(join(dir, 'workflows.json'), JSON.stringify(store))
+
+    const after = persistent(dir, released, clock)
+    await after.reap()
+    assert.deepEqual(released, ['eph-only'])
+    const run = after.state().workflows.find(item => item.id === made.id).run
+    assert.equal(run.status, 'done')
+    assert.equal(run.updatedAt, 6)
+    assert.equal(run.nodeStatus.only, 'done')
+    assert.deepEqual(run.running, [])
+    assert.equal(run.history.length, 1)
+    assert.equal(run.history[0].status, 'done')
+    assert.equal(run.ephemeral, undefined)
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})