From f103fee796d8eaad54b22501a9393c2048b7cb7e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 28 Oct 2026 02:04:29 +0000 Subject: [PATCH] Remove host-wide concurrency cap and add Settings UI control MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove HOST_CAP = 6 and globalRunning() from service - Update hasRoom(), drainSync(), drainAll() to check only per-workflow maxConcurrent - Add concurrent runs number input to Settings → Workflow graphs - Update docs to remove host cap and note Settings UI control - Update queue test to reflect per-workflow-only gating - Add client test for Settings concurrency control Fixes #1525 Co-authored-by: Danyang Chen --- .../1525-workflow-concurrency.changed.md | 1 + docs/subsystems/workflow.md | 4 +- plugins/workflow/README.md | 4 +- plugins/workflow/client.js | 17 +++++ plugins/workflow/src/service.js | 18 ++---- plugins/workflow/test/client.test.js | 29 +++++++++ plugins/workflow/test/queue.test.js | 63 +++++++------------ 7 files changed, 78 insertions(+), 58 deletions(-) create mode 100644 changelog.d/1525-workflow-concurrency.changed.md diff --git a/changelog.d/1525-workflow-concurrency.changed.md b/changelog.d/1525-workflow-concurrency.changed.md new file mode 100644 index 000000000..a58660e22 --- /dev/null +++ b/changelog.d/1525-workflow-concurrency.changed.md @@ -0,0 +1 @@ +Workflow concurrency is now configurable per Workflow in Settings → Workflow graphs. The host-wide 6-run cap is removed; each Workflow's own `maxConcurrent` limit (1–8, default 2) is the only gate. [#1525](https://github.com/ThinkFlowLab/FlyBot/issues/1525) diff --git a/docs/subsystems/workflow.md b/docs/subsystems/workflow.md index bd9ec9331..2cda57cf5 100644 --- a/docs/subsystems/workflow.md +++ b/docs/subsystems/workflow.md @@ -38,7 +38,7 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher, ## Entry points - Endpoint `workflow/state` and `workflow/templates` are reads. -- `workflow/create`, `workflow/rename`, `workflow/delete`, `workflow/assign`, `workflow/concurrency`, `workflow/graph`, `workflow/advance`, `workflow/reset`, `workflow/say`, `workflow/ingest` write. `workflow/concurrency` sets that workflow's `maxConcurrent` (an integer from 1 to 8). `workflow/rename` updates a workflow's name and optional description; read-only hosts are refused. `workflow/delete` removes a workflow from state, releases its ephemeral Bots if a run exists, and cleans logs/runs/routes. +- `workflow/create`, `workflow/rename`, `workflow/delete`, `workflow/assign`, `workflow/concurrency`, `workflow/graph`, `workflow/advance`, `workflow/reset`, `workflow/say`, `workflow/ingest` write. `workflow/concurrency` sets that workflow's `maxConcurrent` (an integer from 1 to 8); the same control is in Settings → Workflow graphs. `workflow/rename` updates a workflow's name and optional description; read-only hosts are refused. `workflow/delete` removes a workflow from state, releases its ephemeral Bots if a run exists, and cleans logs/runs/routes. - Tool `run_workflow` starts a named run with a brief. Tool `define_workflow` writes the same graph the canvas edits. - `flybot.tasks.onWrite` calls `ingest`. A new Bot match returns `{ ownerBotId }`. A Workflow match starts the run inside the plugin and leaves the card's owner to the host. @@ -47,7 +47,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). Status highlighting reads that same graph. Replacing edges does not recompute who is already running. 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. There is no scripted or seeded run path: `workflow/state` exposes only `live === true` runs. 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 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 dispatch that does not land is not marked handed. The chat names the code, and the next hand-off tries that node again. It is not sent twice in the same hand-off. A reply that says `could not answer your message` or `model call failed` fails that node and does not start its successor. A 429 or rate-limit mention in any other reply does not. -4. Each workflow keeps a FIFO queue and up to `maxConcurrent` live runs (default 2, an integer from 1 to 8 on the record or via `workflow/concurrency`). The host runs at most 6 live runs at once. A task that does not fit is enqueued. When a run finishes, every workflow with a free slot is drained, oldest enqueue first, until the host cap. 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, and the chip count does not exceed that list. +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, and the chip count does not exceed that list. 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 : `. The Bot reports `OUTPUT : `; a single declared output may be the reply body. 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. **Every node Bot gets the same full permissions a normal sidebar Bot gets:** the `tools` field is always omitted from `fb.bot.create`, letting the Bot inherit all registered tools — file/shell tools, GitHub (`mcp__github__*`: pull requests, reviews, comments), Composio, and team tools — limited only by `EPHEMERAL_DENIED_TOOLS` at the guard/approval layer (#1466, 29 team-admin tools). `node.tools` is stored as documentation in `fields.tools` but does not restrict the grant. All node Bots get `danger-full-access` permission. A hidden run Bot cannot show an approval card, so a granted tool is allowed and a connector it did not name is refused. The host stores this on the run Bot as `stepTools` and `permission`. 7. A loop edge (`loop: true`, `max` from 1 to 8, default 3, optional `until: { output, eq }`) is the only cycle. The body runs again until `until` matches or `max` runs have finished, and the forward edges wait until then. Restarting the body drops the branch choice made inside it and queues a sibling that the previous pass skipped, so the next pass can take the other side. A `when: { output, eq }` edge is taken only when that output equals `eq`. `when: { else: true }` is taken only when no other `when` matched. A node whose forward edges are all closed is `skipped` and does not block a join. A run cannot spin: every loop has a max. Settings → Workflow graphs can add a link, a condition or a loop. diff --git a/plugins/workflow/README.md b/plugins/workflow/README.md index ee1dc3451..8445cec80 100644 --- a/plugins/workflow/README.md +++ b/plugins/workflow/README.md @@ -24,7 +24,7 @@ Assigning a task, `run_workflow`, or a Workflow match starts a live run. One hid A second task does not replace a run that is already going, and it does not release that run's Bots. It waits in a FIFO queue. The stage shows a chip when more than one run is active or the queue is not empty, and it refreshes `workflow/state` while a live run exists. That refresh does not advance the graph. The product UI does not offer a canned demo replay; only live runs advance the graph. -Each workflow allows `maxConcurrent` live runs at once. The default is 2. The value is an integer from 1 to 8, set on create (`maxConcurrent`) or with `workflow/concurrency` (`{ workflowId, maxConcurrent }`). The stage lists the current value under Details. The host runs at most 6 live runs across every workflow. A task that does not fit is queued. When any run finishes, the oldest queued task whose workflow still has room starts, whichever workflow it belongs to. A new assign waits behind a task already queued there. Lowering the cap does not stop a run that has already started. History keeps finished and failed runs only. +Each workflow allows `maxConcurrent` live runs at once. The default is 2. The value is an integer from 1 to 8, set on create (`maxConcurrent`), with `workflow/concurrency` (`{ workflowId, maxConcurrent }`), or in Settings → Workflow graphs. A task that does not fit is queued. When any run finishes, the oldest queued task whose workflow still has room starts, whichever workflow it belongs to. A new assign waits behind a task already queued there. Lowering the cap does not stop a run that has already started. History keeps finished and failed runs only. ## Inputs and outputs @@ -81,7 +81,7 @@ Workflow 是一组本地 Bot 的群聊式运行。侧栏打开同一次运行： 一次真实运行会为每个节点创建一个隐藏的临时 Bot，用完即删。节点不必绑定侧栏里的永久 Bot。`tools` 决定这个 Bot 能用什么：`github` 授予 GitHub 连接器（拉取请求、评审、评论），`bash` 授予完整权限的 shell，其余步骤至少可以写工作区。节点一直处于运行中，直到该 Bot 回复。回复写入 Workflow 聊天，下一节点才沿边开始。正在进行的运行不会被新任务替换，也不会因此释放它的 Bot。历史里只留下已完成或已失败的运行。 -每个 Workflow 同时运行的上限是 `maxConcurrent`，默认 2，取 1 到 8 的整数，在创建时传入，或调用 `workflow/concurrency`。全部 Workflow 加在一起最多 6 个正在运行的任务。某个运行结束后，所有 Workflow 里最早排队、且自己还有空位的任务会开始。新的分配排在已经排队的任务后面。降低上限不会停掉已经开始的运行。舞台在详情里显示当前上限；有排队或同时多于一个运行时显示队列条。有真实运行时，页面只刷新状态。产品界面不提供预置演示重放，只有真实运行会推进图。 +每个 Workflow 同时运行的上限是 `maxConcurrent`，默认 2，取 1 到 8 的整数，在创建时传入、调用 `workflow/concurrency` 或在设置 → Workflow graphs 里调整。某个运行结束后，所有 Workflow 里最早排队、且自己还有空位的任务会开始。新的分配排在已经排队的任务后面。降低上限不会停掉已经开始的运行。舞台在详情里显示当前上限；有排队或同时多于一个运行时显示队列条。有真实运行时，页面只刷新状态。产品界面不提供预置演示重放，只有真实运行会推进图。 每个节点最多声明 8 个输入和 8 个输出，名字为小写字母开头的短名。必填输入必须由上游同名输出提供；根节点可以用 `task`、`title` 或 `brief`。没有提供者时，运行拒绝开始。Bot 用 `OUTPUT 名字: 值` 交回输出；只声明一个输出时，回复正文即可。 diff --git a/plugins/workflow/client.js b/plugins/workflow/client.js index fe7aa0254..1e0d5ddb0 100644 --- a/plugins/workflow/client.js +++ b/plugins/workflow/client.js @@ -899,6 +899,23 @@ window.__ModuleLoader__.load({ }, h('option', { value: '' }, tr('Each Bot\'s own model')), (pack?.roster?.models ?? []).map(model => h('option', { key: model.ref, value: model.ref }, `${model.name} (${model.provider})`)))), + h('label', { className: 'wf-field' }, + h('span', null, tr('Concurrent runs')), + h('input', { + type: 'number', + 'aria-label': tr('Concurrent runs'), + min: 1, + max: 8, + value: current.maxConcurrent ?? 2, + onChange: (event) => { + const maxConcurrent = Number(event.target.value) + if (!Number.isInteger(maxConcurrent) || maxConcurrent < 1 || maxConcurrent > 8) return + void Promise.resolve(call('workflow/concurrency', { workflowId: current.id, maxConcurrent })).then(value => { + if (value?.ok === false) { setError(value.error ?? ''); return } + if (value?.workflows) setPack(value) + }) + }, + })), h('button', { type: 'button', className: 'wf-btn', onClick: () => setEditing(!editing) }, editing ? tr('Finish editing') : tr('Edit')), editing ? h('span', { className: 'wf-autosave-hint' }, tr('Changes are saved automatically')) : null), diff --git a/plugins/workflow/src/service.js b/plugins/workflow/src/service.js index 46df105de..c2fffe829 100644 --- a/plugins/workflow/src/service.js +++ b/plugins/workflow/src/service.js @@ -20,7 +20,6 @@ const MAX_MESSAGES = 200 const FILE = 'workflows.json' const DEFAULT_CONCURRENT = 2 const MAX_CONCURRENT = 8 -const HOST_CAP = 6 function idOf(prefix) { return `${prefix}-${randomBytes(4).toString('hex')}` @@ -109,8 +108,7 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no if (!state.laneFormat) migrateRuns(state) purgeDemos(state) let seq = state.seq ?? 1 - // One lock for every start and release. The host cap is read and updated - // only while this is held, so two workflows cannot both pass it. + // One lock for every start and release to serialize drain operations. let hostTail = Promise.resolve() function hostSerial(fn) { const run = hostTail.then(() => fn()) @@ -201,12 +199,6 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no return (lane?.active ?? []).filter(run => run && run.live === true && run.status === 'running') } - function globalRunning() { - let count = 0 - for (const lane of Object.values(state.lanes)) count += runningOf(lane).length - return count - } - function maxOf(item) { const value = Number(item?.maxConcurrent) if (!Number.isInteger(value) || value < 1) return DEFAULT_CONCURRENT @@ -450,10 +442,10 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no } // After a slot frees, start the oldest parked task on any workflow that - // still has room, and stop at the host cap. + // still has room. function drainSync() { let guard = 0 - while (globalRunning() < HOST_CAP && guard < 48) { + while (guard < 48) { guard += 1 const next = takeNext() if (!next) return @@ -471,7 +463,7 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no return } let guard = 0 - while (globalRunning() < HOST_CAP && guard < 48) { + while (guard < 48) { guard += 1 const next = takeNext() if (!next) return @@ -563,7 +555,7 @@ export function createWorkflowService({ dir, persist = true, now = () => Date.no } function hasRoom(item) { - return runningOf(laneOf(item.id)).length < maxOf(item) && globalRunning() < HOST_CAP + return runningOf(laneOf(item.id)).length < maxOf(item) } // A non-empty queue is joined at the tail, then the oldest task starts. diff --git a/plugins/workflow/test/client.test.js b/plugins/workflow/test/client.test.js index 8dbd6b92a..2e6bd08d5 100644 --- a/plugins/workflow/test/client.test.js +++ b/plugins/workflow/test/client.test.js @@ -367,6 +367,35 @@ test('apply registers the sidebar section and the settings page', () => { off() }) +test('settings can change the workflow concurrency limit', async () => { + const calls = [] + const graph = delivery() + graph.maxConcurrent = 2 + const call = async (name, body) => { + calls.push({ name, body }) + if (name === 'workflow/state') return { workflows: [graph] } + if (name === 'workflow/concurrency') { + graph.maxConcurrent = body.maxConcurrent + return { ok: true, workflows: [graph] } + } + return { ok: true, workflows: [graph] } + } + const settings = mount(client.WorkflowSettings, { call, roster: { bots: [] } }) + await settle() + const concurrentInput = findAll(settings.tree, node => node.props?.['aria-label'] === 'Concurrent runs')[0] + assert.ok(concurrentInput, 'concurrent runs input should exist') + assert.equal(concurrentInput.props.value, 2) + assert.equal(concurrentInput.props.min, 1) + assert.equal(concurrentInput.props.max, 8) + + concurrentInput.props.onChange({ target: { value: '4' } }) + await settle() + const changed = calls.filter(item => item.name === 'workflow/concurrency').at(-1) + assert.equal(changed.body.workflowId, 'delivery') + assert.equal(changed.body.maxConcurrent, 4) + settings.unmount() +}) + test('settings can add a loop edge and a condition edge', async () => { const calls = [] const graph = delivery() diff --git a/plugins/workflow/test/queue.test.js b/plugins/workflow/test/queue.test.js index ac1b1d21b..9b6f2ae07 100644 --- a/plugins/workflow/test/queue.test.js +++ b/plugins/workflow/test/queue.test.js @@ -326,51 +326,32 @@ test('an old store drops scripted runs and keeps only live abandoned runs', () = } }) -test('a task parked at the host cap starts when any run finishes, ahead of a newer assign', async () => { - let calls = 0 - let release - const gate = new Promise(resolve => { release = resolve }) - const { svc } = harness({ - blockSpawn: () => { - calls += 1 - if (calls === 6) return gate - return undefined - }, - }) - const ids = [] - for (let n = 0; n < 7; n += 1) { - const made = svc.define({ - name: `Cap ${n}`, - nodes: [{ id: 'only', name: 'Only', outputs: [{ name: 'out', type: 'text' }] }], - edges: [], - }).workflows.find(item => item.name === `Cap ${n}`) - ids.push(made.id) - } - for (let n = 0; n < 5; n += 1) { - const started = await svc.assign({ workflowId: ids[n], taskId: `#run${n}`, title: `Run ${n}` }) - assert.equal(started.workflows.find(item => item.id === ids[n]).activeCount, 1) - } - const sixth = svc.assign({ workflowId: ids[5], taskId: '#six', title: 'Six' }) - await new Promise(resolve => setImmediate(resolve)) - const parked = svc.assign({ workflowId: ids[6], taskId: '#parked', title: 'Parked' }) - release() - await sixth - const held = await parked - assert.equal(held.workflows.find(item => item.id === ids[5]).activeCount, 1) - const waiting = held.workflows.find(item => item.id === ids[6]) - assert.equal(waiting.activeCount, 0) - assert.deepEqual(waiting.queue.map(task => task.id), ['#parked']) +test('a queued task starts when its workflow finishes a run, ahead of a newer assign', async () => { + const { svc } = harness() + const made = svc.define({ + name: 'Serial', + nodes: [{ id: 'only', name: 'Only', outputs: [{ name: 'out', type: 'text' }] }], + edges: [], + }).workflows.find(item => item.name === 'Serial') + svc.setConcurrency({ workflowId: made.id, maxConcurrent: 1 }) + + const first = await svc.assign({ workflowId: made.id, taskId: '#first', title: 'First' }) + assert.equal(first.workflows.find(item => item.id === made.id).activeCount, 1) + + const second = await svc.assign({ workflowId: made.id, taskId: '#parked', title: 'Parked' }) + const held = second.workflows.find(item => item.id === made.id) + assert.equal(held.activeCount, 1) + assert.deepEqual(held.queue.map(task => task.id), ['#parked']) - const newer = await svc.assign({ workflowId: ids[6], taskId: '#newer', title: 'Newer' }) - const both = newer.workflows.find(item => item.id === ids[6]) - assert.equal(both.activeCount, 0) + const third = await svc.assign({ workflowId: made.id, taskId: '#newer', title: 'Newer' }) + const both = third.workflows.find(item => item.id === made.id) + assert.equal(both.activeCount, 1) assert.deepEqual(both.queue.map(task => task.id), ['#parked', '#newer']) - const victim = newer.workflows.find(item => item.id === ids[0]).run + const victim = third.workflows.find(item => item.id === made.id).run const finished = await svc.stepResult({ botId: victim.ephemeral.only, ok: true, text: 'OUTPUT out: done' }) - assert.equal(finished.workflows.find(item => item.id === ids[0]).activeCount, 0) - const woke = finished.workflows.find(item => item.id === ids[6]) - assert.equal(woke.activeCount, 1) + assert.equal(finished.workflows.find(item => item.id === made.id).activeCount, 1) + const woke = finished.workflows.find(item => item.id === made.id) assert.equal(woke.run.task.id, '#parked') assert.deepEqual(woke.queue.map(task => task.id), ['#newer']) })