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

A run ended by reap() at load, or by settle()'s dispatch catch, was marked
failed while its nodeStatus still said running for the node the host was on.
publicRun() derives `running` from nodeStatus, so the flow strip kept painting
"Running now" for a run that had ended and was not going to move again.

Both paths now end a run through one helper: the nodes still running become
failed, the run gets one failed line in its history, and updatedAt moves. A run
that already ended keeps its own status and history. A node the graph no longer
has is left as it is rather than thrown, and reap() ends runs one by one so one
bad record cannot leave the rest running.

Author: Birfy

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

@@ -0,0 +1 @@
+A run ended at load, or by a step the host refused, 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` and records one `failed` line in the run's history ([#1424](https://github.com/ThinkFlowLab/FlyBot/issues/1424)).

docs/subsystems/workflow.md

@@ -49,6 +49,8 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher,
 3. Progress lines stay scripted, and a run does not create a schedule. `scheduleLabel` is display text. When the host can create Bots, starting a run (`assign`, `run_workflow`, `ingest`, or a demo that starts or restarts) 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 seeded fixture run.
 
    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).
+
+   Ending a run settles its node map with it: every node still marked `running` becomes `failed`, and the run gets one `failed` line in its history. `publicRun().running` is derived from `nodeStatus`, so a run that says `failed` while a node still says `running` is a ghost the strip paints as "Running now". `settle()`'s dispatch catch ends the run the same way, through the same helper (#1424).
 4. Auto-assign considers an `open` card only. First hit wins: a `workflow:` token, then `workflowId` / `workflow`, then an assignee that names a workflow, then a `bot:` token, then an assignee that names a local Bot, then the longest trigger contained in the title, note or criteria. A miss returns no route. A second ingest of the same task and target is a no-op, and a live run for that task id is not restarted.
 5. The browser draws text only. Names and messages are React text children. The unit renderer asserts a hostile name is not markup.
 6. `enabled: false` registers no endpoint, tool or listener and writes nothing.
@@ -57,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, templates, `define` / `runNamed`, the matcher, ephemeral spawn and cleanup, the client canvas, the settings page, 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 is `core/test/flybot.test.js`.
+`plugins/workflow/test/`: graph advance and join, graph edit, assign and persistence, templates, `define` / `runNamed`, the matcher, ephemeral spawn and cleanup, a reaped run's settled node map (#1424), the client canvas, the settings page, 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 is `core/test/flybot.test.js`.

plugins/workflow/src/service.js

@@ -226,6 +226,32 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no
     if (wrote) write()
   }
 
+  // End a run the host gave up on. Both paths that can abandon one — reap() at load
+  // and settle()'s catch — come through here, so run.status and run.nodeStatus
+  // cannot drift apart and a client stops painting "Running now" (#1424).
+  function endRun(item, run) {
+    // A run that already ended keeps its own ending (#1388); only its records go.
+    if (run.status === 'done' || run.status === 'failed') {
+      delete run.ephemeral
+      delete run.handed
+      return
+    }
+    for (const nodeId of item ? orderedIds(item, run.nodeStatus, 'running') : []) {
+      // The graph can be edited under a live run: a node it no longer has, or one
+      // that moved on already, is left as it is rather than thrown (#1424).
+      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()
+    run.history = [...(run.history ?? []), { task: run.task, status: 'failed', at: now() }].slice(-20)
+    delete run.ephemeral
+    delete run.handed
+  }
+
   async function settle(workflowId) {
     const item = workflow(workflowId)
     const run = item ? state.runs[workflowId] : undefined
@@ -234,17 +260,13 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no
       await handOff(item, run)
     } catch (error) {
       await releaseAll(Object.values(run.ephemeral ?? {}), release, warn)
-      delete run.ephemeral
-      delete run.handed
-      run.status = 'failed'
-      run.updatedAt = now()
+      endRun(item, run)
       write()
       return refuse(error)
     }
     if (run.status === 'done' || run.status === 'failed') {
       await releaseAll(Object.values(run.ephemeral ?? {}), release, warn)
-      delete run.ephemeral
-      delete run.handed
+      endRun(item, run)
       write()
     }
     return present()
@@ -551,13 +573,16 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no
         // 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).
-        const abandoned = run.status === 'running'
-        stale.push(...Object.values(run.ephemeral))
-        delete run.ephemeral
-        delete run.handed
-        if (abandoned) {
-          run.status = 'failed'
-          run.updatedAt = now()
+        try {
+          stale.push(...Object.values(run.ephemeral))
+          if (run.status === 'running') endRun(workflow(run.workflowId), run)
+          else {
+            delete run.ephemeral
+            delete run.handed
+          }
+        } catch (error) {
+          // One run that cannot be ended must not leave the others running.
+          warn(`flybot-workflow: could not end ${run.id}: ${error?.message ?? error}`)
         }
       }
       if (stale.length === 0) return Promise.resolve()

plugins/workflow/test/ephemeral.test.js

@@ -388,3 +388,163 @@ 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('a load settles the node map: no node stays running and the run records one failure (#1424)', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-reap-1424-'))
+  const released = []
+  let clock = 1_700_000_000_000
+  try {
+    const before = persistent(dir, released, () => clock)
+    const made = before.define({
+      name: 'Lane',
+      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 === 'Lane')
+    const started = await before.runNamed({ workflow: 'Lane', brief: 'Ship the note' })
+    assert.equal(started.ok, true)
+    const saved = readStore(dir).runs[made.id]
+    assert.equal(saved.nodeStatus.plan, 'running')
+
+    clock += 60_000
+    const after = persistent(dir, released, () => clock)
+    await after.reap()
+    const run = after.state().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.equal(run.doneCount, 0)
+    assert.equal(run.history.length, 1)
+    assert.equal(run.history[0].status, 'failed')
+    assert.equal(run.history[0].at, clock)
+    assert.equal(run.history[0].task.id, saved.task.id)
+
+    // The store, not just the view, carries the settled map.
+    const settled = readStore(dir).runs[made.id]
+    assert.equal(settled.nodeStatus.plan, 'failed')
+    assert.equal(settled.history.length, 1)
+
+    // A second load has nothing to release and appends no second line.
+    released.length = 0
+    await persistent(dir, released, () => clock).reap()
+    assert.deepEqual(released, [])
+    assert.equal(readStore(dir).runs[made.id].history.length, 1)
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('a run that already ended keeps its ending and its history when reap drops its records (#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)
+    store.runs[made.id] = {
+      id: 'run-done',
+      workflowId: made.id,
+      task: null,
+      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' },
+    }
+    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.equal(run.history.length, 1)
+    assert.equal(run.history[0].status, 'done')
+    assert.equal(run.ephemeral, undefined)
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('a run whose node map names a node the graph no longer has settles the rest and does not throw (#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)
+    store.runs[made.id] = {
+      id: 'run-ghost',
+      workflowId: made.id,
+      task: null,
+      status: 'running',
+      nodeStatus: { plan: 'running', gone: 'running' },
+      steps: 0,
+      stepBudget: 2,
+      failedOnce: {},
+      history: [],
+      startedAt: 4,
+      updatedAt: 5,
+      ephemeral: { plan: 'eph-plan', gone: 'eph-gone' },
+    }
+    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.
+    assert.equal(run.nodeStatus.gone, 'running')
+    assert.deepEqual(run.running, [])
+    assert.equal(run.history.length, 1)
+    assert.equal(run.history[0].status, 'failed')
+    assert.equal(run.ephemeral, undefined)
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('a dispatch that throws ends the run the same way: nodes settled and one failure line (#1424)', async () => {
+  const dispatched = []
+  const { svc, spawned, released } = harness({
+    service: {
+      dispatch: async (step) => { dispatched.push(step); throw new Error('dispatch blew up') },
+    },
+  })
+  const made = lane(svc).workflows.find(item => item.name === 'Lane')
+  const result = await svc.runNamed({ workflow: 'Lane', brief: 'Ship the note' })
+  assert.equal(result.ok, false)
+  assert.equal(dispatched.length, 1)
+  const run = svc.state().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.equal(run.history.length, 1)
+  assert.equal(run.history[0].status, 'failed')
+  assert.deepEqual(released.sort(), spawned.map(entry => entry.id).sort())
+})