diff --git a/changelog.d/2787-takeover-drain.fixed.md b/changelog.d/2787-takeover-drain.fixed.md new file mode 100644 index 000000000..2f16e71cc --- /dev/null +++ b/changelog.d/2787-takeover-drain.fixed.md @@ -0,0 +1 @@ +A card queued behind a run ended by the reap a window runs when it **takes the store over** now starts on its own: the takeover reap drains the lane's queue like the load-time reap, instead of leaving the card parked with a free slot until some unrelated assign or ingest. See [#2787](https://github.com/ThinkFlowLab/FlyBot/issues/2787). diff --git a/docs/subsystems/workflow.md b/docs/subsystems/workflow.md index b7e7c24d5..bc0420b94 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?, attempts? }`) 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, and that ending drains the lane's queue like any other terminal transition, so a card parked behind the dead run starts on load instead of waiting for the next assign (#2752); that drain joins the serial tail the assigns use, because the load does not await the reap and a drain outside it would start a second card beside the assign at the lane's cap (#2757). 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). The deadline is not queued behind the wait it bounds (#2748): an expiry runs off the serial tail, so a host call that never answers — a dispatch, a release, a spawn — cannot park the one timer that ends the run, and the run still reaches `failed` with its Bots released. A start whose Bots the host does not create within `SPAWN_DEADLINE_MS` (60 s) is refused, and the card a drain had already taken out of `lane.queue` goes back to it instead of vanishing (a late spawn's Bots are released whether that answer is a success or a failure: a failed one carries the ids it already made on `error.spawned`, which the deadline branch now releases, #2785). 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 shapes at the start of the text only (` could not answer your message: …`, ` was stopped before it could answer your message.`, and the bare `could not answer your message` / `The model call failed.` an old host wrote), never a free substring, so an ordinary answer that merely ends in the model-failure wording keeps advancing the step (#2741). A stopped turn is a failed step, with the stop notice as its reason, instead of finishing as done and handing that notice to the next node (#2741). 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 once the Bot finally goes; while it still will not, that retry is bounded — `STOP_RELEASE_RETRIES` (two) more deadlines — and the last one ends the run `failed` as `release-failed` with the ids the host would not remove and the attempt count on the row, so the slot comes back and the queue drains (#2453, P-560, #2750). 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?, attempts? }`) 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; the reap itself leaves `lane.queue` alone, so a card that was merely `queued` keeps its place, and the drain that follows the reap starts it (#2787). 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, and that ending drains the lane's queue like any other terminal transition, so a card parked behind the dead run starts on load instead of waiting for the next assign (#2752) — a takeover reap starts the same reaper, so it drains with it too (#2787); that drain joins the serial tail the assigns use, because the load does not await the reap and a drain outside it would start a second card beside the assign at the lane's cap (#2757). 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). The deadline is not queued behind the wait it bounds (#2748): an expiry runs off the serial tail, so a host call that never answers — a dispatch, a release, a spawn — cannot park the one timer that ends the run, and the run still reaches `failed` with its Bots released. A start whose Bots the host does not create within `SPAWN_DEADLINE_MS` (60 s) is refused, and the card a drain had already taken out of `lane.queue` goes back to it instead of vanishing (a late spawn's Bots are released whether that answer is a success or a failure: a failed one carries the ids it already made on `error.spawned`, which the deadline branch now releases, #2785). 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 shapes at the start of the text only (` could not answer your message: …`, ` was stopped before it could answer your message.`, and the bare `could not answer your message` / `The model call failed.` an old host wrote), never a free substring, so an ordinary answer that merely ends in the model-failure wording keeps advancing the step (#2741). A stopped turn is a failed step, with the stop notice as its reason, instead of finishing as done and handing that notice to the next node (#2741). 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 once the Bot finally goes; while it still will not, that retry is bounded — `STOP_RELEASE_RETRIES` (two) more deadlines — and the last one ends the run `failed` as `release-failed` with the ids the host would not remove and the attempt count on the row, so the slot comes back and the queue drains (#2453, P-560, #2750). 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. Since #2742 every reply the host completes carries that field — an answer `failure: false`, a turn stopped before it answered `failure: true` with the stop notice as its reason — so the middleware classifies by the marker and the prose shapes are only the compatibility path for messages saved before it. 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 start that is still waiting on the host counts against that cap: the card leaves the queue before its run exists, so two steps ending in the same turn would otherwise both read the free slot and start a card each, one past `maxConcurrent` (#2748). 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. diff --git a/plugins/workflow/src/service.js b/plugins/workflow/src/service.js index bae4c7e64..92c75a04d 100644 --- a/plugins/workflow/src/service.js +++ b/plugins/workflow/src/service.js @@ -279,7 +279,7 @@ export function createWorkflowService({ if (readOnly()) return false writer = true try { - takingOver = Promise.resolve(reapRuns()) + takingOver = reapOnce() } catch (error) { takingOver = Promise.resolve() warn(`flybot-workflow: the reap after takeover failed: ${error?.message ?? error}`) @@ -1477,8 +1477,8 @@ export function createWorkflowService({ // 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. + // (#1388) and only gives back the Bots it still names. The queue is not this function's: + // a parked card is not a run, and the drain that follows the reap is what starts it. 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 @@ -1548,6 +1548,20 @@ export function createWorkflowService({ }) } + // #2752: a run this reap ended is a terminal transition like a step's own ending, so the + // queue moves on it. With the drain on the finish paths only, the card behind a + // `host-ended` run stays parked while the lane reports a free slot and nothing running, + // and the next assign is what finally starts it. + // #2757: the caller does not await this reap — a load fires it and moves on — so the + // drain joins the serial tail the assigns use. Outside it, this drain and a concurrent + // `assign` both read `runningOf` before either `beginRun` lands, each sees a free slot + // and starts a card: two runs at cap 1, the second one over the lane's ceiling. + // #2787: every path that starts the reaper gets the same answer, so takeover's reap + // drains too instead of leaving the dead window's queue parked. + function reapOnce() { + return reapRuns().then(() => hostSerial(() => drainAll())) + } + return { state(body) { // #2704: a window that has just taken the lock adopts the store and reaps before @@ -2351,17 +2365,11 @@ export function createWorkflowService({ 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. - const pending = takingOver !== null ? takingOver : reapRuns() + // #2787: that reaper carries its own drain, so handing it back is the whole answer. + // Stacking a second drain here would reopen the cap-1 race #2757 closed. + const pending = takingOver !== null ? takingOver : reapOnce() takingOver = null - // #2752: a run this reap ended is a terminal transition like a step's own ending, so - // the queue moves on it. With the drain on the finish paths only, the card behind a - // `host-ended` run stays parked while the lane reports a free slot and nothing - // running, and the next assign is what finally starts it. - // #2757: the caller does not await this reap — a load fires it and moves on — so the - // drain joins the serial tail the assigns use. Outside it, this drain and a concurrent - // `assign` both read `runningOf` before either `beginRun` lands, each sees a free slot - // and starts a card: two runs at cap 1, the second one over the lane's ceiling. - return pending.then(() => hostSerial(() => drainAll())) + return pending }, dispose() { for (const key of [...timers.keys()]) clearTimerAt(key) diff --git a/plugins/workflow/test/takeover-transition-drain.test.js b/plugins/workflow/test/takeover-transition-drain.test.js new file mode 100644 index 000000000..eda638fa6 --- /dev/null +++ b/plugins/workflow/test/takeover-transition-drain.test.js @@ -0,0 +1,129 @@ +// #2787: a takeover reap is a terminal transition like any other, so the lane's queue +// moves on it. `#2757` gave the load-time `reap()` a drain on the serial tail; the +// takeover path (`#2704`) starts the same reaper but stopped there, so the run the dead +// window left `running` was ended `host-ended` while the card parked behind it stayed in +// `lane.queue` until some unrelated assign/ingest. `#2752` is that symptom. +// +// The takeover here is driven the way the plugin drives it: a window opens while another +// holds the lock, so it is a reader; the first call that finds the host writable is what +// takes the store over. No `assign` and no `ingest` follows — the reap itself is the only +// thing that could start the parked card. +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createWorkflowService } from '../src/service.js' + +const AT = 1_700_000_000_000 +const ROSTER = [{ id: 'main', name: 'Chief', isMain: true, remote: false }] +const CLOSED = 'dead-running' +const PARKED = 'dead-queued' + +// One turn of the event loop apiece: the reap and its drain are promises the caller does +// not await (`plugins/workflow/index.js` fires the load reap and moves on), so the test +// waits for the work instead of for a call to return. +const settle = async (rounds = 300) => { + for (let index = 0; index < rounds; index += 1) await new Promise(done => setImmediate(done)) +} + +function bench(dir, options = {}) { + const seen = { released: [], spawned: [] } + let seq = 100 + const svc = createWorkflowService({ + dir, + persist: true, + now: () => AT, + bots: () => ROSTER, + spawn: async (spec) => { + const id = `eph-${spec.nodeId}-${++seq}` + seen.spawned.push(id) + return id + }, + release: async (id) => { seen.released.push(id) }, + dispatch: async () => ({ ok: true }), + // No deadlines or spawn bounds here: the ending under test is the reap's. + setTimer: () => 1, + clearTimer: () => {}, + ...options, + }) + return { svc, seen } +} + +async function seedLane(dir) { + const first = bench(dir) + const made = first.svc.define({ + name: 'Lane', + nodes: [{ id: 'work', name: 'Worker', prompt: 'Work.' }], + edges: [], + }) + assert.equal(made.ok, true, JSON.stringify(made)) + const workflowId = made.workflows.find(item => item.name === 'Lane').id + await first.svc.setConcurrency({ workflowId, maxConcurrent: 1 }) + await first.svc.runNamed({ workflow: 'Lane', brief: 'dead run', task: CLOSED }) + await first.svc.assign({ workflowId, taskId: PARKED, title: 'parked card' }) + const row = first.svc.state().workflows.find(item => item.id === workflowId) + assert.deepEqual(row.runs.filter(run => run.status === 'running').map(run => run.task?.id), [CLOSED]) + assert.equal(row.queueDepth, 1) + const deadBots = row.runs.find(run => run.task?.id === CLOSED)?.ephemeral ?? {} + return { workflowId, deadBots: Object.values(deadBots).sort() } +} + +const rowOf = (svc, workflowId) => svc.state().workflows.find(item => item.id === workflowId) +const running = (svc, workflowId) => rowOf(svc, workflowId).runs.filter(run => run.status === 'running').map(run => run.task?.id) +const depth = (svc, workflowId) => rowOf(svc, workflowId).queueDepth + +test('the takeover reap starts the card parked behind the run it ends (#2787)', { timeout: 30_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-takeover-drain-')) + try { + const seed = await seedLane(dir) + // The other window holds the lock while this one opens, so this one is a reader. + const lock = { writable: false } + const second = bench(dir, { readOnly: () => (lock.writable ? undefined : 'read-only') }) + await second.svc.reap() // what loading does: a reader touches nothing (#2673) + assert.deepEqual(running(second.svc, seed.workflowId), [CLOSED], 'a read-only host ended the writer run') + assert.equal(depth(second.svc, seed.workflowId), 1) + + // The first call that finds the host writable takes the store over (#2704). Nothing + // is assigned and nothing is ingested after it: only the reap can move the queue. + lock.writable = true + second.svc.state() + await settle() + + assert.deepEqual(running(second.svc, seed.workflowId), [PARKED], 'the parked card did not start after the takeover reap ended the run') + assert.equal(depth(second.svc, seed.workflowId), 0, 'the queue kept the card while the line had a free slot') + assert.deepEqual([...second.seen.released].sort(), seed.deadBots, 'the takeover did not release the dead run\'s step Bots once') + } finally { + rmSync(dir, { recursive: true, force: true }) + } +}) + +// The control: the drain belongs to the reap, not to whichever call noticed the takeover. +// (a) a writer that reaps starts the card (#2757's path), (b) a takeover reap starts it +// too (#2787's hole), (c) an explicit reap after the takeover starts it as well — and does +// not drain a second time, because `reap()` hands back the takeover's own work (#2757's +// cap-1 race is what a second drain reopens). +test('a takeover reap drains like the load reap, and a later reap does not drain twice (#2787)', { timeout: 60_000 }, async () => { + for (const shape of ['writer', 'takeover', 'takeover-then-reap']) { + const dir = mkdtempSync(join(tmpdir(), 'wf-takeover-control-')) + try { + const seed = await seedLane(dir) + const lock = { writable: shape === 'writer' } + const next = bench(dir, { readOnly: () => (lock.writable ? undefined : 'read-only') }) + if (shape === 'writer') { + await next.svc.reap() + } else { + lock.writable = true + next.svc.state() + if (shape === 'takeover-then-reap') await next.svc.reap() + await settle() + } + assert.deepEqual(running(next.svc, seed.workflowId), [PARKED], `${shape}: the parked card did not start`) + assert.equal(depth(next.svc, seed.workflowId), 0, `${shape}: the queue kept the card`) + assert.deepEqual([...next.seen.released].sort(), seed.deadBots, `${shape}: the dead run's Bots were not released exactly once`) + assert.equal(rowOf(next.svc, seed.workflowId).activeCount, 1, `${shape}: the lane started more than its cap`) + } finally { + rmSync(dir, { recursive: true, force: true }) + } + } +}) diff --git a/plugins/workflow/test/terminal-transition-drain.test.js b/plugins/workflow/test/terminal-transition-drain.test.js new file mode 100644 index 000000000..5fa25e941 --- /dev/null +++ b/plugins/workflow/test/terminal-transition-drain.test.js @@ -0,0 +1,199 @@ +// #2787 (folded in from P-682): every terminal transition drains. A run that ends is a +// free slot; a card parked behind it must move into that slot in the same turn, or the +// lane reports `queueDepth > 0` with nothing running and looks idle until some unrelated +// assign/ingest. #2752 is that shape on the finish paths; the reap paths (#2757, #2787) +// are the ones that were missing it. +// +// One bench for every row: a one-node lane at cap 1, one run holding the slot, one card +// parked behind it. Each row ends the run a different way and then asks the same question: +// is the lane left with a card parked behind a free slot? A row fails only itself when its +// own ending loses its drain — step success/failure, deadline, Stop, graph edit, load reap +// and takeover reap all start the parked card; Reset drops the queue by design, so its row +// asserts that the lane is emptied instead of left half-moved. +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { STEP_DEADLINE_MS, createWorkflowService } from '../src/service.js' + +const AT = 1_700_000_000_000 +const ROSTER = [{ id: 'main', name: 'Chief', isMain: true, remote: false }] +const RUNNING = '#A' +const PARKED = '#B' +const WORKFLOW = 'Lane' + +// Timers the test fires itself: the deadline row needs the step deadline, and every other +// row never wants a timer to move the run it is about to end. +function scheduler() { + let seq = 0 + const queue = new Map() + const setTimer = (fn, ms) => { + const id = ++seq + queue.set(id, { fn, due: Math.max(1, Number(ms) || 0) }) + return id + } + const clearTimer = id => { queue.delete(id) } + async function tick(ms) { + let left = Math.max(0, Number(ms) || 0) + for (;;) { + let next + for (const [id, entry] of queue) { + if (entry.due > left) continue + if (!next || entry.due < next.entry.due) next = { id, entry } + } + if (!next) return + queue.delete(next.id) + left -= next.entry.due + for (const entry of queue.values()) entry.due -= next.entry.due + await next.entry.fn() + } + } + return { setTimer, clearTimer, tick } +} + +function service(dir, options = {}) { + const seen = { released: [] } + let seq = 0 + const timers = scheduler() + const svc = createWorkflowService({ + dir, + persist: true, + now: () => AT, + bots: () => ROSTER, + spawn: async (spec) => `eph-${spec.nodeId}-${++seq}`, + release: async (id) => { seen.released.push(id) }, + dispatch: async () => ({ ok: true }), + setTimer: timers.setTimer, + clearTimer: timers.clearTimer, + ...options, + }) + return { svc, seen, timers } +} + +// The same bench for every row: cap 1, `#A` holding the slot with a live Bot on it, `#B` +// parked behind it. +async function bench(dir) { + const made = service(dir) + const defined = made.svc.define({ + name: WORKFLOW, + nodes: [{ id: 'work', name: 'Worker', prompt: 'Work.' }], + edges: [], + }) + assert.equal(defined.ok, true, JSON.stringify(defined)) + const workflowId = defined.workflows.find(item => item.name === WORKFLOW).id + await made.svc.setConcurrency({ workflowId, maxConcurrent: 1 }) + await made.svc.runNamed({ workflow: WORKFLOW, brief: 'first', task: RUNNING }) + await made.svc.assign({ workflowId, taskId: PARKED, title: 'parked card' }) + const row = rowOf(made.svc, workflowId) + const live = row.runs.find(run => run.status === 'running') + assert.equal(live?.task?.id, RUNNING, JSON.stringify(row)) + assert.equal(row.queueDepth, 1, 'the bench did not park the second card') + return { ...made, dir, workflowId, runId: live.id } +} + +const settle = async (rounds = 300) => { + for (let index = 0; index < rounds; index += 1) await new Promise(done => setImmediate(done)) +} +const rowOf = (svc, workflowId) => svc.state().workflows.find(item => item.id === workflowId) +const runningOf = (svc, workflowId) => rowOf(svc, workflowId).runs.filter(run => run.status === 'running').map(run => run.task?.id) +const depthOf = (svc, workflowId) => rowOf(svc, workflowId).queueDepth + +const ROWS = [ + { + name: 'a step that succeeds', + ends: 'starts', + async run(h) { + await h.svc.advance({ workflowId: h.workflowId, nodeId: 'work', outcome: 'done', text: 'done' }) + }, + }, + { + name: 'a step that fails', + ends: 'starts', + async run(h) { + await h.svc.advance({ workflowId: h.workflowId, nodeId: 'work', outcome: 'failed', text: 'step failed' }) + }, + }, + { + name: 'a step deadline', + ends: 'starts', + async run(h) { + await h.timers.tick(STEP_DEADLINE_MS) + }, + }, + { + name: 'a Stop', + ends: 'starts', + async run(h) { + await h.svc.reset({ workflowId: h.workflowId, runId: h.runId }) + }, + }, + { + // Reset is the one ending that drops the parked card on purpose: it clears the lane, + // so the row's question is answered by an emptied lane instead of a started card. + name: 'a Reset', + ends: 'cleared', + async run(h) { + await h.svc.reset({ workflowId: h.workflowId }) + }, + }, + { + name: 'a graph edit that drops the running node', + ends: 'starts', + async run(h) { + await h.svc.saveGraph({ + workflowId: h.workflowId, + nodes: [{ id: 'other', name: 'Other', prompt: 'Other work.' }], + edges: [], + }) + }, + }, + { + // The window that loads the dead writer's store: the reap a load fires. + name: 'a load reap', + ends: 'starts', + async run(h, dir) { + const next = service(dir) + await next.svc.reap() + h.after = next + }, + }, + { + // The same reaper started by the first call that finds the host writable (#2704). + name: 'a takeover reap', + ends: 'starts', + async run(h, dir) { + const lock = { writable: false } + const next = service(dir, { readOnly: () => (lock.writable ? undefined : 'read-only') }) + await next.svc.reap() // a reader touches nothing (#2673) + lock.writable = true + next.svc.state() + h.after = next + }, + }, +] + +for (const row of ROWS) { + test(`every terminal transition drains: ${row.name}`, { timeout: 30_000 }, async () => { + const dir = mkdtempSync(join(tmpdir(), 'wf-terminal-drain-')) + try { + const h = await bench(dir) + const at = { ...h } + await row.run(h, dir) + await settle() + const view = h.after?.svc ?? at.svc + const workflowId = at.workflowId + + // The run the row ended is gone either way; what differs is where the parked card went. + if (row.ends === 'starts') { + assert.deepEqual(runningOf(view, workflowId), [PARKED], `${row.name}: the parked card did not start`) + assert.equal(depthOf(view, workflowId), 0, `${row.name}: the lane kept the card behind a free slot`) + } else { + assert.deepEqual(runningOf(view, workflowId), [], `${row.name}: a run is still holding the lane`) + assert.equal(depthOf(view, workflowId), 0, `${row.name}: the lane kept a card it will never start`) + } + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) +}