fix(workflow): refuse a task-less reset before it clears the lane (#1531)

changelog.d/1531.fixed.md

@@ -0,0 +1 @@
+A `workflow/reset` with no task to keep is now refused before it clears the lane, so a workflow that is waiting with queued tasks and no live run keeps its queue instead of losing every queued task to a refused reset, and a run whose Bot instances were never given a task no longer disappears with them ([#1531](https://github.com/ThinkFlowLab/FlyBot/issues/1531)).

plugins/workflow/src/service.js

@@ -167,13 +167,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)
     }
   }
@@ -204,7 +204,7 @@
     // advance(); without this the step could sit unanswered forever (#1530).
     if (run?.live !== true) return
     const running = new Set(orderedIds(item, run.nodeStatus, 'running'))
     for (const [key, entry] of [...timers]) {
       if (entry.runId !== run.id) continue
       if (run.status !== 'running' || run.paused === true || !running.has(entry.nodeId)) clearTimerAt(key)
     }
@@ -408,7 +408,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 })
@@ -421,7 +421,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 }
@@ -582,7 +582,7 @@
         break
       }
       handOffTries.delete(`${current.id}:${nodeId}`)
       current = { ...current, handed: { ...current.handed, [nodeId]: true }, inputs: { ...(current.inputs ?? {}), [nodeId]: body.inputs } }
       storeRun(item.id, current)
     }
     syncTimers(item, current)
@@ -1240,12 +1240,15 @@
         const item = workflow(workflowId)
         if (!item) return { ok: false, error: 'No such workflow' }
         const lane = laneOf(workflowId)
-        const ids = lane.active.flatMap(run => Object.values(run.ephemeral ?? {}))
         const live = lane.active[0] ?? (lane.last?.live === true ? lane.last : null)
+        // A reset that cannot keep a task has to be refused before it touches
+        // the lane: clearing first drops the queue, and a live run's Bot ids, on
+        // a path that never writes or releases them.
+        if (!live?.task) return { ok: false, error: 'This workflow has no task' }
+        const ids = lane.active.flatMap(run => Object.values(run.ephemeral ?? {}))
         for (const run of lane.active) clearRunTimers(run.id)
         lane.active = []
         lane.queue = []
-        if (!live?.task) return { ok: false, error: 'This workflow has no task' }
         const kept = { ...blankRun(item, live.task, now(), lane.history), id: live.id, startedAt: live.startedAt }
         lane.last = kept
         push(item.id, linesFor(item, live.task, { type: 'reset' }), now())
