diff --git a/changelog.d/2704-workflow-takeover.fixed.md b/changelog.d/2704-workflow-takeover.fixed.md new file mode 100644 index 000000000..73dfd6620 --- /dev/null +++ b/changelog.d/2704-workflow-takeover.fixed.md @@ -0,0 +1 @@ +A window that takes over a workflow store after the other window closes no longer writes the copy it loaded while the other one was still running: the first read or write after it becomes the writer re-reads `workflows.json`, and it reaps the run the closed window left behind instead of leaving the lane blocked. A card that was only queued is kept and starts normally. See [#2704](https://github.com/ThinkFlowLab/FlyBot/issues/2704). diff --git a/docs/subsystems/workflow.md b/docs/subsystems/workflow.md index c35adfe50..5e3a16371 100644 --- a/docs/subsystems/workflow.md +++ b/docs/subsystems/workflow.md @@ -59,7 +59,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). A wire or a reconnect that would list a pair the graph already carries commits nothing new, because `normalizeGraph` refuses a graph that lists one pair twice and then saves none of it: unguarded, the plane would show no edge, no change and only a save error (#2163). 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). While another dsh holds the Bot team the host is read-only (`roster.readOnly`): the page still reads through `workflow/state`, keeps its controls disabled, and offers no **Edit**, **Add link** or wire port, so it cannot take a change that core would refuse with `bot/read-only` and then drop (#2187). The same window draws the sidebar's workflow list without **New**, the row's **Rename**/**Delete** or **Hand a task to a Workflow**: the list is read-only too, and an `roster.readOnly` that is not a string, or holds only whitespace, counts as writable — trimmed the way `core/src/client/workflows.js:570` trims it (#2200). The Stage a row opens carries the same lock: it prints the host's sentence under `role="status"` and draws no **Hand a task to a Workflow**, no **Re-match**, no **Pause**/**Resume**/**Stop**, no **Edit**/drag palette and no **Reset**, while the row click, **Progress**, **Flow** and viewing a run still work; a writable roster draws every control exactly as before (#2213). Mounting the list in a read-only window still reads `projects` so the Project filter fills, but it sends no `workflow/ingest` for the open cards: ingest is a claim/match write the host refuses, so a reader opening the list writes nothing; a writable window still ingests every open card (#2222). The read-only sentence can also land after the list has mounted — a cold window starts on the empty roster, `ready: false` and no `readOnly` — so the ingest step asks the value the latest render saw rather than the one the mount saw, and it skips the board read and stops between two cards once the window has turned read-only. The Settings page draws no read-only sentence of its own: the host's Settings header already prints one for every page it shows (`core/src/client/settings.js` `bt-read-only`, `role="status"`, `hostText(roster.readOnly)`), and a second copy inside the page said the same thing twice (#2202). -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, #1546, #1570, #1571):** When a node completes **or fails**, the run ends immediately with status `done`, sets `run.earlyEnd` to the condition name, and does not advance to successors, take a success-path `loop`, or burn `onFail` recovery attempts. The two markers that report the task itself end it only on a structured signal (#1546): `needs-human` and `already-claimed` must be a declared OUTPUT whose value **equals** the marker, or a whole reply line that is exactly `EARLY-END ` in any case. A step that quotes an issue title, negates the label or reports it absent keeps running. `tip-unchanged` (#1570) and `open-pr` (#1571) accept that same structured signal — in a declared OUTPUT the value counts as a whole word, so a brief that names `data-open-project` is not `open-pr` (#1866) — and also honor their prose fallbacks (`tip unchanged` / `unchanged tip`, `open PR already` / `already has an open PR` / `has an open PR` / `found an open PR`), so a Bot that forgets the structured marker cannot soft-lock Claim↔Verify on the same tip. See `checkEarlyEnd` in Key files. Factory prompt guidance lives in `EARLY_END_GUIDANCE` (`plugins/workflow/src/template.js`) for Settings graphs that wire Triage / Claim / Verify: Verify must escalate once on an unchanged tip, not re-run the full suite. There is no scripted or seeded run path: `workflow/state` exposes only `live === true` runs — the running ones, plus the last run when it is still `live` and ended `failed` — a failed sibling stays on the stage beside whatever is running now, with `run.failure` (`{ code, at, nodeName?, text? }`) naming what ended it, so the room does not have to open Runs to learn that a run failed (#1493, #1494). The failed sibling is left out when a running run carries the same task id: that is a retry, and the live run keeps the row (#1507). History entries carry the run `id` and the same `failure`, so a Runs row shows when the run ended and why, and a row for a run already on the stage is not listed twice. 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 **read-only** host — a second window on the same `DSH_HOME`, the reader `home-lock` describes — does neither: it never ran those runs, so it cannot tell which of them a dead host left behind, and it must not end them, release their step Bots or write its older copy over the writer's store; both calls return without touching the record and the writer's own `reap` decides (#2673). 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). A graph edit that drops a node that is still running ends that run the way Stop does: the node is failed, the run is archived `failed`, its node Bots are released, and the queue drains (#1549); an id whose release failed stays named on that archived run, and the next load's `reap` retries it, instead of the edit forgetting it (the #1616 contract). 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 live step also holds a deadline — its workflow's `stepTimeoutMinutes`, `DEFAULT_STEP_TIMEOUT_MINUTES` (45 minutes) unless the record sets one, an integer from 1 to 120 given at create, through `define_workflow`, or through `workflow/step-timeout` (armed even when the host cannot spawn ephemeral Bots; #1530, #1580); when it expires the node and the run fail as `failed`, the run's ephemeral Bots are released, the chat and `run.failure` say the step's deadline was reached (`step deadline reached (45 min)`, the number its lane set), and `warn` names it once; the line never reads as a reply that did not arrive, because a step still working is killed by the same timer (#1868). The deadline is cleared as soon as the step moves, and by Pause, Reset, delete, `reap`, `dispose` and a Stop that released every Bot; a run the owner paused 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 fails that node on a structured signal only: the host's failure notice carries `failure: true`, which the middleware hands `stepResult` as `ok: false` (a caller-set `ok: false` counts too), so an answer that quotes `could not answer your message` or `model call failed` while auditing it advances the step (#2677). An old host without the marker falls back to the notice shape at the start of the text only (` could not answer your message: …`), never a free substring, and a 429 or rate-limit mention in any other reply does not fail a step. 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. It also keeps the deadline of the node it was on — the one paused state that holds one — and a graph save re-syncs timers without taking it away (a save that cleared it left the run holding its slot with nothing left to free it); the deadline releases again there, ends the run as `paused-expired` and frees the slot, re-arming itself while a Bot still will not go (#2453, P-560). 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. +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, #1546, #1570, #1571):** When a node completes **or fails**, the run ends immediately with status `done`, sets `run.earlyEnd` to the condition name, and does not advance to successors, take a success-path `loop`, or burn `onFail` recovery attempts. The two markers that report the task itself end it only on a structured signal (#1546): `needs-human` and `already-claimed` must be a declared OUTPUT whose value **equals** the marker, or a whole reply line that is exactly `EARLY-END ` in any case. A step that quotes an issue title, negates the label or reports it absent keeps running. `tip-unchanged` (#1570) and `open-pr` (#1571) accept that same structured signal — in a declared OUTPUT the value counts as a whole word, so a brief that names `data-open-project` is not `open-pr` (#1866) — and also honor their prose fallbacks (`tip unchanged` / `unchanged tip`, `open PR already` / `already has an open PR` / `has an open PR` / `found an open PR`), so a Bot that forgets the structured marker cannot soft-lock Claim↔Verify on the same tip. See `checkEarlyEnd` in Key files. Factory prompt guidance lives in `EARLY_END_GUIDANCE` (`plugins/workflow/src/template.js`) for Settings graphs that wire Triage / Claim / Verify: Verify must escalate once on an unchanged tip, not re-run the full suite. There is no scripted or seeded run path: `workflow/state` exposes only `live === true` runs — the running ones, plus the last run when it is still `live` and ended `failed` — a failed sibling stays on the stage beside whatever is running now, with `run.failure` (`{ code, at, nodeName?, text? }`) naming what ended it, so the room does not have to open Runs to learn that a run failed (#1493, #1494). The failed sibling is left out when a running run carries the same task id: that is a retry, and the live run keeps the row (#1507). History entries carry the run `id` and the same `failure`, so a Runs row shows when the run ended and why, and a row for a run already on the stage is not listed twice. 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 **read-only** host — a second window on the same `DSH_HOME`, the reader `home-lock` describes — does neither: it never ran those runs, so it cannot tell which of them a dead host left behind, and it must not end them, release their step Bots or write its older copy over the writer's store; both calls return without touching the record and the writer's own `reap` decides (#2673). **Taking the lock over is a state change (#2704):** the first call that finds the host writable after this window started read-only — a read, `reap`, `dispose` or any write — re-reads `workflows.json` once before it changes anything, because the copy it loaded predates what the dead window saved, and then runs the same load-time `reap` once on that adopted store. That reap ends the dead window's `running` run as `host-ended` (or heals a `live:true` run whose Bot map is already gone, #1414) and releases its step Bots; `lane.queue` is never touched, so a card that was merely `queued` keeps its place and starts when the lane drains. Only the first such call re-reads — a writer that later loses the lock is still refused, exactly as before. 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). A graph edit that drops a node that is still running ends that run the way Stop does: the node is failed, the run is archived `failed`, its node Bots are released, and the queue drains (#1549); an id whose release failed stays named on that archived run, and the next load's `reap` retries it, instead of the edit forgetting it (the #1616 contract). 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 live step also holds a deadline — its workflow's `stepTimeoutMinutes`, `DEFAULT_STEP_TIMEOUT_MINUTES` (45 minutes) unless the record sets one, an integer from 1 to 120 given at create, through `define_workflow`, or through `workflow/step-timeout` (armed even when the host cannot spawn ephemeral Bots; #1530, #1580); when it expires the node and the run fail as `failed`, the run's ephemeral Bots are released, the chat and `run.failure` say the step's deadline was reached (`step deadline reached (45 min)`, the number its lane set), and `warn` names it once; the line never reads as a reply that did not arrive, because a step still working is killed by the same timer (#1868). The deadline is cleared as soon as the step moves, and by Pause, Reset, delete, `reap`, `dispose` and a Stop that released every Bot; a run the owner paused 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 fails that node on a structured signal only: the host's failure notice carries `failure: true`, which the middleware hands `stepResult` as `ok: false` (a caller-set `ok: false` counts too), so an answer that quotes `could not answer your message` or `model call failed` while auditing it advances the step (#2677). An old host without the marker falls back to the notice shape at the start of the text only (` could not answer your message: …`), never a free substring, and a 429 or rate-limit mention in any other reply does not fail a step. 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. It also keeps the deadline of the node it was on — the one paused state that holds one — and a graph save re-syncs timers without taking it away (a save that cleared it left the run holding its slot with nothing left to free it); the deadline releases again there, ends the run as `paused-expired` and frees the slot, re-arming itself while a Bot still will not go (#2453, P-560). 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 with its own `status` (never a hardcoded Running), the chip count does not exceed that list, and a failed sibling appears with its time and reason. On the Runs list, the clock stays in a non-shrinking column with a gap before the status chip, and a Failed row paints the Failed chip once — a generic Failed with no host detail does not also draw `wf-hist-reason` (#1618). 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 : `. It also lists the OUTPUT lines the step owes — its declared ports plus every output its outgoing edges read, whether a condition's `when` or a loop's `until` (`edgeOutputs` in `src/graph.js`) — spelled `OUTPUT : ` when an edge compares that output against a value — so a node that declares no outputs is still told which line its own gate reads (#2669, #2685). The Bot reports `OUTPUT : `; a single declared output may be the reply body. A declared output a gate reads is required (§3): when a `when`/`else` or `until` edge reads it and the step's reply does not produce it, the step fails with `failure.text` `Missing OUTPUT ` instead of settling, so the gate never reads silence as "not this branch" and the run cannot close `done` on it; a delivered value that does not equal `eq` still completes the step and skips the untaken branch (#1773). One exception keeps the work a step already did (#2583, #2658): when `state` is the only gate-read output missing and the step's own reply shows its result — a pull request link, `pull/`, a line saying it opened PR #n, a `Claimed-by:` / `Claimed #n` line, or its own declared OUTPUT names the delivery (`pr`/`pull` whose value is only a pull number — a positive integer, optional leading `#`, optional trailing parenthetical such as `2657 (verified)` — or `merged`/`merge-commit` carrying a 40-character commit sha; any other output name counts for nothing) — the step stands as `done`, the run takes the node's first declared `state` branch (on the fast line `state: ready → claim`, so the pull request reaches gate/enqueue instead of the `else` route the step never chose), and the run record carries `outputMissing: true` and `producedPull: ` when a pull request was named; every other missing verdict still fails, and a step whose reply shows no such result still fails the same way. `stepArtifacts` reads that evidence and `firstBranchValue` names the branch, both in `src/graph.js`; the engine has no GitHub client, so the reply is the only evidence it can read. 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/service.js b/plugins/workflow/src/service.js index b2b510edd..ea514dfbe 100644 --- a/plugins/workflow/src/service.js +++ b/plugins/workflow/src/service.js @@ -221,22 +221,68 @@ export function createWorkflowService({ setTimer = (fn, ms) => setTimeout(fn, ms), clearTimer = handle => clearTimeout(handle), } = {}) { const file = dir ? join(dir, FILE) : null - const loaded = load(file, warn) - let state = loaded.state ?? seed() + let state // #2603: a store this service could not read stays a failure instead of turning // into a clean list. `state()` answers it and the first `write()` keeps the bytes // before replacing them; every other endpoint keeps running on the fixture. - let loadError = loaded.error + let loadError // #2681: a store a newer build wrote is the one that must not be replaced at all — its // version travels here so `state()` can name it and `write()` can refuse before the first // byte (a `.broken-` copy is only for a store this build is about to write over). - let loadNewer = loaded.newer ?? null - state.routes ??= {} - // migrateRuns() also normalizes the orphan list a deleted workflow left - // (#1621), so a store written by an older version keeps those ids. - migrateRuns(state) - purgeDemos(state, historyRow) - let seq = state.seq ?? 1 + let loadNewer = null + let seq = 1 + // #2704: whether this process has held the host's lock since it loaded the store. + // A window that starts behind another writer reads that writer's copy, and the first + // call that finds the host writable has to re-read the file before it changes anything. + let writer = !readOnly() + let takingOver = null + + // Read the store whole, the way core's store.refresh() adopts state.json. A window + // that was read-only carries no change of its own to keep, so replacing the copy + // loses nothing; a store it still cannot read keeps the fixture and the reason. + function adoptStore() { + const loaded = load(file, warn) + loadError = loaded.error + loadNewer = loaded.newer ?? null + state = loaded.state ?? seed() + state.routes ??= {} + // migrateRuns() also normalizes the orphan list a deleted workflow left + // (#1621), so a store written by an older version keeps those ids. + migrateRuns(state) + purgeDemos(state, historyRow) + seq = state.seq ?? 1 + } + adoptStore() + + // #2704: becoming the writer is a state change, not a flag flip. The host takes + // `state.lock` before it answers a write (core/src/host/store.js), so the first call + // that finds it writable re-reads `workflows.json` once and reaps the runs the dead + // window left `running` — the same cleanup a load does (#1388, #2673). Only the first + // such call: after it this process is the writer and its copy is current. + function takeover() { + adoptStore() + // Another host may have taken the lock again while the file was read; writing this + // copy after that would be the same clobber. + if (readOnly()) return false + writer = true + try { + takingOver = Promise.resolve(reapRuns()) + } catch (error) { + takingOver = Promise.resolve() + warn(`flybot-workflow: the reap after takeover failed: ${error?.message ?? error}`) + } + takingOver.catch(error => warn(`flybot-workflow: the reap after takeover failed: ${error?.message ?? error}`)) + return true + } + + // Every write path asks this instead of `readOnly()`: false still means the host is a + // reader, and a writer that lost the lock is refused the same way it was before. The + // one path that changes is a window that started read-only and now holds the lock. + function writable() { + if (readOnly()) return false + if (!writer) return takeover() + return true + } // One lock for every start and release to serialize drain operations. let hostTail = Promise.resolve() function hostSerial(fn) { @@ -400,6 +446,17 @@ export function createWorkflowService({ function write() { if (!persist || !file) return + // #2704: a write from a window that has not taken the store over would put the copy + // it read while the other writer held it back over that writer's newer records. + // Every entry point calls writable() first; this catches one that does not. + if (!writer) { + throw new Error(`Not saved: this window has not taken the workflow store ${file} over since the other window let go, so it would write over that window's records.`) + } + // The lock can move to another host between the takeover and this write; a reader + // must not land its copy then either (#2673). + if (readOnly()) { + throw new Error(`Not saved: the host became read-only before the workflow store ${file} was written.`) + } // #2681: a newer build's store is refused before the first byte, not backed up and // replaced. This build cannot read every field of it, so writing would drop them while // the file kept claiming the newer version — the same damage the trace store refuses. @@ -1301,7 +1358,7 @@ export function createWorkflowService({ } function mutate(workflowId, fn) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } const item = workflow(workflowId) if (!item) return { ok: false, error: 'No such workflow' } try { @@ -1313,8 +1370,86 @@ export function createWorkflowService({ } } + // #2704: the reaper a load runs, reusable when a read-only window becomes the writer. + // Only runs the dead writer was holding are collected: a run on `active` (or still the + // archived `last`) that is `running` with step Bots named, or a `live:true` run already + // marked `running` whose Bot map is gone (#1414). An ended run keeps its own status + // (#1388) and only gives back the Bots it still names. `lane.queue` is never touched: + // a parked card is not a run, so a takeover leaves it waiting for the next drain. + function reapRuns() { + // A read-only host never ran these runs, so it cannot tell which of them the + // writer still holds; judging them here would end the writer's live run and + // overwrite the store with this host's older copy (#2673). + if (readOnly()) return Promise.resolve() + if (typeof release !== 'function') return Promise.resolve() + const stale = [] + const staleRuns = [] + const abandoned = [] + const orphans = orphanIds(state) + let wrote = false + for (const [workflowId, lane] of Object.entries(state.lanes ?? {})) { + const seen = new Set() + for (const run of [...(lane.active ?? []), lane.last].filter(Boolean)) { + if (seen.has(run)) continue + seen.add(run) + const hasEphemeral = run?.ephemeral && Object.keys(run.ephemeral).length > 0 + const zombie = run?.live === true && run.status === 'running' && !hasEphemeral + if (!hasEphemeral && !zombie) continue + // A run still marked running here was cut off with the host that ran it, + // and nothing resumes it: its instances are abandoned, not in progress. + // Release them and end the run, or they stay hidden with no way to remove + // them. A run that already ended keeps its own status (#1388). A zombie + // is a run dispose() already stripped: heal it on load (#1414). + if (run.status === 'running') { + abandoned.push({ workflowId, run }) + continue + } + if (hasEphemeral) { + clearRunTimers(run.id) + stale.push(...Object.values(run.ephemeral)) + staleRuns.push(run) + delete run.handed + wrote = true + } + } + } + for (const { workflowId, run } of abandoned) { + try { + clearRunTimers(run.id) + if (run.ephemeral) stale.push(...Object.values(run.ephemeral)) + endRun(workflow(workflowId), run, { code: 'host-ended' }) + push(workflowId, [failureLine(run)], now()) + staleRuns.push(run) + delete run.handed + archiveCutoffRun(workflowId, run) + wrote = true + } catch (error) { + warn(`flybot-workflow: could not end ${run.id}: ${error?.message ?? error}`) + } + } + if (stale.length === 0 && orphans.length === 0 && !wrote) return Promise.resolve() + if (stale.length === 0 && orphans.length === 0) { + write() + return Promise.resolve() + } + return releaseAll([...new Set([...stale, ...orphans])], 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) + // #1621: a deleted workflow's Bot has no lane left to name it, so it is + // the orphan list that keeps a failed id — the same entry, retried until + // the host really removes it. + state.orphans = orphans.filter(id => failed.includes(id)) + if (state.orphans.length === 0) delete state.orphans + write() + }) + } + return { state(body) { + // #2704: a window that has just taken the lock adopts the store and reaps before + // it answers, so the page does not keep showing the dead window's run as live. + writable() // #2603: a store that could not be read is not answered as a list nothing in it // came from. The factory templates are not the store, so `workflow/templates` // still gets them from here. #2681: a newer store says which it is, so the page can @@ -1325,7 +1460,7 @@ export function createWorkflowService({ return present(heldLimitOf(body)) }, rename(body = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } const workflowId = String(body.workflowId ?? '') const item = workflow(workflowId) if (!item) return { ok: false, error: 'No such workflow' } @@ -1354,7 +1489,7 @@ export function createWorkflowService({ } }, delete(body = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } const workflowId = String(body.workflowId ?? '') const item = workflow(workflowId) if (!item) return { ok: false, error: 'No such workflow' } @@ -1421,7 +1556,7 @@ export function createWorkflowService({ })() }, create(body = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } // #2598: `write()` refuses before its first byte when the temp leaf is not // writable, but the caller is told nothing was created. The push and its // creation line below are memory, so remember what they change and undo @@ -1503,7 +1638,7 @@ export function createWorkflowService({ // title is only the label (and its fallback). A field this endpoint does // not read comes back in `ignored` instead of vanishing (#1869). assign(body = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } const ignored = Object.keys(body).filter(key => !ASSIGN_FIELDS.has(key)) const answer = value => (ignored.length === 0 ? value : { ...value, ignored }) if (body.botId) return answer(routeBot(body)) @@ -1582,7 +1717,7 @@ export function createWorkflowService({ return saved }, setModel(body = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } const workflowId = String(body.workflowId ?? '') const model = body.model ? cleanLine(body.model, 80) : null const saved = mutate(workflowId, (item) => { @@ -1592,7 +1727,7 @@ export function createWorkflowService({ return present() }, advance(body = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) 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 ?? '') @@ -1657,7 +1792,7 @@ export function createWorkflowService({ return findEphemeral(botId) !== null }, pause(body = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } const workflowId = String(body.workflowId ?? '') const item = workflow(workflowId) if (!item) return { ok: false, error: 'No such workflow' } @@ -1679,7 +1814,7 @@ export function createWorkflowService({ return present() }, resume(body = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } const workflowId = String(body.workflowId ?? '') const item = workflow(workflowId) if (!item) return { ok: false, error: 'No such workflow' } @@ -1710,7 +1845,7 @@ export function createWorkflowService({ }) }, reset(body = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } const workflowId = String(body.workflowId ?? '') const runId = body.runId ? String(body.runId) : '' // Release first, then forget the ids (#1476, #1387). A failed release is @@ -2002,7 +2137,7 @@ export function createWorkflowService({ // keeps it too, and nothing is written. Passing `rematch: true` drops that one // claim first — the panel's per-task action. ingest(task = {}) { - if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' } + if (!writable()) return { ok: false, code: 'read-only', error: 'The host is read-only' } const taskId = cleanLine(task.id, 80) if (!taskId) return { ok: false, error: 'A task needs an id' } if (task.rematch === true && forgetClaim(taskId)) write() @@ -2107,75 +2242,25 @@ export function createWorkflowService({ return rows }, reap() { - // A read-only host never ran these runs, so it cannot tell which of them the - // writer still holds; judging them here would end the writer's live run and - // overwrite the store with this host's older copy (#2673). - if (readOnly()) return Promise.resolve() - if (typeof release !== 'function') return Promise.resolve() - const stale = [] - const staleRuns = [] - const abandoned = [] - const orphans = orphanIds(state) - let wrote = false - for (const [workflowId, lane] of Object.entries(state.lanes ?? {})) { - const seen = new Set() - for (const run of [...(lane.active ?? []), lane.last].filter(Boolean)) { - if (seen.has(run)) continue - seen.add(run) - const hasEphemeral = run?.ephemeral && Object.keys(run.ephemeral).length > 0 - const zombie = run?.live === true && run.status === 'running' && !hasEphemeral - if (!hasEphemeral && !zombie) continue - // A run still marked running here was cut off with the host that ran it, - // and nothing resumes it: its instances are abandoned, not in progress. - // Release them and end the run, or they stay hidden with no way to remove - // them. A run that already ended keeps its own status (#1388). A zombie - // is a run dispose() already stripped: heal it on load (#1414). - if (run.status === 'running') { - abandoned.push({ workflowId, run }) - continue - } - if (hasEphemeral) { - clearRunTimers(run.id) - stale.push(...Object.values(run.ephemeral)) - staleRuns.push(run) - delete run.handed - wrote = true - } - } + if (!writable()) return Promise.resolve() + // #2704: the call that noticed the takeover already started the same reaper a load + // runs; hand that one back instead of releasing the dead window's Bots twice. + if (takingOver !== null) { + const pending = takingOver + takingOver = null + return pending } - for (const { workflowId, run } of abandoned) { - try { - clearRunTimers(run.id) - if (run.ephemeral) stale.push(...Object.values(run.ephemeral)) - endRun(workflow(workflowId), run, { code: 'host-ended' }) - push(workflowId, [failureLine(run)], now()) - staleRuns.push(run) - delete run.handed - archiveCutoffRun(workflowId, run) - wrote = true - } catch (error) { - warn(`flybot-workflow: could not end ${run.id}: ${error?.message ?? error}`) - } - } - if (stale.length === 0 && orphans.length === 0 && !wrote) return Promise.resolve() - if (stale.length === 0 && orphans.length === 0) { - write() - return Promise.resolve() - } - return releaseAll([...new Set([...stale, ...orphans])], 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) - // #1621: a deleted workflow's Bot has no lane left to name it, so it is - // the orphan list that keeps a failed id — the same entry, retried until - // the host really removes it. - state.orphans = orphans.filter(id => failed.includes(id)) - if (state.orphans.length === 0) delete state.orphans - write() - }) + return reapRuns() }, dispose() { for (const key of [...timers.keys()]) clearTimerAt(key) + // #2704: a window that is only now taking the store over has no run of its own to + // end — the runs on disk belong to the writer that died — so the takeover reap is + // the cleanup. Nothing here would be needed on top of it. + if (!writer) { + if (!writable()) return Promise.resolve() + return takingOver ?? Promise.resolve() + } // A read-only host must not end the writer's runs, release its step Bots or // overwrite the store with its own older copy (#2673). Only local timers are // cleared; ending belongs to the writer's own dispose or the next reap. diff --git a/plugins/workflow/test/takeover.test.js b/plugins/workflow/test/takeover.test.js new file mode 100644 index 000000000..46665359b --- /dev/null +++ b/plugins/workflow/test/takeover.test.js @@ -0,0 +1,246 @@ +// #2704: a second window on one home starts read-only because the first one holds +// `state.lock`. When that window dies the second takes the lock, and until now it kept +// the copy of `workflows.json` it loaded at construction and never ran its reaper again: +// its first write put the stale copy back over what the dead window had saved, and the +// dead window's `running` run held the lane for good. +// +// Becoming the writer is a state change: the first call that finds the host writable +// re-reads the store once (the way core's store.mutate() refreshes before a change) and +// reaps the runs the dead window left. These tests drive that with real child processes, +// a real `state.lock` and a real `kill -9`; neither the lock nor the store is mocked. +import assert from 'node:assert/strict' +import { spawn } from 'node:child_process' +import { mkdtempSync, readFileSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { fileURLToPath } from 'node:url' + +const REPO = fileURLToPath(new URL('../../..', import.meta.url)) +const SERVICE = `${REPO}plugins/workflow/src/service.js` +const LOCK = `${REPO}core/src/host/home-lock.js` +const J = JSON.stringify +const NOW = 1_700_000_000_000 +const DELIVERY = 'delivery' + +const live = [] + +// A real node child: the script is the module graph under test, argv carries its home. +function start(code, args = []) { + const proc = spawn(process.execPath, ['--input-type=module', '-e', code, ...args], { stdio: ['pipe', 'pipe', 'pipe'] }) + live.push(proc) + let out = '' + let err = '' + const waiters = [] + const sweep = () => { + for (let index = 0; index < waiters.length;) { + const waiter = waiters[index] + if (!waiter.re.test(out)) { index += 1; continue } + waiters.splice(index, 1) + waiter.resolve(out) + } + } + proc.stdout.setEncoding('utf8') + proc.stderr.setEncoding('utf8') + proc.stdout.on('data', chunk => { out += chunk; sweep() }) + proc.stderr.on('data', chunk => { err += chunk }) + const exited = new Promise(resolve => proc.on('exit', (code, signal) => resolve({ code, signal }))) + return { + proc, + get output() { return out }, + get errors() { return err }, + exited, + waitFor(re, timeoutMs = 15_000) { + if (re.test(out)) return Promise.resolve(out) + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(`timeout waiting for ${re}\nstdout: ${out}\nstderr: ${err}`)), timeoutMs) + waiters.push({ re, resolve: value => { clearTimeout(timer); resolve(value) } }) + }) + }, + send() { proc.stdin.write('go\n') }, + } +} + +const readStore = (dir) => JSON.parse(readFileSync(join(dir, 'workflows.json'), 'utf8')) +const laneOf = (dir, workflowId = DELIVERY) => readStore(dir).lanes?.[workflowId] ?? {} +const taskIds = (dir) => (laneOf(dir).active ?? []).map(run => run.task?.id) + +test.after(() => { + for (const proc of live) if (proc.exitCode === null && proc.signalCode === null) proc.kill('SIGKILL') +}) + +const lockPrelude = ` + import { join } from 'node:path' + import { homeLock } from ${J(LOCK)} + import { createWorkflowService } from ${J(SERVICE)} +` +const readOnlyByLock = 'readOnly: () => (lock.owns() ? undefined : \'read-only\'),' + +// --------------------------------------------------------------------------- +// F1. The second window loads while the first holds state.lock, so it holds an older +// copy of the store. The first window records another task and is killed with -9; the +// second takes the lock and writes. Its first write has to land on the file the dead +// window left, not on the copy it loaded before that record existed. +test('F1 a window that takes over re-reads the store before its first write (#2704)', { timeout: 30_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-takeover-f1-')) + try { + const writer = `${lockPrelude} + const dir = process.argv[1] + const lock = homeLock(join(dir, 'state.lock')) + if (lock.claim() !== undefined) { console.error('lock not free'); process.exit(2) } + const svc = createWorkflowService({ dir, persist: true, now: () => ${NOW}, bots: () => [], ${readOnlyByLock} }) + svc.setConcurrency({ workflowId: '${DELIVERY}', maxConcurrent: 2 }) + svc.assign({ workflowId: '${DELIVERY}', taskId: 'before', title: 'before' }) + process.stdout.write('@@ready\\n') + process.stdin.once('data', () => { + svc.assign({ workflowId: '${DELIVERY}', taskId: 'after', title: 'after' }) + process.stdout.write('@@after\\n') + }) + setInterval(() => {}, 1000) + ` + const second = `${lockPrelude} + const dir = process.argv[1] + const lock = homeLock(join(dir, 'state.lock')) + lock.claim() + const svc = createWorkflowService({ dir, persist: true, now: () => ${NOW}, bots: () => [], ${readOnlyByLock} }) + const current = () => svc.state().workflows.find(item => item.id === '${DELIVERY}')?.run?.task?.id ?? null + process.stdout.write('@@loaded ' + JSON.stringify(current()) + '\\n') + // A reader refuses every write and leaves the writer's file alone (#2673). + const refused = svc.assign({ workflowId: '${DELIVERY}', taskId: 'stale-write', title: 'stale' }) + process.stdout.write('@@refused ' + JSON.stringify({ ok: refused.ok, code: refused.code }) + '\\n') + process.stdin.once('data', () => { + if (lock.claim() !== undefined) { console.error('takeover failed'); process.exit(3) } + svc.setConcurrency({ workflowId: '${DELIVERY}', maxConcurrent: 1 }) + process.stdout.write('@@wrote\\n') + process.exit(0) + }) + ` + + const first = start(writer, [dir]) + await first.waitFor(/@@ready/) + const other = start(second, [dir]) + // It opens on the copy without the record the writer is about to add. + assert.match(await other.waitFor(/@@loaded/), /@@loaded "before"/) + assert.match(await other.waitFor(/@@refused/), /@@refused \{"ok":false,"code":"read-only"\}/) + assert.deepEqual(taskIds(dir), ['before'], 'a read-only window wrote over the writer') + + first.send() + await first.waitFor(/@@after/) + assert.deepEqual([...taskIds(dir)].sort(), ['after', 'before']) + first.proc.kill('SIGKILL') + await first.exited + + other.send() + await other.waitFor(/@@wrote/) + await other.exited + + // The takeover reloaded before it wrote: the record the dead window saved is still + // there. Without the reload this is ['before'] — the case #2704 reports. + assert.deepEqual([...taskIds(dir)].sort(), ['after', 'before'], 'the takeover dropped the record the dead window wrote after this window loaded') + const restarted = readStore(dir) + const persisted = (restarted.lanes[DELIVERY].active ?? []).map(run => run.task?.id) + assert.deepEqual([...persisted].sort(), ['after', 'before'], 'the restart does not read both records') + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +// --------------------------------------------------------------------------- +// F2. The reaper ran once at load, and a read-only host returns from it without touching +// the record (#2673). After the takeover it must run once more: the run the dead window +// left `running` is ended `host-ended`, its step Bots are released exactly once, and a +// card that was merely `queued` is not killed — it keeps its place and starts. +test('F2 a takeover reaps the run the dead window left running, and leaves queued cards (#2704)', { timeout: 30_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-takeover-f2-')) + try { + const writer = `${lockPrelude} + const dir = process.argv[1] + const lock = homeLock(join(dir, 'state.lock')) + if (lock.claim() !== undefined) { console.error('lock not free'); process.exit(2) } + const svc = createWorkflowService({ + dir, persist: true, now: () => ${NOW}, bots: () => [], dispatch: async () => ({ ok: true }), + spawn: async (spec) => 'eph-' + spec.nodeId, release: async () => {}, ${readOnlyByLock} + }) + const made = svc.define({ + name: 'Lane', + nodes: [{ id: 'plan', name: 'Planner', prompt: 'Plan.' }, { id: 'code', name: 'Coder', prompt: 'Code.' }], + edges: [{ from: 'plan', to: 'code' }], + }).workflows.find(item => item.name === 'Lane') + await svc.setConcurrency({ workflowId: made.id, maxConcurrent: 1 }) + await svc.runNamed({ workflow: 'Lane', brief: 'dead run', task: 'dead-running' }) + await svc.assign({ workflowId: made.id, taskId: 'dead-queued', title: 'dead queued' }) + const item = svc.state().workflows.find(row => row.id === made.id) + process.stdout.write('@@started ' + JSON.stringify({ + id: made.id, + task: item.run?.task?.id ?? null, + status: item.run?.status ?? null, + active: item.activeCount, + queue: item.queueDepth, + }) + '\\n') + setInterval(() => {}, 1000) + ` + const second = `${lockPrelude} + const dir = process.argv[1] + const workflowId = process.argv[2] + const lock = homeLock(join(dir, 'state.lock')) + lock.claim() // the other window is alive: this one is a reader + const released = [] + const svc = createWorkflowService({ + dir, persist: true, now: () => ${NOW}, bots: () => [], dispatch: async () => ({ ok: true }), + spawn: async (spec) => 'eph-second-' + spec.nodeId, release: async (id) => { released.push(id) }, ${readOnlyByLock} + }) + const view = () => { + const item = svc.state().workflows.find(row => row.id === workflowId) + return { + task: item.run?.task?.id ?? null, + status: item.run?.status ?? null, + active: item.activeCount, + queue: item.queueDepth, + queued: item.queue.map(card => card.id), + } + } + await svc.reap() // what loading does: read-only, so nothing happens + process.stdout.write('@@startup ' + JSON.stringify({ view: view(), released }) + '\\n') + process.stdin.once('data', async () => { + if (lock.claim() !== undefined) { console.error('takeover failed'); process.exit(3) } + const assigned = svc.assign({ workflowId, taskId: 'new-card', title: 'new' }) + if (typeof assigned?.then === 'function') await assigned + process.stdout.write('@@after ' + JSON.stringify({ view: view(), released: [...released].sort() }) + '\\n') + process.exit(0) + }) + ` + + const first = start(writer, [dir]) + const started = JSON.parse((await first.waitFor(/@@started/)).match(/@@started (.*)/)[1]) + assert.equal(started.status, 'running') + assert.equal(started.active, 1) + assert.equal(started.queue, 1) + + const other = start(second, [dir, started.id]) + const startup = JSON.parse((await other.waitFor(/@@startup/)).match(/@@startup (.*)/)[1]) + // The reader opened on the dead window's run and does not reap it (#2673). + assert.deepEqual(startup.view, { task: 'dead-running', status: 'running', active: 1, queue: 1, queued: ['dead-queued'] }) + assert.deepEqual(startup.released, [], "a read-only host released the writer's step Bots") + first.proc.kill('SIGKILL') + await first.exited + + other.send() + const after = JSON.parse((await other.waitFor(/@@after/)).match(/@@after (.*)/)[1]) + await other.exited + + // The dead run is over and its Bots are released exactly once, so the lane moved: + // the card that was already queued starts, and only the card this window added waits. + assert.deepEqual(after.released, ['eph-code', 'eph-plan'], "the takeover did not release the dead run's step Bots once") + assert.deepEqual(after.view, { task: 'dead-queued', status: 'running', active: 1, queue: 1, queued: ['new-card'] }) + + const lane = laneOf(dir, started.id) + assert.deepEqual((lane.active ?? []).map(run => run.task?.id), ['dead-queued'], 'the queued card was killed or the dead run still holds a slot') + assert.deepEqual((lane.queue ?? []).map(card => card.id), ['new-card']) + assert.equal(lane.last?.task?.id, 'dead-running') + assert.equal(lane.last?.status, 'failed') + assert.equal(lane.last?.failure?.code, 'host-ended') + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) +