fix(workflow): keep a pause, a resume and a parked card when a dispatch lands

plugins/workflow/src/service.js

@@ -313,13 +313,13 @@
   }
 
   function clearRunTimers(runId) {
     for (const [key, entry] of [...timers]) {
       if (entry.runId === runId) clearTimerAt(key)
     }
     for (const key of [...warned]) {
       if (key.endsWith(`:${runId}`)) warned.delete(key)
     }
     for (const key of [...handOffTries.keys()]) {
       if (key.startsWith(`${runId}:`)) handOffTries.delete(key)
     }
   }
@@ -356,7 +356,7 @@
     // keep its deadline: without it the run holds its lane slot with nothing left
     // to free it until the owner presses Stop again (P-560).
     const stopRetry = run.status === 'running' && run.paused === true && run.releasePending === true
     for (const [key, entry] of [...timers]) {
       if (entry.runId !== run.id) continue
       if (run.status !== 'running' || (run.paused === true && !stopRetry) || !running.has(entry.nodeId)) clearTimerAt(key)
     }
@@ -661,7 +661,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 })
@@ -674,7 +674,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 }
@@ -1010,8 +1010,10 @@
     if (!best) return null
     const task = laneOf(best.item.id).queue.shift()
     // #1869: the row is rebuilt from the parked fields, so the brief has to be