@@ -1417,12 +1420,12 @@
       return releaseAll(stale, release, warn).then(() => { write() })
     },
     dispose() {
       for (const key of [...timers.keys()]) clearTimerAt(key)
       const ids = []
       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') {
             endRun(workflow(workflowId), run, { code: 'host-ended' })
             archiveCutoffRun(workflowId, run)

plugins/workflow/test/reset.test.js

@@ -0,0 +1,162 @@
+// #1531: a `workflow/reset` that is going to be refused must be refused
+// *before* it touches the lane. Clearing first and validating after throws away
+// a queue, and drops a live run (and its Bot ids) on a path that never writes
+// or releases them, so the next ordinary write flushes the empty lane to disk.
+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 { freshStatus, normalizeGraph } from '../src/graph.js'
+import { createWorkflowService } from '../src/service.js'
+import { DELIVERY_ID, deliveryWorkflow } from '../src/template.js'
+
+const roster = [
+  { id: 'main', name: 'Chief', role: 'lead', isMain: true, remote: false },
+]
+const CLOCK = 1_700_000_000_000
+const FILE = 'workflows.json'
+
+function readStore(dir) {
+  return JSON.parse(readFileSync(join(dir, FILE), 'utf8'))
+}
+
+function lane(id, value) {
+  return { active: [], queue: [], last: null, history: [], ...value }
+}
+
+function liveRun(overrides = {}) {
+  return {
+    id: 'run-1',
+    workflowId: DELIVERY_ID,
+    task: null,
+    status: 'running',
+    live: true,
+    nodeStatus: { ...freshStatus(normalizeGraph(deliveryWorkflow())), planner: 'running' },
+    steps: 0,
+    stepBudget: 10,
+    failedOnce: {},
+    produced: {},
+    history: [],
+    startedAt: CLOCK,
+    updatedAt: CLOCK,
+    ephemeral: { plan: 'eph-plan' },
+    ...overrides,
+  }
+}
+
+function open(dir) {
+  const released = []
+  const spawned = []
+  let seq = 0
+  const svc = createWorkflowService({
+    dir,
+    persist: true,
+    now: () => CLOCK,
+    bots: () => roster,
+    spawn: async () => {
+      seq += 1
+      const id = `eph-new-${seq}`
+      spawned.push(id)
+      return id
+    },
+    release: async (id) => { released.push(id) },
+    dispatch: async () => ({ ok: true }),
+  })
+  return { svc, released, spawned }
+}
+
+// Write a lane fixture through the real store: boot the service, let a plain
+// write create the file, then put the lane under test in it and reload.
+function seed(dir, value) {
+  const first = open(dir)
+  first.svc.setConcurrency({ workflowId: DELIVERY_ID, maxConcurrent: 2 })
+  const store = readStore(dir)
+  store.lanes[DELIVERY_ID] = lane(DELIVERY_ID, value)
+  writeFileSync(join(dir, FILE), JSON.stringify(store))
+}
+
+function itemOf(state, id = DELIVERY_ID) {
+  return state.workflows.find(workflow => workflow.id === id)
+}
+
+test('a refused reset leaves the queue untouched', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-reset-'))
+  try {
+    seed(dir, { queue: [{ id: 'task-1', title: 'A queued task that must not vanish' }] })
+    const { svc, released } = open(dir)
+
+    const refused = await svc.reset({ workflowId: DELIVERY_ID })
+    assert.equal(refused.ok, false)
+    assert.equal(refused.error, 'This workflow has no task')
+
+    const after = itemOf(svc.state())
+    assert.equal(after.queueDepth, 1)
+    assert.equal(after.queue.length, 1)
+    assert.equal(after.queue[0].id, 'task-1')
+    assert.deepEqual(released, [])
+
+    // The next ordinary write must not flush a lane the reset was refused for.
+    const renamed = await svc.rename({ workflowId: DELIVERY_ID, name: 'Renamed after the refusal' })
+    assert.equal(renamed.ok, true)
+    assert.equal(itemOf(svc.state()).queue.length, 1)
+    assert.equal(readStore(dir).lanes[DELIVERY_ID].queue.length, 1)
+
+    // And a reload of the store still has the queued task.
+    assert.equal(itemOf(open(dir).svc.state()).queue.length, 1)
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('a refused reset keeps a run that has no task', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-reset-'))
+  try {
+    const run = liveRun()
+    seed(dir, { active: [run] })
+    const { svc, released } = open(dir)
+    const before = itemOf(svc.state())
+
+    const refused = await svc.reset({ workflowId: DELIVERY_ID })
+    assert.equal(refused.ok, false)
+    assert.equal(refused.error, 'This workflow has no task')
+
+    const after = itemOf(svc.state())
+    assert.equal(after.activeCount, 1)
+    assert.equal(after.runs.length, 1)
+    assert.equal(after.runs[0].id, run.id)
+    assert.deepEqual(after.runs[0].nodeStatus, before.runs[0].nodeStatus)
+    assert.equal(after.run?.status, 'running')
+    assert.deepEqual(released, [])
+    assert.deepEqual(readStore(dir).lanes[DELIVERY_ID].active.map(item => item.id), [run.id])
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('a reset with a task still clears the lane and releases its Bots', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'wf-reset-'))
+  try {
+    const run = liveRun({
+      task: { id: '#1472', title: 'A task that exists' },
+      ephemeral: { plan: 'eph-plan', code: 'eph-code' },
+    })
+    seed(dir, { active: [run], queue: [{ id: 'task-2', title: 'Queued behind' }] })
+    const { svc, released } = open(dir)
+
+    const reset = await svc.reset({ workflowId: DELIVERY_ID })
+    assert.equal(reset.ok, true)
+
+    const after = itemOf(svc.state())
+    assert.equal(after.activeCount, 0)
+    assert.equal(after.queueDepth, 0)
+    assert.deepEqual(after.runs, [])
+    assert.deepEqual([...released].sort(), ['eph-code', 'eph-plan'])
+    const stored = readStore(dir).lanes[DELIVERY_ID]
+    assert.equal(stored.active.length, 0)
+    assert.equal(stored.queue.length, 0)
+    assert.equal(stored.last?.task?.id, '#1472')
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})