diff --git a/changelog.d/1616-workflow-release-keeps-id.fixed.md b/changelog.d/1616-workflow-release-keeps-id.fixed.md new file mode 100644 index 000000000..21ee12916 --- /dev/null +++ b/changelog.d/1616-workflow-release-keeps-id.fixed.md @@ -0,0 +1 @@ +A run Bot the host could not remove after a Workflow run ended, was stopped, or the app closed is no longer forgotten: its id stays on the run record, so the next load retries the removal instead of leaving a hidden Bot that no screen can find again. ([#1616](https://github.com/ThinkFlowLab/FlyBot/issues/1616)) diff --git a/plugins/workflow/src/service.js b/plugins/workflow/src/service.js index 3e89485ee..093e60b3c 100644 --- a/plugins/workflow/src/service.js +++ b/plugins/workflow/src/service.js @@ -595,7 +595,6 @@ export function createWorkflowService({ // with the Bots, or a timer fires into a run that no longer exists. clearRunTimers(run.id) const ids = Object.values(run.ephemeral ?? {}) - delete run.ephemeral delete run.handed delete run.paused delete run.releasePending @@ -605,7 +604,10 @@ export function createWorkflowService({ if (run.status === 'done' || run.status === 'failed') { lane.history = [...lane.history, historyRow(run)].slice(-20) } - await releaseAll(ids, release, warn) + // A Bot the host could not remove keeps its id on the record: the next + // load's reap() is the only thing that retries the removal (#1616). + const failed = await releaseAll(ids, release, warn) + keepUnreleased(run, failed) write() } @@ -1479,6 +1481,7 @@ export function createWorkflowService({ reap() { if (typeof release !== 'function') return Promise.resolve() const stale = [] + const staleRuns = [] const abandoned = [] let wrote = false for (const [workflowId, lane] of Object.entries(state.lanes ?? {})) { @@ -1501,7 +1504,7 @@ export function createWorkflowService({ if (hasEphemeral) { clearRunTimers(run.id) stale.push(...Object.values(run.ephemeral)) - delete run.ephemeral + staleRuns.push(run) delete run.handed wrote = true } @@ -1513,7 +1516,7 @@ export function createWorkflowService({ if (run.ephemeral) stale.push(...Object.values(run.ephemeral)) endRun(workflow(workflowId), run, { code: 'host-ended' }) push(workflowId, [failureLine(run)], now()) - delete run.ephemeral + staleRuns.push(run) delete run.handed archiveCutoffRun(workflowId, run) wrote = true @@ -1526,7 +1529,12 @@ export function createWorkflowService({ write() return Promise.resolve() } - return releaseAll(stale, release, warn).then(() => { write() }) + return releaseAll(stale, release, warn).then((failed) => { + // Only the ids the host really removed leave the record. A failed one + // stays named so this same path retries it on the next load (#1616). + for (const run of staleRuns) keepUnreleased(run, failed) + write() + }) }, dispose() { for (const key of [...timers.keys()]) clearTimerAt(key) @@ -1556,9 +1564,9 @@ export function createWorkflowService({ // release cannot skip the write. The caller awaits this: a disposer that // returns before the Bots are gone is not a cleanup. const shouldWrite = ids.length > 0 || endedLive - return releaseAll(ids, release, warn).then(() => { + return releaseAll(ids, release, warn).then((failed) => { for (const run of runs) { - delete run.ephemeral + keepUnreleased(run, failed) delete run.handed } if (shouldWrite) write() @@ -1567,6 +1575,19 @@ export function createWorkflowService({ } } +// releaseAll() reports the ids it could not remove, and those Bots are still +// there. A record that forgets one leaves a hidden Bot that no state can name +// again, so keep exactly those named and drop the ids the host did remove (a +// second remove of a gone id throws, #1476). The next load's reap() retries the +// rest (#1616). +function keepUnreleased(run, failed) { + if (!run?.ephemeral) return + for (const [nodeId, id] of Object.entries(run.ephemeral)) { + if (!failed.includes(id)) delete run.ephemeral[nodeId] + } + if (Object.keys(run.ephemeral).length === 0) delete run.ephemeral +} + function seat(nodes, roster) { return (nodes ?? []).map(node => { const botId = node?.botId ? String(node.botId) : '' diff --git a/plugins/workflow/test/dispose.test.js b/plugins/workflow/test/dispose.test.js index a3974dbbd..b0386f267 100644 --- a/plugins/workflow/test/dispose.test.js +++ b/plugins/workflow/test/dispose.test.js @@ -307,3 +307,90 @@ test('#1472: dispose() releases the Bots of every concurrent live run, not just rmSync(dir, { recursive: true, force: true }) } }) + +// #1616: releaseAll() reports the ids its release could not remove, and that Bot +// is still there. Dropping the id first left a hidden Bot no state named, so the +// record must keep naming it until the host really removed it, or nothing can +// ever find it again. +test('#1616: an id whose release failed stays in the record and the next load retries it', async () => { + // (1) releaseRun(): the normal end of every run used to forget the id first. + const endDir = mkdtempSync(join(tmpdir(), 'wf-1616-end-')) + try { + const svc = createWorkflowService({ + dir: endDir, + persist: true, + now: () => CLOCK, + bots: () => roster, + spawn: async (spec) => `eph-${spec.nodeId}`, + release: async (id) => { if (id === 'eph-only') throw new Error('host call failed') }, + dispatch: async () => ({ ok: true }), + }) + const one = svc.define({ + name: 'One', + nodes: [{ id: 'only', name: 'Planner', prompt: 'Say it once.' }], + edges: [], + }).workflows.find(item => item.name === 'One') + await svc.runNamed({ workflow: 'One', brief: 'One step' }) + await svc.advance({ workflowId: one.id, nodeId: 'only', outcome: 'done' }) + const finished = readStore(endDir).lanes[one.id].last + assert.equal(finished.status, 'done') + assert.deepEqual(finished.ephemeral, { only: 'eph-only' }) + + const retried = [] + const after = createWorkflowService({ + dir: endDir, + persist: true, + now: () => CLOCK + 60_000, + bots: () => roster, + spawn: async (spec) => `eph2-${spec.nodeId}`, + release: async (id) => { retried.push(id) }, + dispatch: async () => ({ ok: true }), + }) + await after.reap() + assert.deepEqual(retried, ['eph-only']) + assert.equal(readStore(endDir).lanes[one.id].last.ephemeral, undefined) + } finally { + rmSync(endDir, { recursive: true, force: true }) + } + + // (2) dispose(): one of the two Bots refuses to go; only the other is forgotten. + const dir = mkdtempSync(join(tmpdir(), 'wf-1616-dispose-')) + const warned = [] + try { + const svc = createWorkflowService({ + dir, + persist: true, + now: () => CLOCK, + bots: () => roster, + spawn: async (spec) => `eph-${spec.nodeId}`, + release: async (id) => { if (id === 'eph-code') throw new Error('host call failed') }, + warn: (line) => warned.push(line), + dispatch: async () => ({ ok: true }), + }) + const made = pair('Leak', svc) + const started = await svc.runNamed({ workflow: 'Leak', brief: 'Ship it' }) + const run = started.workflows.find(item => item.id === made.id).run + await svc.dispose() + + const row = storedRun(readStore(dir), made.id, run.id) + assert.equal(row.status, 'failed') + assert.deepEqual(row.ephemeral, { code: 'eph-code' }) + assert.deepEqual(warned, ['flybot-workflow: could not remove eph-code: host call failed']) + + const retried = [] + const after = createWorkflowService({ + dir, + persist: true, + now: () => CLOCK + 60_000, + bots: () => roster, + spawn: async (spec) => `eph2-${spec.nodeId}`, + release: async (id) => { retried.push(id) }, + dispatch: async () => ({ ok: true }), + }) + await after.reap() + assert.deepEqual(retried, ['eph-code']) + assert.equal(storedRun(readStore(dir), made.id, run.id).ephemeral, undefined) + } finally { + rmSync(dir, { recursive: true, force: true }) + } +})