From d29e25fca6bd907f1239f1300a717b8afeddcf0a Mon Sep 17 00:00:00 2001 From: Day Dreamer <87640074+HugoXK@users.noreply.github.com> Date: Wed, 28 Oct 2026 10:33:07 +0800 Subject: [PATCH] fix(workflow): refuse a task-less reset before it clears the lane `workflow/reset` without a `runId` emptied `lane.active` and `lane.queue` and only then answered "This workflow has no task" when there was nothing to keep. The refusal neither writes nor releases, so the cleared lane stayed in memory and the next ordinary write flushed it to disk: a lane holding a queue and no live run lost every queued task, and a live run without a task (the shape an older unload leaves, #1414) lost its ephemeral Bot ids, which nothing could release afterwards. Check the task before the two clears. The success path is unchanged. Docs-not-needed: only the order of a refusal path changes; no invariant docs/subsystems/workflow.md states moves. --- changelog.d/1531.fixed.md | 1 + plugins/workflow/src/service.js | 7 +- plugins/workflow/test/reset.test.js | 162 ++++++++++++++++++++++++++++ 3 files changed, 168 insertions(+), 2 deletions(-) create mode 100644 changelog.d/1531.fixed.md create mode 100644 plugins/workflow/test/reset.test.js diff --git a/changelog.d/1531.fixed.md b/changelog.d/1531.fixed.md new file mode 100644 index 000000000..71fb4c408 --- /dev/null +++ b/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)). diff --git a/plugins/workflow/src/service.js b/plugins/workflow/src/service.js index c2fffe829..f85ed116f 100644 --- a/plugins/workflow/src/service.js +++ b/plugins/workflow/src/service.js @@ -897,11 +897,14 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no 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 ?? {})) 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()) diff --git a/plugins/workflow/test/reset.test.js b/plugins/workflow/test/reset.test.js new file mode 100644 index 000000000..e8bf1f420 --- /dev/null +++ b/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 }) + } +})