fix(workflow): one writer per workflow so a concurrent assign cannot orphan a set (#1390)

changelog.d/1390-workflow-assign-race.fixed.md

@@ -0,0 +1 @@
+Two `assign()` calls for one Workflow started together each spawned a full set of run Bots, and one set was left referenced by no run and removed by nobody. The writers for a Workflow are now chained, so the second call releases the set it replaced; the same chain covers a demo restart racing an assign ([#1390](https://github.com/ThinkFlowLab/FlyBot/issues/1390)).

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).
+
+   The writers for one Workflow are chained: a second `assign`, or a demo restart, waits for the one in flight, reads the ephemeral set it left, and releases it. Two calls started together therefore cannot both spawn a set and leave one referenced by no run (#1390). The chain is per Workflow, so a slow spawn on one does not hold up another.
 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, two assigns that overlap (#1390), 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

@@ -153,6 +153,25 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no
     return { ok: false, error: error?.message ?? String(error) }
   }
 
+  // One writer at a time per workflow. A spawn hook makes assign() and the demo
+  // restart await before they write, and two bodies for one workflow would both
+  // read the previous ephemeral set, both spawn a full set, and the second write
+  // would replace the run — leaving the first set referenced by no run and
+  // released by nobody. Chaining the bodies keeps the replaced set readable.
+  const chains = new Map()
+  function queue(workflowId, task) {
+    const previous = chains.get(workflowId) ?? Promise.resolve()
+    const result = previous.then(task, task)
+    // The chained entry never rejects, so a failed call cannot wedge the
+    // workflow for the ones behind it.
+    const tail = result.then(() => {}, () => {})
+    chains.set(workflowId, tail)
+    void tail.then(() => {
+      if (chains.get(workflowId) === tail) chains.delete(workflowId)
+    })
+    return result
+  }
+
   function applyOp(item, run, op) {
     let result
     if (op.type === 'start') result = { ...startFrontier(item, run.nodeStatus), joined: false, layerSize: 0 }
@@ -337,11 +356,13 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no
         state.runs[row.id] = run
       })
       if (typeof spawn !== 'function') return go(undefined)
-      return (async () => {
+      return queue(workflowId, async () => {
+        const live = workflow(workflowId)
+        if (!live) return { ok: false, error: 'No such workflow' }
         const previousIds = Object.values(state.runs[workflowId]?.ephemeral ?? {})
         let ephemeral
         try {
-          ephemeral = await spawnAll(item.nodes, bots() ?? [], spawn)
+          ephemeral = await spawnAll(live.nodes, bots() ?? [], spawn)
         } catch (error) {
           await releaseAll(error.spawned ?? [], release, warn)
           return refuse(error)
@@ -353,7 +374,7 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no
         }
         await releaseAll(previousIds, release, warn)
         return settle(workflowId)
-      })()
+      })
     },
     advance(body = {}) {
       const workflowId = String(body.workflowId ?? '')
@@ -393,20 +414,22 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no
         push(item.id, stepped.lines, now())
       })
       if (typeof spawn !== 'function') return step(undefined)
-      return (async () => {
-        if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' }
-        const item = workflow(workflowId)
-        if (!item) return { ok: false, error: 'No such workflow' }
-        const run = state.runs[item.id]
+      if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' }
+      const item = workflow(workflowId)
+      if (!item) return { ok: false, error: 'No such workflow' }
+      return queue(workflowId, async () => {
+        const live = workflow(workflowId)
+        if (!live) return { ok: false, error: 'No such workflow' }
+        const run = state.runs[live.id]
         if (!run) return { ok: false, error: 'This workflow has no task' }
         if (run.status === 'failed') return { ok: false, error: 'This run is already finished' }
-        const op = nextDemoOp(item, run.nodeStatus, { failedOnce: run.failedOnce })
+        const op = nextDemoOp(live, run.nodeStatus, { failedOnce: run.failedOnce })
         let ephemeral
         let previousIds = []
         if (op.type === 'restart' || op.type === 'start') {
           previousIds = Object.values(run.ephemeral ?? {})
           try {
-            ephemeral = await spawnAll(item.nodes, bots() ?? [], spawn)
+            ephemeral = await spawnAll(live.nodes, bots() ?? [], spawn)
           } catch (error) {
             await releaseAll(error.spawned ?? [], release, warn)
             return refuse(error)
@@ -419,7 +442,7 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no
         }
         await releaseAll(previousIds, release, warn)
         return settle(workflowId)
-      })()
+      })
     },
     reset(body = {}) {
       const workflowId = String(body.workflowId ?? '')

plugins/workflow/test/ephemeral.test.js

@@ -388,3 +388,131 @@ test('the plugin load itself reaps a run the restart left running (#1388)', asyn
     rmSync(dir, { recursive: true, force: true })
   }
 })