-    // named here too or it is lost exactly when the run starts.
-    return { item: best.item, task: carriedTask(task) }
+    // named here too or it is lost exactly when the run starts. The parked row
+    // itself travels back too, so a start the host refuses puts the card where
+    // it was instead of taking it out of the queue for good.
+    return { item: best.item, task: carriedTask(task), row: task }
   }
 
   function noteUnstarted(item, error) {
@@ -1029,6 +1031,10 @@
       if (!next) return
       const begun = beginRun(next.item, next.task)
       if (begun.ok === false) {
+        // The card never started, so it is still the owner's parked work. Taking it out
+        // of the queue here loses it: no run holds it, no claim is retried for it, and
+        // the next ingest answers "already" for a card in nobody's hands.
+        laneOf(next.item.id).queue.unshift(next.row)
         noteUnstarted(next.item, begun.error)
         return
       }
@@ -1047,6 +1053,7 @@
       if (!next) return
       const started = await startLive(next.item, next.task)
       if (started?.ok === false) {
+        laneOf(next.item.id).queue.unshift(next.row)
         noteUnstarted(next.item, started.error)
         return
       }
@@ -1801,8 +1808,11 @@
       let run = runId ? lane.active.find(itemRun => itemRun.id === runId) : lane.active.find(itemRun => itemRun.live === true && itemRun.status === 'running')
       if (!run || run.live !== true) return { ok: false, error: 'No live run to pause' }
       if (run.paused === true) return present()
-      run = { ...run, paused: true, updatedAt: now() }
-      storeRun(workflowId, run)
+      // Set on the record the lane holds, not on a copy: a hand-off waiting on its dispatch
+      // still holds that same object and stores it back when the dispatch lands, so a copy
+      // here is replaced by the hand-off's copy and the pause is silently lost.
+      run.paused = true
+      run.updatedAt = now()
       syncTimers(item, run)
       push(workflowId, [{
         kind: 'system',
@@ -1827,8 +1837,8 @@
       // silent skip. Stop again is the retry (#1476).
       if (run.releasePending === true) return { ok: false, error: 'This run is still waiting to release a Bot; stop it again to retry' }
       if (run.paused !== true) return present()
-      run = { ...run, paused: false, updatedAt: now() }
-      storeRun(workflowId, run)
+      run.paused = false
+      run.updatedAt = now()
       syncTimers(item, run)
       push(workflowId, [{
         kind: 'system',
@@ -2475,7 +2485,12 @@
       warn(`flybot-workflow: ${problem}; starting from the fixture`)
       return { state: null, error: problem, newer: version }
     }
-    if (version !== STORE_VERSION || !Array.isArray(parsed.workflows) || typeof parsed.runs !== 'object') {
+    // A store this build can parse but not read is the same failure as one that does not parse:
+    // the read has to say so instead of throwing out of present() on every render. A workflow
+    // entry that is not an object, or carries no nodes, is not one this build wrote.
+    const entriesReadable = Array.isArray(parsed.workflows)
+      && parsed.workflows.every(item => item !== null && typeof item === 'object' && Array.isArray(item.nodes) && item.nodes.length > 0)
+    if (version !== STORE_VERSION || !entriesReadable || parsed.runs === null || typeof parsed.runs !== 'object') {
       const problem = `the workflow store ${file} is not a workflow file`
       warn(`flybot-workflow: ${problem}; starting from the fixture`)
       return { state: null, error: problem }

plugins/workflow/test/pause-during-dispatch.test.js

@@ -0,0 +1,31 @@
+// A Pause that arrives while a step's dispatch is in flight must not be thrown away.
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { createWorkflowService } from '../src/service.js'
+import { DELIVERY_ID } from '../src/template.js'
+
+test('pause during an in-flight step dispatch is not lost', async () => {
+  let release
+  const gate = new Promise(resolve => { release = resolve })
+  let seq = 0
+  const svc = createWorkflowService({
+    persist: false,
+    now: () => 1_700_000_000_000,
+    bots: () => [{ id: 'main', name: 'Chief', isMain: true, remote: false }],
+    spawn: async () => { seq += 1; return `eph-${seq}` },
+    release: () => {},
+    dispatch: async () => { await gate; return { ok: true } },
+  })
+  const row = () => svc.state().workflows.find(item => item.id === DELIVERY_ID)
+
+  const assigning = svc.assign({ workflowId: DELIVERY_ID, taskId: '#9001', title: 'Pause while dispatching' })
+  await new Promise(resolve => setTimeout(resolve, 20)) // the dispatch is now blocked in flight
+  svc.pause({ workflowId: DELIVERY_ID })
+  assert.equal(row().run.paused, true)
+  release()
+  await assigning
+  await new Promise(resolve => setTimeout(resolve, 20))
+
+  assert.equal(row().run.paused, true, 'the run came back unpaused after the dispatch landed')
+  assert.doesNotMatch(row().preview, /正在处理/)
+})

plugins/workflow/test/queued-card-not-dropped.test.js

@@ -0,0 +1,48 @@
+// A card that was parked and could not start is still the owner's parked work: it stays
+// in the queue and starts on the next drain, instead of leaving the queue for good.
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { createWorkflowService } from '../src/service.js'
+import { DELIVERY_ID } from '../src/template.js'
+
+function bench() {
+  const state = { armed: false, seed: 0 }
+  const svc = createWorkflowService({
+    persist: false,
+    now: () => 1_700_000_000_000,
+    bots: () => [{ id: 'main', name: 'Chief', isMain: true, remote: false }],
+    spawn: async () => {
+      if (state.armed) throw new Error('the host could not start a Bot (transient)')
+      state.seed += 1
+      return `eph-${state.seed}`
+    },
+    release: () => {},
+    dispatch: async () => ({ ok: true }),
+  })
+  return { svc, state }
+}
+
+const row = svc => svc.state().workflows.find(item => item.id === DELIVERY_ID)
+const queued = svc => row(svc).queue.map(task => task.id)
+const running = svc => row(svc).runs.filter(run => run.status === 'running').map(run => run.task?.id)
+
+test('a parked card whose start fails stays in the queue', async () => {
+  const { svc, state } = bench()
+  svc.setConcurrency({ workflowId: DELIVERY_ID, maxConcurrent: 1 })
+  await svc.ingest({ id: '#A', title: 'First card', status: 'open', workflowId: DELIVERY_ID })
+  await svc.ingest({ id: '#B', title: 'Second card', status: 'open', workflowId: DELIVERY_ID })
+  assert.deepEqual(queued(svc), ['#B'])
+
+  const first = row(svc).runs.find(run => run.status === 'running')
+  state.armed = true // free the slot; the spawn for the parked card fails once
+  await svc.reset({ workflowId: DELIVERY_ID, runId: first.id })
+  state.armed = false
+
+  assert.deepEqual(queued(svc), ['#B'], 'the parked card left the queue without starting')
+  assert.deepEqual(running(svc), [])
+
+  // A later card joins the tail and drains the queue: the parked card starts first.
+  await svc.ingest({ id: '#C', title: 'Third card', status: 'open', workflowId: DELIVERY_ID })
+  assert.deepEqual(running(svc), ['#B'], 'the parked card never got its turn')
+  assert.deepEqual(queued(svc), ['#C'])
+})

plugins/workflow/test/resume-during-dispatch.test.js

@@ -0,0 +1,59 @@
+// A Resume is thrown away the same way a Pause is: the hand-off that was already in
+// flight stores the run record it grabbed before, so the resolve never reaches the lane.
+// The second card is the control — unless a pause survives its own landing, the resume
+// case above it would pass without the fix purely because the pause was lost too.
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { createWorkflowService } from '../src/service.js'
+import { DELIVERY_ID } from '../src/template.js'
+
+const settle = (ms = 30) => new Promise(resolve => setTimeout(resolve, ms))
+
+function bench() {
+  const gates = []
+  let seq = 0
+  const svc = createWorkflowService({
+    persist: false,
+    now: () => 1_700_000_000_000,
+    bots: () => [{ id: 'main', name: 'Chief', isMain: true, remote: false }],
+    spawn: async () => { seq += 1; return `eph-${seq}` },
+    release: () => {},
+    dispatch: async () => { await new Promise(resolve => { gates.push(resolve) }); return { ok: true } },
+  })
+  const row = () => svc.state().workflows.find(item => item.id === DELIVERY_ID)
+  const runOf = taskId => row().runs.find(run => run.task?.id === taskId)
+  return { svc, gates, runOf }
+}
+
+test('resume during an in-flight step dispatch is not lost', async () => {
+  const { svc, gates, runOf } = bench()
+
+  // #9002 — the resume case: a pause and then a resume, both while the dispatch is in flight.
+  const first = svc.assign({ workflowId: DELIVERY_ID, taskId: '#9002', title: 'Resume while dispatching' })
+  await settle()
+  const a = runOf('#9002').id
+  svc.pause({ workflowId: DELIVERY_ID, runId: a })
+  assert.equal(runOf('#9002').paused, true, 'the pause did not take effect')
+  svc.resume({ workflowId: DELIVERY_ID, runId: a })
+  assert.notEqual(runOf('#9002').paused, true, 'the resume did not take effect')
+  gates[0]()
+  await first
+  await settle()
+  assert.notEqual(runOf('#9002').paused, true, 'the resume was thrown away by the dispatch that landed')
+
+  // #9003 — the control: the same landing with no resume has to leave the run paused.
+  const second = svc.assign({ workflowId: DELIVERY_ID, taskId: '#9003', title: 'Pause while dispatching' })
+  await settle()
+  const b = runOf('#9003').id
+  svc.pause({ workflowId: DELIVERY_ID, runId: b })
+  assert.equal(runOf('#9003').paused, true, 'the pause did not take effect')
+  gates[1]()
+  await second
+  await settle()
+  assert.equal(runOf('#9003').paused, true, 'the pause was thrown away by the dispatch that landed')
+
+  for (let round = 0; round < 3; round += 1) {
+    for (const resolve of gates.splice(0)) resolve()
+    await settle(20)
+  }
+})

plugins/workflow/test/store-entry-unreadable.test.js

@@ -0,0 +1,34 @@
+// A store that parses but whose workflow entries are not workflows is the same failure
+// as a store that does not parse: the read answers `store-unreadable` instead of throwing.
+import assert from 'node:assert/strict'
+import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import { createWorkflowService } from '../src/service.js'
+
+const STORE_VERSION = 1
+
+function openWith(entry) {
+  const dir = mkdtempSync(join(tmpdir(), 'store-entry-'))
+  writeFileSync(join(dir, 'workflows.json'), JSON.stringify({ version: STORE_VERSION, workflows: [entry], runs: {} }))
+  const svc = createWorkflowService({ dir, persist: true, now: () => 1, bots: () => [], warn: () => {} })
+  return { svc, dir }
+}
+
+for (const [label, entry] of [
+  ['an entry that is not an object', null],
+  ['an entry with no nodes', { id: 'w1', name: 'Broken' }],
+  ['an entry whose nodes are not a list', { id: 'w1', name: 'Broken', nodes: null, edges: [], layers: [] }],
+]) {
+  test(`a store with ${label} answers store-unreadable`, () => {
+    const { svc, dir } = openWith(entry)
+    try {
+      const state = svc.state()
+      assert.equal(state.ok, false, 'the read answered a list instead of the failure')
+      assert.equal(state.code, 'store-unreadable')
+    } finally {
+      rmSync(dir, { recursive: true, force: true })
+    }
+  })
+}