From 7b723b7dc52f76797834aa5bdceff90bc8b980e1 Mon Sep 17 00:00:00 2001 From: FlyBot Dev-4 Date: Wed, 28 Oct 2026 07:46:44 +0000 Subject: [PATCH] Freeze a run whose Stop could not release its Bots A failed release leaves the run live with only the failed id named, but the node after it has no Bot: the next step reply moved the graph there and handOff() skipped it silently, so the run held a concurrency slot forever. Mark that run paused and releasePending when the failure is written, push a system line that names the retry, and refuse resume() while the marker is set. A whole-workflow Reset freezes every run it did not reset, because each one has lost the ids whose release did go. The markers go when the run ends (Stop again, dispose, reap, or completion). Refs #1476 Rebased onto main 620235991e46c24836d03b16b76d642c9f7a4aa1 by Rework: kept every change of the reviewed tip 8d899805 (5 files) and resolved the conflicts with the newer main work on the same files (#1424 endRun, Written-by: Rework Rebased onto main bf5e8780f68fdd4b620c3004ef4d9127e9a0335a by Dev-4: kept every change of 0a70dd80 (5 files) and resolved the conflicts with #1540 (step deadline and bounded retry) on the same files; content unchanged. --- ...1476-workflow-reset-release-order.fixed.md | 1 + docs/subsystems/workflow.md | 2 +- plugins/workflow/src/ephemeral.js | 8 +- plugins/workflow/src/service.js | 110 ++++-- plugins/workflow/test/ephemeral.test.js | 332 ++++++++++++++++++ 5 files changed, 427 insertions(+), 26 deletions(-) create mode 100644 changelog.d/1476-workflow-reset-release-order.fixed.md diff --git a/changelog.d/1476-workflow-reset-release-order.fixed.md b/changelog.d/1476-workflow-reset-release-order.fixed.md new file mode 100644 index 000000000..14538954c --- /dev/null +++ b/changelog.d/1476-workflow-reset-release-order.fixed.md @@ -0,0 +1 @@ +Fixed workflow Stop and Reset writing the state change before the run's Bots were released, which erased the only id record of a Bot whose release failed. A failed release now keeps that id named on the still-live run, so the next Stop, Reset, dispose or reload retries it. The run is left Paused with `releasePending`, so it hands off nothing more and `resume` is refused; Stop again is the retry, and a whole-workflow Reset freezes every run it did not reset. ([#1476](https://github.com/ThinkFlowLab/FlyBot/issues/1476)) diff --git a/docs/subsystems/workflow.md b/docs/subsystems/workflow.md index a0baeb606..a49306034 100644 --- a/docs/subsystems/workflow.md +++ b/docs/subsystems/workflow.md @@ -46,7 +46,7 @@ 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`, but only while **Edit** is on — a factory preset is copied first (#1378). Status highlighting reads that same graph. Replacing edges does not recompute who is already running. The Settings canvas paints that graph on the same 960px design plane, scaled to the width the settings page gives it, so the canvas is not a second scroll container inside the page's own scroll; its pointer deltas are divided by the scale, and its wires keep a stronger stroke than a plain divider so an edge can be followed at that size (#1483). -3. **A live run advances only when its ephemeral Bot replies, following the graph's edges in order.** That reply advances the run and may appear on the workflow stage log; it is not delivered into a persistent sidebar Bot conversation (for example Chief). The `workflow-step` send middleware denies those replies after `stepResult` runs. Nodes start only when their forward predecessors finish and edges reach them. **Early termination (#1537, #1538):** When a node completes with `needs-human` or `already-claimed` detected in its outputs or reply text (see `checkEarlyEnd` in Key files), the run ends immediately with status `done`, sets `run.earlyEnd` to the condition name, and does not advance to or spawn successor nodes. This allows Triage, Claim, or any step to terminate the pipeline cleanly when human intervention is required or the issue is already owned by another agent. There is no scripted or seeded run path: `workflow/state` exposes only `live === true` runs. A run does not create a schedule. `scheduleLabel` is display text. When the host can create Bots, starting a live run (`assign`, `run_workflow`, `ingest`) 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 default workflows. Plugin unload (`dispose`) and the load-time `reap` both end a live run still marked `running`: in-flight nodes are failed, the run is archived as `failed`, and any ephemeral Bot ids are released. A run already written as `running` with no ephemeral map (an older unload) is healed the same way on the next load (#1414). Loading is also the one moment a run the host left behind is noticed when its ephemeral map is still present: `reap` releases those instances and records the run as `failed`. Nothing resumes it. A run that already ended keeps its own status (#1388). `workflow/reset` stopping one live run ends it the same way before its Bots are released: nodes still `running` become `failed`, so the strip cannot keep painting a run that is over (#1424). Every wait is bounded (#1530). A dispatch that does not land is not marked handed: the chat names the code and the attempt, and it is retried after a short backoff at most twice, never twice inside one hand-off, so a 429 is not hammered. When it still does not land, the step and the run fail with the reason written into the chat. Each running step also holds a deadline — `STEP_DEADLINE_MS`, ten minutes, one constant in `service.js`; when it expires the node and the run fail as `failed`, the run's ephemeral Bots are released, the chat says the step timed out, and `warn` names it once. The deadline is cleared as soon as the step moves, and by Pause, Stop, Reset, delete, `reap` and `dispose`; a paused run holds none, and Resume arms it again from that moment. A timer never outlives the run that started it. A reply that matches no running step — its run ended, or the node already moved — is reported through `warn` once per Bot and run instead of being dropped silently. A reply that says `could not answer your message` or `model call failed` fails that node and does not start its successor. A 429 or rate-limit mention in any other reply does not. +3. **A live run advances only when its ephemeral Bot replies, following the graph's edges in order.** That reply advances the run and may appear on the workflow stage log; it is not delivered into a persistent sidebar Bot conversation (for example Chief). The `workflow-step` send middleware denies those replies after `stepResult` runs. Nodes start only when their forward predecessors finish and edges reach them. **Early termination (#1537, #1538):** When a node completes with `needs-human` or `already-claimed` detected in its outputs or reply text (see `checkEarlyEnd` in Key files), the run ends immediately with status `done`, sets `run.earlyEnd` to the condition name, and does not advance to or spawn successor nodes. This allows Triage, Claim, or any step to terminate the pipeline cleanly when human intervention is required or the issue is already owned by another agent. There is no scripted or seeded run path: `workflow/state` exposes only `live === true` runs. A run does not create a schedule. `scheduleLabel` is display text. When the host can create Bots, starting a live run (`assign`, `run_workflow`, `ingest`) 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 default workflows. Plugin unload (`dispose`) and the load-time `reap` both end a live run still marked `running`: in-flight nodes are failed, the run is archived as `failed`, and any ephemeral Bot ids are released. A run already written as `running` with no ephemeral map (an older unload) is healed the same way on the next load (#1414). Loading is also the one moment a run the host left behind is noticed when its ephemeral map is still present: `reap` releases those instances and records the run as `failed`. Nothing resumes it. A run that already ended keeps its own status (#1388). `workflow/reset` stopping one live run ends it the same way before its Bots are released: nodes still `running` become `failed`, so the strip cannot keep painting a run that is over (#1424). Every wait is bounded (#1530). A dispatch that does not land is not marked handed: the chat names the code and the attempt, and it is retried after a short backoff at most twice, never twice inside one hand-off, so a 429 is not hammered. When it still does not land, the step and the run fail with the reason written into the chat. Each running step also holds a deadline — `STEP_DEADLINE_MS`, ten minutes, one constant in `service.js`; when it expires the node and the run fail as `failed`, the run's ephemeral Bots are released, the chat says the step timed out, and `warn` names it once. The deadline is cleared as soon as the step moves, and by Pause, Stop, Reset, delete, `reap` and `dispose`; a paused run holds none, and Resume arms it again from that moment. A timer never outlives the run that started it. A reply that matches no running step — its run ended, or the node already moved — is reported through `warn` once per Bot and run instead of being dropped silently. A reply that says `could not answer your message` or `model call failed` fails that node and does not start its successor. A 429 or rate-limit mention in any other reply does not. Stop and Reset (`workflow/reset`) follow the same order as `dispose`: they release the run's Bots before the record forgets their ids, and an id whose release failed stays named on the still-live run, so the next Stop, Reset, dispose or load-time `reap` retries it (#1476). A run left that way is marked `paused` and `releasePending` and shows as Paused: it hands off no further node, `resume` is refused with an error, and Stop again is the retry. A whole-workflow Reset that could not release every Bot freezes every run it did not reset, because each of them has lost the ids whose release did go. The markers go away when the run ends: Stop again, `dispose`, `reap`, or its own completion. 4. Each workflow keeps a FIFO queue and up to `maxConcurrent` live runs (default 2, an integer from 1 to 8 on the record, via `workflow/concurrency`, or in Settings → Workflow graphs). A task that does not fit is enqueued. When a run finishes, every workflow with a free slot is drained, oldest enqueue first. A new assign on a workflow that already has a queue waits behind it. A task does not replace a running run or release that run's Bots. Migrating an old store keeps only runs that were still `running` with ephemeral Bot ids; other legacy scripted rows are dropped from active state (finished tasks may remain in lane history). History records a finished or failed live run. Lowering the cap does not stop a run that has already started. Assigning `{ botId }` claims the card for that local Bot and does not enqueue. A remote Bot is refused. `activeCount` is `runningOf(lane)` — live runs with `live === true` and `status === 'running'`. The stage aggregate chip and the Runs tab (运行记录) must agree: every such run appears as its own running row and progress card, and the chip count does not exceed that list. 5. A node declares up to eight inputs and eight outputs. A required input is an output of the same name on an ancestor, or `task`, `title`, or `brief` on a root. A start with no provider is refused, and Bots spawned for that attempt are released. On a Bot-backed run a successor stays queued until its required inputs are present. If every forward predecessor is done, failed or skipped and an input is still missing, that node fails and the run stops. The dispatch text carries `Input : `. The Bot reports `OUTPUT : `; a single declared output may be the reply body. Nodes start in edge order. 6. Each node is a real ephemeral Bot for that run, then it is removed. A permanent sidebar Bot (`botId`) is optional and is not required for the step to run. **File and shell rights are the same as a normal sidebar Bot's:** every node Bot gets the `danger-full-access` preset, and `EPHEMERAL_DENIED_TOOLS` still refuses the team-admin tools at the guard and approval layers (#1466, 29 names). **`node.tools` selects connectors only:** `spawn` passes the node's cleaned list to `fb.bot.create`, the host records it as `stepTools`, and `connector-access.js` allows `mcp__github__*` for `github` and `mcp__composio__*` for `composio` — a connector the node did not name is refused, and a node that names none reaches none. The list is also stored in `fields.tools` for the UI. A hidden run Bot cannot show an approval card, so the approval seam answers the same way the guard does. The host stores this on the run Bot as `stepTools` and `permission`. diff --git a/plugins/workflow/src/ephemeral.js b/plugins/workflow/src/ephemeral.js index 483c3f9c4..2e404aa52 100644 --- a/plugins/workflow/src/ephemeral.js +++ b/plugins/workflow/src/ephemeral.js @@ -50,15 +50,19 @@ export async function spawnAll(nodes, roster, spawn, workflowModel = '') { } } +// Returns the ids whose release failed: that Bot is still there, so the caller +// must keep naming it and let a later attempt retry the release (#1476). export async function releaseAll(ids, release, warn = () => {}) { - if (typeof release !== 'function') return + const failed = [] + if (typeof release !== 'function') return failed for (const id of ids) { if (!id) continue try { await release(id) } catch (error) { - // The instance may already be gone. The run record still drops the id. + failed.push(id) warn(`flybot-workflow: could not remove ${id}: ${error?.message ?? error}`) } } + return failed } diff --git a/plugins/workflow/src/service.js b/plugins/workflow/src/service.js index 45f7109e3..f36952133 100644 --- a/plugins/workflow/src/service.js +++ b/plugins/workflow/src/service.js @@ -522,6 +522,8 @@ export function createWorkflowService({ const ids = Object.values(run.ephemeral ?? {}) delete run.ephemeral delete run.handed + delete run.paused + delete run.releasePending const lane = laneOf(workflowId) lane.active = lane.active.filter(item => item.id !== run.id) lane.last = run @@ -549,6 +551,8 @@ export function createWorkflowService({ } run.status = 'failed' run.updatedAt = now() + delete run.paused + delete run.releasePending return run } @@ -1007,6 +1011,10 @@ export function createWorkflowService({ const runId = body.runId ? String(body.runId) : '' 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 resume' } + // A failed release leaves the run paused on purpose: its next node has no + // Bot to hand the step to, so resuming would only move the graph into a + // 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) @@ -1023,6 +1031,21 @@ export function createWorkflowService({ reset(body = {}) { const workflowId = String(body.workflowId ?? '') const runId = body.runId ? String(body.runId) : '' + // Release first, then forget the ids (#1476, #1387). A failed release is + // reported through `warn` rather than rejected, so a transition written + // before the release would erase the only name of a Bot that is still + // alive. Only the failed ids stay named; the ones that did go are dropped + // because a retry must not ask the host to remove an id it already deleted + // (that throws), and the next Stop, Reset, dispose() or reload retries. + const forgetReleased = (runs, released) => { + const gone = new Set(released) + for (const run of runs) { + for (const [nodeId, id] of Object.entries(run?.ephemeral ?? {})) { + if (!gone.has(id)) continue + delete run.ephemeral[nodeId] + } + } + } if (runId) { const stopOne = () => { const item = workflow(workflowId) @@ -1031,29 +1054,50 @@ export function createWorkflowService({ const run = lane.active.find(itemRun => itemRun.id === runId) if (!run || run.live !== true || run.status !== 'running') return { ok: false, error: 'No live run to stop' } const ids = Object.values(run.ephemeral ?? {}) - // Stopping one run ends it the same way a reap does: the node map settles - // with the run, or the strip keeps painting the node it was on (#1424). - clearRunTimers(run.id) - const ended = { ...endRun(item, run) } - delete ended.ephemeral - delete ended.handed - lane.active = lane.active.filter(itemRun => itemRun.id !== runId) - lane.last = ended - lane.history = [...lane.history, { task: ended.task, status: ended.status, at: ended.updatedAt }].slice(-20) - push(workflowId, [{ kind: 'system', tone: 'bad', text: `■ 已停止 · ${summarizeTask(run.task)}` }], now()) - write() - return { ok: true, ids } + const stop = () => { + clearRunTimers(run.id) + // Stopping one run ends it the same way a reap does: the node map settles + // with the run, or the strip keeps painting the node it was on (#1424). + const ended = { ...endRun(item, run) } + delete ended.ephemeral + delete ended.handed + delete ended.paused + delete ended.releasePending + lane.active = lane.active.filter(itemRun => itemRun.id !== runId) + lane.last = ended + lane.history = [...lane.history, { task: ended.task, status: ended.status, at: ended.updatedAt }].slice(-20) + push(workflowId, [{ kind: 'system', tone: 'bad', text: `■ 已停止 · ${summarizeTask(run.task)}` }], now()) + write() + } + // The run stays live with only the failed ids named, and it is paused + // so handOff() dispatches nothing. Without releasePending a Resume + // would move the graph to a node that has no Bot and wait forever. + const hold = (failed) => { + forgetReleased([run], ids.filter(id => !failed.includes(id))) + run.paused = true + run.releasePending = true + run.updatedAt = now() + push(workflowId, [{ kind: 'system', tone: 'bad', text: `停止失败：还有 ${failed.length} 个 Bot 没释放，再点一次停止。` }], now()) + write() + } + return { ok: true, ids, stop, hold } } if (typeof spawn !== 'function') { const saved = stopOne() if (saved.ok === false) return saved + saved.stop() drainSync() return present() } return hostSerial(async () => { const saved = stopOne() if (saved.ok === false) return saved - await releaseAll(saved.ids, release, warn) + const failed = await releaseAll(saved.ids, release, warn) + if (failed.length > 0) { + saved.hold(failed) + return present() + } + saved.stop() await drainAll() return present() }) @@ -1062,26 +1106,46 @@ export function createWorkflowService({ 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) 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()) - write() - return { ok: true, ids } + const ids = lane.active.flatMap(run => Object.values(run.ephemeral ?? {})) + const resetLane = () => { + lane.active = [] + lane.queue = [] + lane.last = { ...blankRun(item, live.task, now(), lane.history), id: live.id, startedAt: live.startedAt } + push(item.id, linesFor(item, live.task, { type: 'reset' }), now()) + write() + } + // Every live run is left without the Bots that did go, so every one of + // them is frozen, not just the run that named a failed id (#1476). + const hold = (failed) => { + forgetReleased(lane.active, ids.filter(id => !failed.includes(id))) + for (const run of lane.active) { + run.paused = true + run.releasePending = true + run.updatedAt = now() + } + push(item.id, [{ kind: 'system', tone: 'bad', text: `重置失败：还有 ${failed.length} 个 Bot 没释放，再点一次重置。` }], now()) + write() + } + return { ok: true, ids, resetLane, hold } } if (typeof spawn !== 'function') { const saved = take() - return saved.ok === false ? saved : present() + if (saved.ok === false) return saved + saved.resetLane() + return present() } return hostSerial(async () => { const saved = take() if (saved.ok === false) return saved - await releaseAll(saved.ids, release, warn) + const failed = await releaseAll(saved.ids, release, warn) + if (failed.length > 0) { + saved.hold(failed) + return present() + } + saved.resetLane() await drainAll() return present() }) diff --git a/plugins/workflow/test/ephemeral.test.js b/plugins/workflow/test/ephemeral.test.js index 7144a97dc..a1d56a3e1 100644 --- a/plugins/workflow/test/ephemeral.test.js +++ b/plugins/workflow/test/ephemeral.test.js @@ -763,3 +763,335 @@ test('the built-in delivery nodes hand their connector list to the host', async assert.deepEqual(byNode.get('planner').tools, [], 'Planner names no connector') for (const spec of created) assert.equal(spec.permission, 'danger-full-access', 'file and shell rights are a normal Bot\'s') }) + +// #1476: Stop and Reset persist the run's end before its Bots are released, so +// a release that fails leaves a live Bot whose only id has already been written +// away. Both paths release first and keep the failed id named, so the next Stop, +// Reset, dispose() or reload can retry it. +function resettable(dir, log) { + let seq = 0 + return createWorkflowService({ + dir, + persist: true, + now: () => 1_700_000_000_000, + bots: () => roster, + spawn: async (spec) => `eph-${spec.nodeId}-${++seq}`, + release: async (id) => { + log.attempted.push(id) + if (log.fail.has(id)) { + log.fail.delete(id) + throw new Error('bot remove failed') + } + log.done.push(id) + }, + warn: (line) => { log.warned.push(line) }, + dispatch: async (step) => { + log.dispatched?.push(`${step.nodeId}:${step.botId}`) + return { ok: true } + }, + }) +} + +function twoNodes(svc, name = 'Lane') { + 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('a release that fails during Stop keeps the id, and a second Stop retries it (#1476)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-1476-stop-')) + const log = { attempted: [], done: [], warned: [], fail: new Set(['eph-plan-1']) } + try { + const svc = resettable(dir, log) + const made = twoNodes(svc) + svc.setConcurrency({ workflowId: made.id, maxConcurrent: 2 }) + await svc.assign({ workflowId: made.id, taskId: '#a', title: 'First' }) + const both = await svc.assign({ workflowId: made.id, taskId: '#b', title: 'Second' }) + const lane = both.workflows.find(item => item.id === made.id) + const victim = lane.runs.find(run => run.task.id === '#a') + const survivor = lane.runs.find(run => run.task.id === '#b') + const victimIds = Object.values(victim.ephemeral) + + const stopped = await svc.reset({ workflowId: made.id, runId: victim.id }) + assert.equal(stopped.ok, true) + assert.match(log.warned.join('\n'), /could not remove eph-plan-1/) + // The release failed, so the run is still live and the record still names it. + const during = readStore(dir) + assert.equal(JSON.stringify(during).includes('eph-plan-1'), true, 'a failed release erased the only id') + const kept = (during.lanes[made.id].active ?? []).find(run => run.id === victim.id) + assert.ok(kept, 'Stop ended the run although its Bot could not be released') + assert.equal(kept.status, 'running') + assert.deepEqual(kept.ephemeral, { plan: 'eph-plan-1' }) + // The run is halted on purpose, and the marker says why: resuming it would + // move the graph to a node whose Bot is already gone (#1476). + assert.equal(kept.paused, true) + assert.equal(kept.releasePending, true) + // The id whose release did go is forgotten, so no retry asks for it again. + assert.equal(JSON.stringify(during).includes('eph-code-2'), false) + assert.equal(stopped.workflows.find(item => item.id === made.id).runs.some(run => run.id === victim.id), true) + + // A second Stop retries the one id that is still there and then ends the run. + const retried = await svc.reset({ workflowId: made.id, runId: victim.id }) + assert.equal(retried.ok, true) + assert.equal(log.warned.length, 1) + assert.deepEqual([...new Set(log.done)].sort(), [...victimIds].sort()) + assert.equal(log.attempted.filter(id => id === 'eph-plan-1').length, 2) + assert.equal(log.attempted.includes('eph-code-2'), true) + assert.equal(log.attempted.filter(id => id === 'eph-code-2').length, 1) + const after = readStore(dir) + assert.deepEqual((after.lanes[made.id].active ?? []).map(run => run.id), [survivor.id]) + assert.equal(after.lanes[made.id].last.id, victim.id) + assert.equal(after.lanes[made.id].last.status, 'failed') + assert.equal(after.lanes[made.id].last.ephemeral, undefined) + assert.equal(after.lanes[made.id].last.paused, undefined) + assert.equal(after.lanes[made.id].last.releasePending, undefined) + const text = readFileSync(join(dir, 'workflows.json'), 'utf8') + for (const id of victimIds) assert.equal(text.includes(id), false) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('a failed Stop freezes the run as Paused, so nothing more is handed and the queue waits (#1476)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-1476-freeze-')) + const log = { attempted: [], done: [], warned: [], dispatched: [], fail: new Set(['eph-plan-1']) } + try { + const svc = resettable(dir, log) + const made = twoNodes(svc) + svc.setConcurrency({ workflowId: made.id, maxConcurrent: 2 }) + await svc.assign({ workflowId: made.id, taskId: '#a', title: 'First' }) + const both = await svc.assign({ workflowId: made.id, taskId: '#b', title: 'Second' }) + const lane = both.workflows.find(item => item.id === made.id) + const victim = lane.runs.find(run => run.task.id === '#a') + const survivor = lane.runs.find(run => run.task.id === '#b') + const handedBefore = [...log.dispatched] + + const stopped = await svc.reset({ workflowId: made.id, runId: victim.id }) + assert.equal(stopped.ok, true) + const during = readStore(dir) + const kept = during.lanes[made.id].active.find(run => run.id === victim.id) + assert.equal(kept.status, 'running') + assert.equal(kept.paused, true) + assert.equal(kept.releasePending, true) + const shown = stopped.workflows.find(item => item.id === made.id).runs.find(run => run.id === victim.id) + assert.equal(shown.paused, true, 'the UI must show the run as Paused') + const note = stopped.workflows.find(item => item.id === made.id).messages.at(-1) + assert.equal(note.tone, 'bad') + assert.match(note.text, /停止失败：还有 1 个 Bot 没释放，再点一次停止/) + + // The last reply still moves the graph, but the next node has no Bot id and + // handOff() must send nothing: before the freeze it stayed running forever. + await svc.stepResult({ botId: 'eph-plan-1', text: 'plan done' }) + assert.deepEqual(log.dispatched, handedBefore) + assert.equal(log.dispatched.some(line => line.startsWith('code:')), false) + + // The wedged run holds its slot, so a queued task is not drained into it. + const third = await svc.assign({ workflowId: made.id, taskId: '#c', title: 'Third' }) + assert.deepEqual(readStore(dir).lanes[made.id].queue.map(task => task.id), ['#c']) + assert.deepEqual(third.workflows.find(item => item.id === made.id).runs.map(run => run.id).sort(), [victim.id, survivor.id].sort()) + assert.deepEqual(log.dispatched, handedBefore) + + // A person's own Pause and Resume on a healthy run are unchanged. + assert.equal(svc.pause({ workflowId: made.id, runId: survivor.id }).ok, true) + assert.equal((await svc.resume({ workflowId: made.id, runId: survivor.id })).ok, true) + assert.equal(readStore(dir).lanes[made.id].active.find(run => run.id === survivor.id).paused, false) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('resume refuses a run whose failed Stop still waits to release a Bot (#1476)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-1476-resume-')) + const log = { attempted: [], done: [], warned: [], dispatched: [], fail: new Set(['eph-plan-1']) } + try { + const svc = resettable(dir, log) + const made = twoNodes(svc) + await svc.assign({ workflowId: made.id, taskId: '#a', title: 'Only' }) + const run = svc.state().workflows.find(item => item.id === made.id).run + await svc.reset({ workflowId: made.id, runId: run.id }) + const before = readStore(dir) + const frozen = before.lanes[made.id].active.find(item => item.id === run.id) + assert.equal(frozen.paused, true) + assert.equal(frozen.releasePending, true) + const messagesBefore = (before.logs?.[made.id] ?? []).length + + const refused = await svc.resume({ workflowId: made.id, runId: run.id }) + assert.equal(refused.ok, false) + assert.match(refused.error, /release|stop/i) + // Nothing moved: not a flag, not a timestamp, not a log line. + const after = readStore(dir) + assert.deepEqual(after.lanes[made.id].active.find(item => item.id === run.id), frozen) + assert.equal((after.logs?.[made.id] ?? []).length, messagesBefore) + assert.equal(after.logs?.[made.id]?.at(-1)?.text?.includes('已继续') ?? false, false) + // The name-less form the UI may send is refused the same way. + assert.equal((await svc.resume({ workflowId: made.id })).ok, false) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('a reload reaps a run a failed Stop froze, and the markers go with it (#1476)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-1476-reap-')) + const log = { attempted: [], done: [], warned: [], dispatched: [], fail: new Set(['eph-plan-1']) } + try { + const svc = resettable(dir, log) + const made = twoNodes(svc) + await svc.assign({ workflowId: made.id, taskId: '#a', title: 'Only' }) + const run = svc.state().workflows.find(item => item.id === made.id).run + await svc.reset({ workflowId: made.id, runId: run.id }) + assert.equal(readStore(dir).lanes[made.id].active[0].releasePending, true) + + log.attempted.length = 0 + log.done.length = 0 + const reloaded = createWorkflowService({ + dir, + persist: true, + now: () => 1_700_000_001_000, + bots: () => roster, + spawn: async (spec) => `eph-${spec.nodeId}-reload`, + release: async (id) => { log.attempted.push(id); log.done.push(id) }, + warn: (line) => { log.warned.push(line) }, + }) + await reloaded.reap() + const after = readStore(dir) + assert.equal((after.lanes[made.id].active ?? []).length, 0) + assert.equal(after.lanes[made.id].last.status, 'failed') + assert.equal(after.lanes[made.id].last.ephemeral, undefined) + assert.equal(after.lanes[made.id].last.paused, undefined) + assert.equal(after.lanes[made.id].last.releasePending, undefined) + assert.deepEqual(log.done, ['eph-plan-1']) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('a release that fails during Stop leaves the id for a following dispose() to retry (#1476)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-1476-dispose-')) + const log = { attempted: [], done: [], warned: [], fail: new Set(['eph-plan-1']) } + try { + const svc = resettable(dir, log) + const made = twoNodes(svc) + await svc.assign({ workflowId: made.id, taskId: '#a', title: 'Only' }) + const run = svc.state().workflows.find(item => item.id === made.id).run + const ids = Object.values(run.ephemeral) + + const stopped = await svc.reset({ workflowId: made.id, runId: run.id }) + assert.equal(stopped.ok, true) + assert.match(log.warned.join('\n'), /could not remove eph-plan-1/) + const during = readStore(dir) + assert.equal((during.lanes[made.id].active ?? []).length, 1) + assert.deepEqual(during.lanes[made.id].active[0].ephemeral, { plan: 'eph-plan-1' }) + + // dispose() is the next path that reaches the retained id, and it releases it. + await svc.dispose() + assert.deepEqual([...new Set(log.done)].sort(), [...ids].sort()) + assert.equal(log.attempted.filter(id => id === 'eph-plan-1').length, 2) + const after = readStore(dir) + assert.equal((after.lanes[made.id].active ?? []).length, 0) + assert.equal(after.lanes[made.id].last.ephemeral, undefined) + // The run ended, so the frozen markers are gone with it. + assert.equal(after.lanes[made.id].last.paused, undefined) + assert.equal(after.lanes[made.id].last.releasePending, undefined) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('a release that fails during a whole-workflow Reset keeps that id, and a second Reset retries it (#1476)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-1476-reset-')) + const log = { attempted: [], done: [], warned: [], fail: new Set(['eph-plan-1']) } + try { + const svc = resettable(dir, log) + const made = twoNodes(svc) + svc.setConcurrency({ workflowId: made.id, maxConcurrent: 2 }) + await svc.assign({ workflowId: made.id, taskId: '#a', title: 'First' }) + const both = await svc.assign({ workflowId: made.id, taskId: '#b', title: 'Second' }) + const lane = both.workflows.find(item => item.id === made.id) + const allIds = lane.runs.flatMap(run => Object.values(run.ephemeral)) + + const reset = await svc.reset({ workflowId: made.id }) + assert.equal(reset.ok, true) + assert.match(log.warned.join('\n'), /could not remove eph-plan-1/) + // Nothing was reset: the lanes are still live, and the failed id still has a name. + const during = readStore(dir) + assert.equal((during.lanes[made.id].active ?? []).length, 2) + assert.equal(JSON.stringify(during).includes('eph-plan-1'), true) + // Every live run lost the ids that did go, so every one of them is frozen. + for (const run of during.lanes[made.id].active) { + assert.equal(run.paused, true) + assert.equal(run.releasePending, true) + } + for (const id of allIds.filter(id => id !== 'eph-plan-1')) { + assert.equal(JSON.stringify(during).includes(id), false) + } + assert.equal((during.lanes[made.id].queue ?? []).length, 0) + + // The next Reset is the retry: the same endpoint releases and then resets. + const retried = await svc.reset({ workflowId: made.id }) + assert.equal(retried.ok, true) + assert.deepEqual([...new Set(log.done)].sort(), [...new Set(allIds)].sort()) + const after = readStore(dir) + assert.equal((after.lanes[made.id].active ?? []).length, 0) + assert.equal(after.lanes[made.id].last.ephemeral, undefined) + assert.equal(after.lanes[made.id].last.paused, undefined) + assert.equal(after.lanes[made.id].last.releasePending, undefined) + const text = readFileSync(join(dir, 'workflows.json'), 'utf8') + for (const id of allIds) assert.equal(text.includes(id), false) + assert.equal(log.warned.length, 1) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +test('Stop and Reset still forget every id and archive the run when no release fails (#1476)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-1476-ok-')) + const log = { attempted: [], done: [], warned: [], fail: new Set() } + try { + const svc = resettable(dir, log) + const made = twoNodes(svc) + svc.setConcurrency({ workflowId: made.id, maxConcurrent: 2 }) + await svc.assign({ workflowId: made.id, taskId: '#a', title: 'First' }) + const both = await svc.assign({ workflowId: made.id, taskId: '#b', title: 'Second' }) + const lane = both.workflows.find(item => item.id === made.id) + const victim = lane.runs.find(run => run.task.id === '#a') + const survivor = lane.runs.find(run => run.task.id === '#b') + const victimIds = Object.values(victim.ephemeral) + + const stopped = await svc.reset({ workflowId: made.id, runId: victim.id }) + assert.equal(stopped.ok, true) + assert.deepEqual(log.warned, []) + const afterStop = readStore(dir) + const archived = afterStop.lanes[made.id].last + assert.equal(archived.id, victim.id) + assert.equal(archived.status, 'failed') + assert.equal(archived.ephemeral, undefined) + assert.equal('handed' in archived, false) + assert.deepEqual((afterStop.lanes[made.id].active ?? []).map(run => run.id), [survivor.id]) + const text = readFileSync(join(dir, 'workflows.json'), 'utf8') + for (const id of victimIds) assert.equal(text.includes(id), false) + const presented = stopped.workflows.find(item => item.id === made.id) + assert.equal(presented.history.at(-1).status, 'failed') + assert.match(presented.messages.at(-1).text, /已停止/) + + // The whole-workflow Reset still blanks the lane and keeps the record clean. + const reset = await svc.reset({ workflowId: made.id }) + assert.equal(reset.ok, true) + const afterReset = readStore(dir) + assert.equal((afterReset.lanes[made.id].active ?? []).length, 0) + assert.equal((afterReset.lanes[made.id].queue ?? []).length, 0) + assert.equal(afterReset.lanes[made.id].last.ephemeral, undefined) + assert.equal(afterReset.lanes[made.id].last.status, 'idle') + assert.match(reset.workflows.find(item => item.id === made.id).messages.at(-1).text, /重置|reset/i) + const finalText = readFileSync(join(dir, 'workflows.json'), 'utf8') + for (const id of Object.values(survivor.ephemeral)) assert.equal(finalText.includes(id), false) + assert.deepEqual([...new Set(log.done)].sort(), [...new Set([...victimIds, ...Object.values(survivor.ephemeral)])].sort()) + assert.equal(log.warned.length, 0) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +})