fix(workflow): reap a run a restart left running (#1388)

changelog.d/1388-workflow-reap-running.fixed.md

@@ -0,0 +1 @@
+A workflow run that a restart interrupted was never cleaned up: the run stayed `running` and its run Bots stayed hidden with no way to remove them. Loading the plugin now releases those Bots and records the run as `failed` ([#1388](https://github.com/ThinkFlowLab/FlyBot/issues/1388)).

docs/subsystems/workflow.md

@@ -47,6 +47,8 @@ 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. 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).
 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.

plugins/workflow/src/service.js

@@ -547,10 +547,18 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no
       const stale = []
       for (const run of Object.values(state.runs)) {
         if (!run?.ephemeral) continue
-        if (run.status === 'running') 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).
+        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()
+        }
       }
       if (stale.length === 0) return Promise.resolve()
       return releaseAll(stale, release, warn).then(() => { write() })

plugins/workflow/test/ephemeral.test.js

@@ -2,6 +2,9 @@
 // The instance is hidden (the sidebar predicate rejects it), does the step,
 // and is removed when the run finishes, fails for good, or is reset.
 import assert from 'node:assert/strict'
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
 import { test } from 'node:test'
 import { isShownRemoteChat } from '../../../core/src/shared/remote-label.js'
 import { apply } from '../index.js'
@@ -223,3 +226,165 @@ test('#1389: a step the host refuses is not marked as handed, and a later settle
   assert.equal(codeCalls[0].botId, codeCalls[1].botId)
   assert.match(warnings.join('\n'), /no-sender/)
 })
+
+// #1388: reap is the only path that cleans up a run whose host went away, and the
+// plugin calls it exactly once, at load. A run still marked `running` at that
+// moment is what a restart leaves behind: nothing resumes it, so its ephemeral
+// Bots are abandoned. Loading must release them and end the run as failed, or
+// they stay hidden in the roster with no way to remove them.
+function persistent(dir, released, clock) {
+  return createWorkflowService({
+    dir,
+    persist: true,
+    now: () => clock(),
+    bots: () => roster,
+    spawn: async (spec) => `eph-${spec.nodeId}`,
+    release: async (id) => { released.push(id) },
+    dispatch: async () => ({ ok: true }),
+  })
+}
+
+function readStore(dir) {
+  return JSON.parse(readFileSync(join(dir, 'workflows.json'), 'utf8'))
+}
+
+test('a load reaps the run a restart left running: its Bots go and the run ends failed (#1388)', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-reap-'))
+  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')
+    await before.runNamed({ workflow: 'Lane', brief: 'Ship the note' })
+    const saved = readStore(dir).runs[made.id]
+    assert.equal(saved.status, 'running')
+    assert.deepEqual(Object.values(saved.ephemeral).sort(), ['eph-code', 'eph-plan'])
+    assert.equal(released.length, 0)
+
+    // The host restarted here. This load is the only chance to clean up.
+    clock += 60_000
+    const after = persistent(dir, released, () => clock)
+    await after.reap()
+    assert.deepEqual(released.sort(), ['eph-code', 'eph-plan'])
+    const run = after.state().workflows.find(item => item.id === made.id).run
+    assert.equal(run.status, 'failed')
+    assert.equal(run.ephemeral, undefined)
+    assert.equal(run.handed, undefined)
+    assert.equal(run.updatedAt, clock)
+    const settled = readStore(dir).runs[made.id]
+    assert.equal(settled.status, 'failed')
+    assert.equal(settled.ephemeral, undefined)
+    assert.equal('handed' in settled, false)
+    assert.equal(settled.updatedAt, clock)
+
+    // A second load finds nothing left to release.
+    released.length = 0
+    await persistent(dir, released, () => clock).reap()
+    assert.deepEqual(released, [])
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('a finished run stays done, and a run with no ephemeral Bots is not touched (#1388)', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-reap-done-'))
+  const released = []
+  let clock = 1_700_000_000_000
+  try {
+    const svc = persistent(dir, released, () => clock)
+    const quick = svc.define({
+      name: 'Quick',
+      nodes: [{ id: 'only', name: 'Planner', role: 'plan', prompt: 'Say it once.' }],
+      edges: [],
+    }).workflows.find(item => item.name === 'Quick')
+    await svc.runNamed({ workflow: 'Quick', brief: 'One step' })
+    await svc.advance({ workflowId: quick.id, nodeId: 'only', outcome: 'done' })
+    released.length = 0
+    const finished = readStore(dir).runs[quick.id]
+    assert.equal(finished.status, 'done')
+    assert.equal(finished.ephemeral, undefined)
+    const finishedAt = finished.updatedAt
+
+    // A run record whose ephemeral map is already gone (the shape the restart
+    // leaves when it lost the map) is outside this fix: reap must not rewrite it.
+    const idle = svc.define({ name: 'Idle', nodes: [{ id: 'one', name: 'Planner', role: 'plan', prompt: 'Wait.' }], edges: [] })
+      .workflows.find(item => item.name === 'Idle')
+    const store = readStore(dir)
+    store.runs[idle.id] = {
+      id: 'run-bare',
+      workflowId: idle.id,
+      task: null,
+      status: 'running',
+      nodeStatus: {},
+      steps: 0,
+      stepBudget: 1,
+      failedOnce: {},
+      history: [],
+      startedAt: 5,
+      updatedAt: 6,
+    }
+    const written = JSON.stringify(store)
+    writeFileSync(join(dir, 'workflows.json'), written)
+
+    clock += 60_000
+    const after = persistent(dir, released, () => clock)
+    await after.reap()
+    assert.deepEqual(released, [])
+    const kept = after.state().workflows.find(item => item.id === quick.id).run
+    assert.equal(kept.status, 'done')
+    assert.equal(kept.updatedAt, finishedAt)
+    const bare = after.state().workflows.find(item => item.id === idle.id).run
+    assert.equal(bare.status, 'running')
+    assert.equal(bare.updatedAt, 6)
+    // Nothing to release means nothing is written: the file it loaded is intact.
+    assert.equal(readFileSync(join(dir, 'workflows.json'), 'utf8'), written)
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('the plugin load itself reaps a run the restart left running (#1388)', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-reap-apply-'))
+  const released = []
+  const 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.' }],
+      edges: [],
+    }).workflows.find(item => item.name === 'Lane')
+    await before.runNamed({ workflow: 'Lane', brief: 'Ship the note' })
+    assert.equal(readStore(dir).runs[made.id].status, 'running')
+
+    // Loading the plugin is where the abandoned run is recognised.
+    const fb = {
+      bot: {
+        list: () => roster,
+        async create() { return 'bot-1' },
+        async remove(id) { released.push(id); return true },
+        async send() { return { action: 'allow' } },
+      },
+      readOnly: () => false,
+      api: { handle() { return () => {} } },
+      tools: { register() { return () => {} } },
+      warn() {},
+    }
+    const loaded = apply({ flybot: { scope: () => fb }, effect(fn) { fn() } }, { dir })
+    await new Promise((resolve) => setTimeout(resolve, 20))
+    assert.deepEqual(released, ['eph-plan'])
+    const run = loaded.service.state().workflows.find(item => item.id === made.id).run
+    assert.equal(run.status, 'failed')
+    assert.equal(run.ephemeral, undefined)
+    loaded.dispose()
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})