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/1569.fixed.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
On a read-only host — another `dsh` holds this team — `workflow/advance` and `workflow/reset` no longer move a run or write `workflows.json`: both refuse with `{ ok: false, code: 'read-only', error: 'The host is read-only' }` before they touch the run, the same way `pause` does. A node Bot's reply that would advance a run is reported once through the plugin's warning path instead of being swallowed, and a writable host is unchanged. ([#1569](https://github.com/ThinkFlowLab/FlyBot/issues/1569))
2 changes: 2 additions & 0 deletions plugins/workflow/src/service.js
Original file line number Diff line number Diff line change
Expand Up @@ -151,13 +151,13 @@
}

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

Check warning on line 154 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 157 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 160 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 @@ -185,7 +185,7 @@
function syncTimers(item, run) {
if (run?.live !== true || !run.ephemeral) return
const running = new Set(orderedIds(item, run.nodeStatus, 'running'))
for (const [key, entry] of [...timers]) {

Check warning on line 188 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 !== run.id) continue
if (run.status !== 'running' || run.paused === true || !running.has(entry.nodeId)) clearTimerAt(key)
}
Expand Down Expand Up @@ -360,7 +360,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')

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

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-length-check)

Found a useless array length check
if (stuck) {
status[id] = 'failed'
blocked.push({ id, missing })
Expand All @@ -373,7 +373,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 } }

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

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-fallback-in-spread)

Empty fallbacks in spreads are unnecessary
}
let result
if (op.type === 'start') result = { ...startFrontier(item, working.nodeStatus, working.fired), joined: false, layerSize: 0 }
Expand Down Expand Up @@ -507,7 +507,7 @@
break
}
handOffTries.delete(`${current.id}:${nodeId}`)
current = { ...current, handed: { ...current.handed, [nodeId]: true }, inputs: { ...(current.inputs ?? {}), [nodeId]: body.inputs } }

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

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-fallback-in-spread)

Empty fallbacks in spreads are unnecessary
storeRun(item.id, current)
}
syncTimers(item, current)
Expand Down Expand Up @@ -918,6 +918,7 @@
return present()
},
advance(body = {}) {
if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' }
const workflowId = String(body.workflowId ?? '')
const outcome = body.outcome === 'failed' ? 'fail' : 'complete'
const nodeId = String(body.nodeId ?? '')
Expand Down Expand Up @@ -1021,6 +1022,7 @@
})
},
reset(body = {}) {
if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' }
const workflowId = String(body.workflowId ?? '')
const runId = body.runId ? String(body.runId) : ''
if (runId) {
Expand Down Expand Up @@ -1230,12 +1232,12 @@
return releaseAll(stale, release, warn).then(() => { write() })
},
dispose() {
for (const key of [...timers.keys()]) clearTimerAt(key)

Check warning on line 1235 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.
const ids = []
const runs = []
let endedLive = false
for (const [workflowId, lane] of Object.entries(state.lanes ?? {})) {
for (const run of [...(lane.active ?? [])]) {

Check warning on line 1240 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 (run?.live === true && run.status === 'running') {
endRun(workflow(workflowId), run)
archiveCutoffRun(workflowId, run)
Expand Down
32 changes: 32 additions & 0 deletions plugins/workflow/test/assign.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,38 @@ test('a read-only host refuses assign', () => {
assert.equal(refused.code, 'read-only')
})

// #1569: advance and reset are writes like pause. A read-only host refuses them
// before they touch the run or the file, so neither the live run nor its queue moves.
test('#1569: a read-only host refuses advance and reset without moving the run', () => {
let readOnly = false
const svc = service({ readOnly: () => readOnly })
svc.setConcurrency({ workflowId: DELIVERY_ID, maxConcurrent: 1 })
svc.assign({ workflowId: DELIVERY_ID, taskId: 'task-first', title: 'First' })
svc.assign({ workflowId: DELIVERY_ID, taskId: 'task-queued', title: 'Queued behind it' })
const before = delivery(svc.state())
assert.equal(before.run.nodeStatus.planner, 'running')
assert.equal(before.queueDepth, 1)

readOnly = true
const refusedAdvance = svc.advance({ workflowId: DELIVERY_ID, nodeId: 'planner', outcome: 'done' })
assert.deepEqual(refusedAdvance, { ok: false, code: 'read-only', error: 'The host is read-only' })
const refusedReset = svc.reset({ workflowId: DELIVERY_ID })
assert.deepEqual(refusedReset, { ok: false, code: 'read-only', error: 'The host is read-only' })

const after = delivery(svc.state())
assert.equal(after.activeCount, 1)
assert.equal(after.run.status, 'running')
assert.equal(after.run.nodeStatus.planner, 'running')
assert.equal(after.queueDepth, 1)
assert.deepEqual(after.queue.map(task => task.id), ['task-queued'])

// Nothing else changes on a writable host: the same call still advances.
readOnly = false
const moved = svc.advance({ workflowId: DELIVERY_ID, nodeId: 'planner', outcome: 'done' })
assert.equal(moved.ok, true)
assert.equal(delivery(moved).run.nodeStatus.writer, 'running')
})

test('the store reloads a workflow file and drops a file that does not parse', () => {
const dir = mkdtempSync(join(tmpdir(), 'flybot-wf-'))
try {
Expand Down
33 changes: 33 additions & 0 deletions plugins/workflow/test/ephemeral.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,39 @@ test('#1389: a step the host refuses is not marked as handed, and a later settle
assert.match(warnings.join('\n'), /no-sender/)
})

// #1569: the Bot reply hook calls service.stepResult directly, so it never passes
// through the host channel's read-only check. advance's own guard has to stop it,
// and the refusal is reported once instead of being swallowed.
test('#1569: a read-only host refuses the Bot reply that would advance a run', async () => {
let readOnly = false
const warnings = []
const { svc } = harness({ service: { readOnly: () => readOnly, warn: (...args) => warnings.push(args.join(' ')) } })
const made = svc.define({
name: 'Pair',
nodes: [
{ id: 'plan', name: 'Planner', role: 'plan', prompt: 'Write the plan.' },
{ id: 'code', name: 'Coder', role: 'code', prompt: 'Implement it.' },
],
edges: [],
}).workflows.find(item => item.name === 'Pair')
const started = await svc.runNamed({ workflow: 'Pair', brief: 'Ship the note' })
const run = started.workflows.find(item => item.id === made.id).run
const planner = run.ephemeral.plan
assert.equal(run.nodeStatus.plan, 'running')

readOnly = true
await svc.stepResult({ botId: planner, ok: true, text: 'The plan is ready.' })
const after = svc.state().workflows.find(item => item.id === made.id).run
assert.equal(after.status, 'running')
assert.equal(after.nodeStatus.plan, 'running')
assert.equal(after.doneCount, run.doneCount)
assert.match(warnings.join('\n'), /read-only/)

// The same step replies again: still refused, and the warning is not repeated.
await svc.stepResult({ botId: planner, ok: true, text: 'The plan is ready.' })
assert.equal(warnings.filter(line => line.includes('read-only')).length, 1)
})

// #1388: reap is the only path that cleans up a run whose host went away, and the
// plugin calls it exactly once, at load. A run still marked `running` at that
// moment is what a restart leaves behind: nothing resumes it, so its ephemeral
Expand Down
Loading