Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/1616-workflow-release-keeps-id.fixed.md
Original file line number Diff line number Diff line change
@@ -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))
35 changes: 28 additions & 7 deletions plugins/workflow/src/service.js
Original file line number Diff line number Diff line change
Expand Up @@ -167,13 +167,13 @@
}

function clearRunTimers(runId) {
for (const [key, entry] of [...timers]) {

Check warning on line 170 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-spread)

Using a spread operator here creates a new array unnecessarily.
if (entry.runId === runId) clearTimerAt(key)
}
for (const key of [...warned]) {

Check warning on line 173 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-spread)

Using a spread operator here creates a new array unnecessarily.
if (key.endsWith(`:${runId}`)) warned.delete(key)
}
for (const key of [...handOffTries.keys()]) {

Check warning on line 176 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-spread)

Using a spread operator here creates a new array unnecessarily.
if (key.startsWith(`${runId}:`)) handOffTries.delete(key)
}
}
Expand Down Expand Up @@ -595,7 +595,6 @@
// 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
Expand All @@ -605,7 +604,10 @@
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()
}

Expand Down Expand Up @@ -1479,6 +1481,7 @@
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 ?? {})) {
Expand All @@ -1501,7 +1504,7 @@
if (hasEphemeral) {
clearRunTimers(run.id)
stale.push(...Object.values(run.ephemeral))
delete run.ephemeral
staleRuns.push(run)
delete run.handed
wrote = true
}
Expand All @@ -1513,7 +1516,7 @@
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
Expand All @@ -1526,7 +1529,12 @@
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)
Expand Down Expand Up @@ -1556,9 +1564,9 @@
// 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()
Expand All @@ -1567,6 +1575,19 @@
}
}

// 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) : ''
Expand Down
87 changes: 87 additions & 0 deletions plugins/workflow/test/dispose.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -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 })
}
})
Loading