+
+// #1390: with a spawn hook the assign body awaits before it writes, so two calls
+// for one workflow used to read the same previous set, both spawn a full set, and
+// the second write replaced the run — the first set, referenced by no run, was
+// released by nobody. `spawn` here yields a macrotask, so bodies started together
+// interleave the way they do on a host whose spawn is not instant.
+function racer(control = {}) {
+  const spawned = []
+  const released = []
+  let seq = 0
+  const svc = createWorkflowService({
+    persist: false,
+    now: () => 1_700_000_000_000,
+    bots: () => roster,
+    spawn: async (spec) => {
+      const id = `race-${++seq}`
+      spawned.push({ id, spec })
+      if (control.fail?.(spec)) throw new Error('spawn failed')
+      if (control.gate?.[spec.nodeId]) await control.gate[spec.nodeId]
+      else await new Promise((resolve) => setImmediate(resolve))
+      return id
+    },
+    release: async (id) => { released.push(id) },
+    dispatch: async () => ({ ok: true }),
+    warn: () => {},
+  })
+  return { svc, spawned, released }
+}
+
+function racePair(svc, name) {
+  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('two assigns started together leave one set of run Bots and release the other (#1390)', async () => {
+  const { svc, spawned, released } = racer()
+  const made = racePair(svc, 'Race')
+  const first = svc.assign({ workflowId: made.id, taskId: 'T1', title: 'First task' })
+  const second = svc.assign({ workflowId: made.id, taskId: 'T2', title: 'Second task' })
+  const [one, two] = await Promise.all([first, second])
+  assert.equal(one.ok, true)
+  assert.equal(two.ok, true)
+
+  const run = svc.state().workflows.find(item => item.id === made.id).run
+  assert.equal(run.task.id, 'T2')
+  const survivors = new Set(Object.values(run.ephemeral))
+  assert.equal(survivors.size, 2)
+  assert.deepEqual(Object.keys(run.ephemeral).sort(), ['code', 'plan'])
+  // Every Bot either serves the surviving run or was released, and none twice.
+  const expected = spawned.map(entry => entry.id).filter(id => !survivors.has(id))
+  assert.deepEqual([...released].sort(), [...expected].sort())
+  assert.equal(new Set(released).size, released.length)
+  // The replaced task is recorded once, in the shape the history already uses.
+  const replaced = run.history.filter(entry => entry.task?.id === 'T1')
+  assert.equal(replaced.length, 1)
+  assert.equal(replaced[0].status, 'running')
+})
+
+test('a demo restart racing an assign leaves no orphaned set either (#1390)', async () => {
+  const { svc, spawned, released } = racer()
+  const made = racePair(svc, 'Demo')
+  await svc.assign({ workflowId: made.id, taskId: 'T1', title: 'First task' })
+  await svc.demoStep({ workflowId: made.id })
+  const finished = await svc.demoStep({ workflowId: made.id })
+  const done = finished.workflows.find(item => item.id === made.id).run
+  assert.equal(done.status, 'done')
+  assert.equal(done.ephemeral, undefined)
+
+  const restart = svc.demoStep({ workflowId: made.id })
+  const assigned = svc.assign({ workflowId: made.id, taskId: 'T2', title: 'Second task' })
+  const [re, as] = await Promise.all([restart, assigned])
+  assert.equal(re.ok, true)
+  assert.equal(as.ok, true)
+
+  const run = svc.state().workflows.find(item => item.id === made.id).run
+  assert.equal(run.task.id, 'T2')
+  const survivors = new Set(Object.values(run.ephemeral))
+  const expected = spawned.map(entry => entry.id).filter(id => !survivors.has(id))
+  assert.deepEqual([...released].sort(), [...expected].sort())
+  assert.equal(new Set(released).size, released.length)
+})
+
+test('a slow spawn on one workflow does not hold up an assign on another (#1390)', async () => {
+  let openSlow
+  const slowGate = new Promise((resolve) => { openSlow = resolve })
+  const { svc } = racer({ gate: { slow: slowGate } })
+  const slow = svc.define({ name: 'Slow', nodes: [{ id: 'slow', name: 'Coder', role: 'code', prompt: 'Take your time.' }], edges: [] })
+    .workflows.find(item => item.name === 'Slow')
+  const quick = svc.define({ name: 'Quick', nodes: [{ id: 'quick', name: 'Planner', role: 'plan', prompt: 'Just start.' }], edges: [] })
+    .workflows.find(item => item.name === 'Quick')
+
+  const waiting = svc.assign({ workflowId: slow.id, taskId: 'S1', title: 'Slow task' })
+  const started = svc.assign({ workflowId: quick.id, taskId: 'Q1', title: 'Quick task' })
+  let timer
+  const stuck = new Promise((resolve) => { timer = setTimeout(() => resolve('stuck'), 2000) })
+  const outcome = await Promise.race([started.then(() => 'started'), stuck])
+  clearTimeout(timer)
+  openSlow()
+  assert.equal(outcome, 'started')
+  const [slowDone, quickDone] = await Promise.all([waiting, started])
+  assert.equal(slowDone.ok, true)
+  assert.equal(quickDone.ok, true)
+  assert.equal(svc.state().workflows.find(item => item.id === quick.id).run.task.id, 'Q1')
+})
+
+test('a spawn that throws halfway releases what it made and leaves the workflow usable (#1390)', async () => {
+  let failOn = 'code'
+  const { svc, spawned, released } = racer({ fail: (spec) => spec.nodeId === failOn })
+  const made = racePair(svc, 'Broken')
+  const failed = await svc.assign({ workflowId: made.id, taskId: 'T1', title: 'Broken task' })
+  assert.equal(failed.ok, false)
+  assert.match(failed.error, /spawn failed/)
+  assert.deepEqual(released, [spawned[0].id])
+  assert.equal(svc.state().workflows.find(item => item.id === made.id).run, null)
+
+  failOn = ''
+  const again = await svc.assign({ workflowId: made.id, taskId: 'T2', title: 'Second try' })
+  assert.equal(again.ok, true)
+  const run = again.workflows.find(item => item.id === made.id).run
+  assert.equal(run.task.id, 'T2')
+  assert.equal(Object.keys(run.ephemeral).length, 2)
+})