diff --git a/.github/scripts/ds-bot-patches.test.mjs b/.github/scripts/ds-bot-patches.test.mjs index b896a6795..7be9c8d1b 100644 --- a/.github/scripts/ds-bot-patches.test.mjs +++ b/.github/scripts/ds-bot-patches.test.mjs @@ -193,7 +193,7 @@ if (gitAvailable) { const root = join(import.meta.dirname, '..', '..') const patchesDir = join(root, 'patches', 'ds-bot') const stored = readdirSync(patchesDir).filter(name => /^\d{4}-.*\.patch$/.test(name)).sort() - assert.equal(stored.length, 10) + assert.equal(stored.length, 11) const viaGit = mkdtempSync(join(tmpdir(), 'patch-git-')) const viaOurs = mkdtempSync(join(tmpdir(), 'patch-ours-')) try { diff --git a/changelog.d/1729.added.md b/changelog.d/1729.added.md new file mode 100644 index 000000000..434a7eb69 --- /dev/null +++ b/changelog.d/1729.added.md @@ -0,0 +1 @@ +Role queues now carry a stated cap (8 by default, override per graph), every claim runs in its own worker with no shared state, and one owner recovers an expired lease — once — recording it on the card's evidence. diff --git a/core/src/host/hooks.js b/core/src/host/hooks.js index 7a804740d..1de63b2f3 100644 --- a/core/src/host/hooks.js +++ b/core/src/host/hooks.js @@ -1336,6 +1336,14 @@ export function install(rt) { onProjectProvider(fn) { return rt.onProjectProvider?.(fn) ?? (() => {}) }, + // #1729: a workflow reads a card's lease, and the delivery/resource counts, through the + // host — never a copy of its own, so the two cannot disagree about who holds a card. + lease(taskId) { + return rt.taskLease?.(taskId) + }, + counters(query) { + return rt.taskCounters?.(query) ?? { roles: {}, runs: {} } + }, }, tools, api, diff --git a/core/src/host/tasks.js b/core/src/host/tasks.js index 10bd79012..9828d4e9f 100644 --- a/core/src/host/tasks.js +++ b/core/src/host/tasks.js @@ -47,6 +47,20 @@ const PRIORITY = new Set(['urgent', 'high', 'normal', 'low']) const ROLE_MAX = 24 const QUEUE_RETRY_DEFAULT_SEC = 60 const QUEUE_RETRY_MAX_SEC = 7 * 24 * 60 * 60 +// One queue per role, with a stated ceiling (#1729). The default lives here; a graph that knows +// how many workers a role runs overrides it on the card it enqueues, and a re-offer keeps it. +const ROLE_QUEUE_CAP = 8 +const QUEUE_CAP_MAX = 64 +// A claim runs in its own worker, and two claims never share one (#1729). Its state is a pair of +// maps the caller writes; nothing carries over from the previous claim's worker, and returning the +// card to the queue drops the worker so the next claimant starts clean. +const WORKERS_KEPT = 64 +const readCap = (value) => { + if (value === undefined || value === null || value === '') return undefined + const cap = Number(value) + if (!Number.isInteger(cap) || cap < 1 || cap > QUEUE_CAP_MAX) throw invalid(`A queue cap must be a whole number from 1 to ${QUEUE_CAP_MAX}`) + return cap +} const readRole = (value) => { if (value === undefined || value === null) return undefined const role = oneLine(value, ROLE_MAX) @@ -480,6 +494,63 @@ export function install(rt) { task.taskClass = 'work' return task } + // ---------------------------------------------------------------- isolated workers (#1729) + const workers = new Map() + function newWorker() { + const id = `worker-${randomBytes(6).toString('hex')}` + workers.set(id, { id, files: new Map(), vars: new Map(), claimedAt: Date.now() }) + while (workers.size > WORKERS_KEPT) workers.delete(workers.keys().next().value) + return id + } + const dropWorker = (workerId) => { if (typeof workerId === 'string') workers.delete(workerId) } + // The worker of one claim, or nothing: a caller reaches its own claim's maps only. + rt.taskWorkerState = (workerId) => { + const worker = workers.get(String(workerId ?? '')) + return worker === undefined ? undefined : { id: worker.id, files: worker.files, vars: worker.vars, claimedAt: worker.claimedAt } + } + // Measured deliveries from the rows themselves (#1729): no second store, the numbers are a + // read of the cards. `roles` counts per role, `runs` per run id a card carries. + rt.taskCounters = function taskCounters(query = {}) { + const roleOf = (task) => { + const role = oneLine(task?.queueRole ?? '', ROLE_MAX) + if (role !== '') return role + return oneLine(botRecord(task?.ownerBotId)?.role ?? '', ROLE_MAX) + } + const init = () => ({ queued: 0, claimed: 0, working: 0, blocked: 0, accepted: 0, failed: 0, cancelled: 0, deliveries: 0, workerSeconds: 0 }) + const roles = {} + const runs = {} + const wanted = oneLine(query?.role ?? '', ROLE_MAX) + const wantedRun = query?.run === undefined || query?.run === null ? '' : String(query.run) + for (const task of Object.values(store())) { + const role = roleOf(task) + const run = String(task?.origin?.runId ?? task?.runId ?? '') + if (wanted !== '' && role !== wanted) continue + if (wantedRun !== '' && run !== wantedRun) continue + const bucket = (map, key) => (map[key] ??= init()) + const row = bucket(roles, role || 'unassigned') + if (row[task.status] !== undefined) row[task.status] += 1 + if (task.status === 'accepted') { + row.deliveries += 1 + const lease = leaseOf(task) + if (lease !== undefined && Number.isFinite(lease.takenAt)) row.workerSeconds += Math.max(0, Math.round((task.updatedAt - lease.takenAt) / 1000)) + } + if (run !== '') { + const runRow = bucket(runs, run) + if (runRow[task.status] !== undefined) runRow[task.status] += 1 + if (task.status === 'accepted') runRow.deliveries += 1 + } + } + return { roles, runs } + } + // The workflow plugin reads a card's lease through the host (#1729), never a copy of its own. + rt.taskLease = function taskLease(id) { + const task = find(String(id ?? '')) + if (task === undefined) return undefined + const lease = leaseOf(task) + if (lease === undefined) return undefined + const now = Date.now() + return { ...lease, live: lease.expiresAt > now, heldBy: holderName(lease.holder) } + } const leaseOf = task => (task?.lease !== null && typeof task?.lease === 'object' ? task.lease : undefined) const leaseLive = (task, now = Date.now()) => { const lease = leaseOf(task) @@ -494,16 +565,28 @@ export function install(rt) { for (const task of Object.values(store())) { const lease = leaseOf(task) if (lease === undefined || !Number.isFinite(lease.expiresAt) || lease.expiresAt > now) continue - if (!task.queueRole || (task.status !== 'claimed' && task.status !== 'working')) { - delete task.lease - continue - } - requeue(task, 'lease expired', 0) + // Read once and clear before the requeue: a second read in the same tick sees no lease, so + // two expired reads return the card once, not twice (#1729). The recovery is recorded on the + // card itself, and the dead worker is dropped so the next claim starts with no state of it. + const heldBy = holderName(lease.holder) delete task.lease + dropWorker(task.worker) + delete task.worker + if (!task.queueRole || (task.status !== 'claimed' && task.status !== 'working')) continue + const entry = { check: `lease expired (held by ${heldBy})`, at: now } + task.evidence = [...(Array.isArray(task.evidence) ? task.evidence : []), entry].slice(-EVIDENCE_MAX) + requeue(task, 'lease expired', 0) expired += 1 } return expired } + // The one owner of lease recovery (#1729): a read here is the same read the claim path makes, + // so a restart and a claim cannot both return the same expired card. + rt.taskExpireLeases = async (now = Date.now()) => { + const expired = expireLeases(now) + if (expired > 0) await save() + return expired + } // A rejection is a handover, not a status flip (#1720): the original's `handoff` names the // roles, and the rework is one child card linked by `reworkOf` — never `parentId`, which // means "part of an overall card" and must never make `settleOverall` fire. A replay of the @@ -642,6 +725,8 @@ export function install(rt) { // blocked → open is the queue's own edge: a member that cannot finish returns the item, and // no other path sends a card back to the queue, so the edge is not on TRANSITIONS. function requeue(task, reason, retryInSec) { + dropWorker(task.worker) + delete task.worker delete task.waitingFor delete task.inputRequest delete task.inputApprovalIds @@ -828,6 +913,8 @@ export function install(rt) { delete live.queueReason const claimedAt = Date.now() live.lease = { holder: botId, takenAt: claimedAt, expiresAt: claimedAt + LEASE_TTL_SEC * 1000 } + // Every claim gets a worker nobody else has started (#1729). + live.worker = newWorker() transition(live, 'claimed', null, { by: 'claimed from the queue' }) announceClaim(live) await save() @@ -972,8 +1059,17 @@ export function install(rt) { if (overall) task.kind = 'group' if (queue) { if (task.claimedBy) throw invalid('That queued item is already claimed; enqueue a new one') + // A graph may state the role's ceiling; otherwise the one constant does (#1729). The + // count leaves this card out, so re-offering a queued item keeps its id and is never + // refused for the place it already holds. + const cap = readCap(item.cap) ?? (Number.isInteger(task.queueCap) ? task.queueCap : ROLE_QUEUE_CAP) + const queued = Object.values(store()).filter(other => other.id !== task.id + && other.origin?.roomId === roomId && unclaimed(other) && other.status === 'open' + && other.queueRole && sameRole(other.queueRole, role)) + if (queued.length >= cap) throw refuse('queue-full', `the ${role} queue already holds ${queued.length} items, its cap is ${cap}`) task.queueRole = role task.queueBy = botId + task.queueCap = cap delete task.retryAt delete task.queueReason } diff --git a/core/src/host/tools.js b/core/src/host/tools.js index 51f6f3ec5..1d5f6b994 100644 --- a/core/src/host/tools.js +++ b/core/src/host/tools.js @@ -930,6 +930,7 @@ ${task.writeBack.body}`) assignee: { type: 'string', description: 'Name of the Bot that owns this subtask. Only the room\'s task lead may name another member. The card is claimed, and that Bot starts the work without waiting for a confirmation.' }, queue: { type: 'boolean', description: 'Enqueue this item for a role instead of naming an assignee. Only the room\'s task lead may enqueue. Pass role with it: the item belongs to the room until a member of that role claims it, and enqueuing wakes one idle member of that role.' }, role: { type: 'string', description: 'The role a queued item is for: the Bot role label, matched case-insensitively. It is not a group membership role.' }, + cap: { type: 'integer', description: 'The role queue ceiling this graph runs with. Omit for the default of 8. An enqueue over it is refused, naming the role and the cap; re-offering an item already in the queue is counted once.' }, ask: { type: 'string', description: 'The question you need before this task can continue. The task waits. Call ask_user in your own chat. For a group card, post one short question in the group only when a person has to decide.' }, answer: { type: 'string', description: 'What the user or the group answered. Pass it to resume a task that was waiting.' }, target: { type: 'string', description: 'Where a decision on this task is written back: a GitHub issue or pull request, as https://github.com/owner/repo/issues/12 or owner/repo#12. Optional.' }, diff --git a/core/test/task-queue-lease.test.js b/core/test/task-queue-lease.test.js new file mode 100644 index 000000000..b664851ac --- /dev/null +++ b/core/test/task-queue-lease.test.js @@ -0,0 +1,219 @@ +// Role queues, isolated workers and lease recovery (#1729). Stage 3.2 adds a stated per-role +// ceiling, one worker per claim that no other claim can read, and one owner for an expired lease +// that recovers a card exactly once and records the recovery on the card. This file pins those +// four behaviors through the host surface the workflow plugin reads. +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createRuntime } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +// The room's roles are the Bot labels (`roster.js`: free text, matched case-insensitively — +// `Dev2` below carries `role: 'dev'` so a case-sensitive match would leave it in the queue). +const team = () => ({ + version: 2, + revision: 1, + mainBotIds: ['a'], + members: { 'owner-1': { id: 'owner-1', createdAt: 1, designatedChiefId: 'a' } }, + bots: { + a: { id: 'a', name: 'Chief', role: 'Lead', sessionId: 'a', createdAt: 1, ownerId: 'owner-1' }, + d1: { id: 'd1', name: 'Dev1', role: 'Dev', sessionId: 'd1', createdAt: 2, ownerId: 'owner-1', createdBy: 'a' }, + d2: { id: 'd2', name: 'Dev2', role: 'dev', sessionId: 'd2', createdAt: 3, ownerId: 'owner-1', createdBy: 'a' }, + r1: { id: 'r1', name: 'Rev1', role: 'Reviewer', sessionId: 'r1', createdAt: 4, ownerId: 'owner-1', createdBy: 'a' }, + }, + rooms: { + war: { + id: 'war', name: 'War', members: ['a', 'd1', 'd2', 'r1'], admin: 'a', mode: 'everyone', named: true, + sessions: {}, log: [], seq: 0, seen: {}, + participants: [ + { kind: 'human', id: 'owner-1', role: 'member', displayName: 'Alice', joinedSeq: 0, joinedAt: 1 }, + { kind: 'bot', id: 'a', role: 'admin', displayName: 'Chief', joinedSeq: 0, joinedAt: 1 }, + { kind: 'bot', id: 'd1', role: 'member', displayName: 'Dev1', joinedSeq: 0, joinedAt: 1 }, + { kind: 'bot', id: 'd2', role: 'member', displayName: 'Dev2', joinedSeq: 0, joinedAt: 1 }, + { kind: 'bot', id: 'r1', role: 'member', displayName: 'Rev1', joinedSeq: 0, joinedAt: 1 }, + ], + }, + }, + grants: {}, + agents: {}, exchanges: [], exchangeTurns: {}, questions: {}, prefs: {}, +}) + +async function start(t) { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-queue-')) + writeFileSync(join(home, 'state.json'), JSON.stringify(team())) + const ctx = fakeContext() + const agents = new Map() + const agentOf = (sessionId) => { + if (!agents.has(sessionId)) { + const inbox = { + nextStep: [], + nextTurn: [], + remove(id) { + const keep = message => message.id !== id + const before = this.nextTurn.length + this.nextStep.length + this.nextTurn = this.nextTurn.filter(keep) + this.nextStep = this.nextStep.filter(keep) + return this.nextTurn.length + this.nextStep.length !== before + }, + } + agents.set(sessionId, { + session: { id: sessionId }, id: sessionId, calls: [], status: 'idle', inbox, + followup(message) { this.calls.push(message); this.inbox.nextTurn.push(message) }, + steer() {}, inject() {}, cancel() {}, async whenIdle() {}, + }) + } + return agents.get(sessionId) + } + let created = 0 + Object.assign(ctx, { + llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) }, + workspaceRegistry: { get: () => undefined, resolveByPath: async () => undefined, create: async path => ({ id: 'w1', path }), archiveSession: async () => {} }, + sessionController: { + create: async () => ({ sessionId: `made-${++created}` }), + resolveAgent: async id => ({ agent: agentOf(id) }), + selectModel: async () => {}, + page: async () => ({ records: [], hasMore: false }), + }, + sessionTitle: { rename: async () => {} }, + sessions: { flush: async () => {} }, + }) + const rt = createRuntime(ctx, { home }) + await rt.load() + t.after(async () => { + await ctx.dispose() + rmSync(home, { recursive: true, force: true, maxRetries: 5 }) + }) + const track = (sessionId, args) => ctx.registeredTools.get('track_tasks').execute(args ?? {}, { agent: { id: sessionId } }) + const text = (sessionId) => (agents.get(sessionId)?.calls ?? []).map(message => message.content.map(block => block.text ?? '').join('\n')).join('\n') + const calls = (sessionId) => agents.get(sessionId)?.calls ?? [] + const emit = (sessionId, type, data) => ctx.emitArgs('session/event', { id: sessionId }, { type, data, time: Date.now() }) + const idOf = (body, title) => { + const line = body.split('\n').find(item => item.startsWith('- ') && item.includes(title)) + const id = line?.match(/task-[0-9a-f]+/)?.[0] + assert.ok(id, body) + return id + } + const enqueue = async (spec = {}) => { + const body = await track('a', { tasks: [{ group: 'War', queue: true, ...spec }] }) + return { body, id: idOf(body, spec.title) } + } + // DSH claims the queued follow-up when the turn starts, so the fake inbox drops it too. The turn + // that ends here is a member's own work: the next pass sees a member with a card and no wake. + const finishWake = async (sessionId) => { + const message = calls(sessionId).at(-1) + assert.ok(message, `${sessionId} has a wake to consume`) + agents.get(sessionId).inbox.remove(message.id) + emit(sessionId, 'user/message', { content: message.content, source: message.source }) + emit(sessionId, 'turn/start', { turn: 1 }) + emit(sessionId, 'turn/end', { reason: { kind: 'completed' } }) + await ctx.taskDrive.settled() + } + return { ctx, rt, track, text, calls, emit, idOf, enqueue, finishWake, home } +} + + + +test('a queue over the role cap is refused, naming the role and the cap (#1729)', async (t) => { + const app = await start(t) + const enqueueError = async (args) => { + try { return String(await app.track('a', { tasks: [args] })) } catch (error) { return String(error?.message ?? error) } + } + for (let n = 0; n < 8; n += 1) await app.enqueue({ title: `Queue fill ${n}`, role: 'Dev' }) + const refused = await enqueueError({ group: 'War', queue: true, title: 'One too many', role: 'Dev' }) + assert.match(refused, /queue-full/) + assert.match(refused, /Dev/) + assert.match(refused, /cap is 8/) + // A graph that knows how many workers it runs may state a smaller ceiling on the card. + const small = await start(t) + await small.enqueue({ title: 'A', role: 'Dev', cap: 2 }) + await small.enqueue({ title: 'B', role: 'Dev', cap: 2 }) + const over = String(await small.track('a', { tasks: [{ group: 'War', queue: true, title: 'C', role: 'Dev', cap: 2 }] })) + assert.match(over, /queue-full/) + assert.match(over, /cap is 2/) +}) + +test('re-offering a queued item keeps its id and is never its own refusal (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev', cap: 1 }) + const again = app.rt.state.tasks[id] + const body = await app.track('a', { tasks: [{ id, group: 'War', queue: true, role: 'Dev', title: 'Fix the flaky test' }] }) + assert.match(body, new RegExp(id)) + assert.equal(app.rt.state.tasks[id].id, id) + assert.equal(app.rt.state.tasks[id].queueCap, 1) + assert.equal(again.claimedBy, undefined) +}) + +test('two concurrent claims run in workers that share no state (#1729)', async (t) => { + const app = await start(t) + const one = await app.enqueue({ title: 'First', role: 'Dev' }) + const two = await app.enqueue({ title: 'Second', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id: one.id }) + await app.track('d2', { claim: true, group: 'War', id: two.id }) + const w1 = app.rt.state.tasks[one.id].worker + const w2 = app.rt.state.tasks[two.id].worker + assert.ok(w1 && w2, 'each claim names a worker') + assert.notEqual(w1, w2, 'two claims never share one worker') + app.rt.taskWorkerState(w1).vars.set('scratch', 'first') + app.rt.taskWorkerState(w1).files.set('tmp.txt', 'first') + assert.equal(app.rt.taskWorkerState(w2).vars.get('scratch'), undefined) + assert.equal(app.rt.taskWorkerState(w2).files.get('tmp.txt'), undefined) +}) + +test('two expired reads requeue the card once and record the recovery (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Lease me', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id }) + const task = app.rt.state.tasks[id] + task.lease = { ...task.lease, expiresAt: Date.now() - 1000 } + assert.equal(await app.rt.taskExpireLeases(), 1) + assert.equal(await app.rt.taskExpireLeases(), 0, 'the second read finds nothing to recover') + assert.equal(app.rt.state.tasks[id].status, 'open') + assert.equal(app.rt.state.tasks[id].lease, undefined) + assert.equal(app.rt.state.tasks[id].worker, undefined, 'the dead worker is dropped') + assert.equal(app.rt.taskWorkerState(task.worker), undefined) + const evidence = app.rt.state.tasks[id].evidence ?? [] + assert.equal(evidence.length, 1) + assert.match(evidence[0].check, /lease expired/) + assert.match(evidence[0].check, /Dev1/, 'the recovery names the holder it recovered from') +}) + +test('a claim on a live lease is refused and names the holder (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Held', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id }) + const refused = await app.track('d2', { claim: true, group: 'War', id }) + assert.match(refused, /lease-held/) + assert.match(refused, /Dev1/) + assert.equal(app.rt.state.tasks[id].claimedBy, 'd1') + // The host reads that same lease for the workflow plugin. + assert.equal(app.rt.taskLease(id).live, true) + assert.equal(app.rt.taskLease(id).heldBy, 'Dev1') +}) + +test('a restart with a live lease changes nothing (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Still running', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id }) + const before = app.rt.state.tasks[id] + assert.equal(await app.rt.taskExpireLeases(), 0) + assert.equal(app.rt.state.tasks[id].status, 'claimed') + assert.equal(app.rt.state.tasks[id].claimedBy, 'd1') + assert.equal(app.rt.state.tasks[id].lease.expiresAt, before.lease.expiresAt) + assert.deepEqual(app.rt.state.tasks[id].evidence ?? [], []) +}) + +test('deliveries and held resources are counted from the rows, per role (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Count me', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id }) + await app.track('d1', { tasks: [{ id, column: 'done', progress: 100 }] }) + const counts = app.rt.taskCounters() + assert.equal(counts.roles.Dev.queued, 0) + assert.equal(counts.roles.Dev.accepted, 1) + assert.equal(counts.roles.Dev.deliveries, 1) + assert.ok(counts.roles.Dev.workerSeconds >= 0) + const onlyDev = app.rt.taskCounters({ role: 'Dev' }) + assert.deepEqual(Object.keys(onlyDev.roles), ['Dev']) +}) diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index 9da705b84..71b59d186 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -584,3 +584,13 @@ copy of the rule `tasks` itself answers with, so the board cannot paint a card t refused there; a board with nothing to paint is still `[]`, not an error. `project-board` is listed in `authz.js`'s `TASK_ENDPOINTS`, so a signed-in deployment does not refuse it as an unknown endpoint. + +### Role queues, isolated workers and lease recovery (#1729) + +Stage 3.2 of the task pool. One queue per role, one worker per claim, one owner for an expired lease; all of it reads the cards already on the board, so there is no second store. + +- **Per-role cap.** `ROLE_QUEUE_CAP` (8) is the default ceiling for one role's queue in one group. A graph states its own through `cap` on the enqueued item, kept on the card as `queueCap`; a re-offer keeps the card's id and is never counted against itself. An enqueue over the ceiling is refused `queue-full` with a sentence naming the role and the cap (`the Dev queue already holds 8 items, its cap is 8`), so the caller knows which queue is full rather than seeing a silent null. +- **Isolated workers.** A claim creates a worker (`worker-`, kept in `tasks.js`'s own map, capped at 64, oldest dropped), and the card names it in `worker`. `rt.taskWorkerState(id)` is the only read: an unknown id answers `undefined`, so a second claim cannot see the first's `files` or `vars`. `requeue` and an expired lease drop the worker, so the next claim starts clean. +- **One owner for an expired lease.** `expireLeases` reads the lease once, clears it, then returns a queued card to its role queue — a second read in the same tick finds no lease and answers 0. The recovery is written to the card's `evidence` (`lease expired (held by )`) within the existing eight-entry cap, the dead worker is dropped, and a card that is not a queued item only loses the dead lease. A live lease is untouched, so a restart with a live lease is a no-op; a claim on a live lease is refused `lease-held` naming the holder. `rt.taskExpireLeases` exposes the same read, and `rt.taskLease(id)` gives `{ holder, takenAt, expiresAt, live, heldBy }`; the workflow plugin reads the lease through `fb.tasks.lease(id)` and never keeps a copy. +- **Measured deliveries.** `rt.taskCounters({ role?, run? })` counts `queued`, `claimed`, `working`, `blocked`, `accepted`, `failed`, `cancelled` per role and per run id from the rows, plus `deliveries` (accepted cards) and `workerSeconds` (accepted cards with a lease, measured from `takenAt` to the card's last update). `fb.tasks.counters(query)` is the plugin seam. No metric store is added. +- Tests: `core/test/task-queue-lease.test.js` — the cap refusal names role and cap, a re-offer keeps its id, two concurrent claims share no worker state, two expired reads requeue once and record the holder, a live lease refuses a claim naming the holder, a restart with a live lease changes nothing, and the counters read from the rows. diff --git a/docs/subsystems/workflow.md b/docs/subsystems/workflow.md index 82c8f75a6..69912a805 100644 --- a/docs/subsystems/workflow.md +++ b/docs/subsystems/workflow.md @@ -73,3 +73,5 @@ Core never reads `$DSH_HOME/bot/flybot-workflow`. The plugin registers `flybot.t ## Connector ceiling (#1544) A connector a Bot switched off is not reopened by a Workflow, whichever surface started the run. A run records the Bots whose switches cap it, both in `cappedBy`: the Bot whose `run_workflow` call started it, and the Bot that saved the graph (`authoredBy` on the workflow record, set by `define_workflow` and by any save that names a Bot). A run Bot carries no `createdBy` (#1582): the roster reads that field as "a Bot made this Bot" and would start the run Bot on its maker's model, so the caller travels in `cappedBy` only — a ceiling, not a parent. Every node's run Bot carries `cappedBy` — the caller, else the graph's author — and `core/src/host/connector-access.js` intersects the node's `tools` with those switches, so a node that names a connector its caller or author turned off is refused with `turned off for ; the step's tools list cannot turn it back on`, and the tool is masked out of the run Bot's list too. A run nobody's tool call started (the board's `ingest`, the canvas, the `workflow/*` API) has no caller and is capped by the graph's author alone; a save with no Bot behind it records no author, which is the owner's own edit and leaves no ceiling; a graph saved before this has none either. `visibleTo` on `run_workflow`/`define_workflow` is unchanged: the ceiling is the fix, not limiting who may define a Workflow. +- **Strict order (#1729).** A graph may carry `strict: true` (normalized to a boolean on the record). A strict graph starts one node at a time in declared order: `startFrontier` keeps only the first ready node running and leaves its ready siblings `queued`, so no parallel fan-out happens even when two nodes share a rank. They start when the node before them finishes, in list order. The default is unchanged — a graph without the flag still fans out — and the release-shaped delivery template passes unchanged. Tests: `plugins/workflow/test/strict-order.test.js`. +- **Lease and delivery reads go through the host (#1729).** The plugin reads a card's lease with `fb.tasks.lease(taskId)` (`{ holder, takenAt, expiresAt, live, heldBy }`, or `undefined`) and the per-role/per-run delivery and resource counts with `fb.tasks.counters({ role, run })`. Both are host reads of the existing task rows (`rt.taskLease`, `rt.taskCounters`); the plugin keeps no copy, so an expired lease looks expired to both sides at once. diff --git a/patches/ds-bot/0017-task-pool-role-queues.patch b/patches/ds-bot/0017-task-pool-role-queues.patch new file mode 100644 index 000000000..be47a216f --- /dev/null +++ b/patches/ds-bot/0017-task-pool-role-queues.patch @@ -0,0 +1,189 @@ +FlyBot patch 0017-task-pool-role-queues on Fei2411/DeepSeek-Bot (ds-bot/). Generated by scripts/ds-bot-patches.mjs; edits to upstream files only (files FlyBot adds are in files/). + +diff --git a/src/host/tasks.js b/src/host/tasks.js +--- a/src/host/tasks.js ++++ b/src/host/tasks.js +@@ -47,6 +47,20 @@ + const ROLE_MAX = 24 + const QUEUE_RETRY_DEFAULT_SEC = 60 + const QUEUE_RETRY_MAX_SEC = 7 * 24 * 60 * 60 ++// One queue per role, with a stated ceiling (#1729). The default lives here; a graph that knows ++// how many workers a role runs overrides it on the card it enqueues, and a re-offer keeps it. ++const ROLE_QUEUE_CAP = 8 ++const QUEUE_CAP_MAX = 64 ++// A claim runs in its own worker, and two claims never share one (#1729). Its state is a pair of ++// maps the caller writes; nothing carries over from the previous claim's worker, and returning the ++// card to the queue drops the worker so the next claimant starts clean. ++const WORKERS_KEPT = 64 ++const readCap = (value) => { ++ if (value === undefined || value === null || value === '') return undefined ++ const cap = Number(value) ++ if (!Number.isInteger(cap) || cap < 1 || cap > QUEUE_CAP_MAX) throw invalid(`A queue cap must be a whole number from 1 to ${QUEUE_CAP_MAX}`) ++ return cap ++} + const readRole = (value) => { + if (value === undefined || value === null) return undefined + const role = oneLine(value, ROLE_MAX) +@@ -480,6 +494,63 @@ + task.taskClass = 'work' + return task + } ++ // ---------------------------------------------------------------- isolated workers (#1729) ++ const workers = new Map() ++ function newWorker() { ++ const id = `worker-${randomBytes(6).toString('hex')}` ++ workers.set(id, { id, files: new Map(), vars: new Map(), claimedAt: Date.now() }) ++ while (workers.size > WORKERS_KEPT) workers.delete(workers.keys().next().value) ++ return id ++ } ++ const dropWorker = (workerId) => { if (typeof workerId === 'string') workers.delete(workerId) } ++ // The worker of one claim, or nothing: a caller reaches its own claim's maps only. ++ rt.taskWorkerState = (workerId) => { ++ const worker = workers.get(String(workerId ?? '')) ++ return worker === undefined ? undefined : { id: worker.id, files: worker.files, vars: worker.vars, claimedAt: worker.claimedAt } ++ } ++ // Measured deliveries from the rows themselves (#1729): no second store, the numbers are a ++ // read of the cards. `roles` counts per role, `runs` per run id a card carries. ++ rt.taskCounters = function taskCounters(query = {}) { ++ const roleOf = (task) => { ++ const role = oneLine(task?.queueRole ?? '', ROLE_MAX) ++ if (role !== '') return role ++ return oneLine(botRecord(task?.ownerBotId)?.role ?? '', ROLE_MAX) ++ } ++ const init = () => ({ queued: 0, claimed: 0, working: 0, blocked: 0, accepted: 0, failed: 0, cancelled: 0, deliveries: 0, workerSeconds: 0 }) ++ const roles = {} ++ const runs = {} ++ const wanted = oneLine(query?.role ?? '', ROLE_MAX) ++ const wantedRun = query?.run === undefined || query?.run === null ? '' : String(query.run) ++ for (const task of Object.values(store())) { ++ const role = roleOf(task) ++ const run = String(task?.origin?.runId ?? task?.runId ?? '') ++ if (wanted !== '' && role !== wanted) continue ++ if (wantedRun !== '' && run !== wantedRun) continue ++ const bucket = (map, key) => (map[key] ??= init()) ++ const row = bucket(roles, role || 'unassigned') ++ if (row[task.status] !== undefined) row[task.status] += 1 ++ if (task.status === 'accepted') { ++ row.deliveries += 1 ++ const lease = leaseOf(task) ++ if (lease !== undefined && Number.isFinite(lease.takenAt)) row.workerSeconds += Math.max(0, Math.round((task.updatedAt - lease.takenAt) / 1000)) ++ } ++ if (run !== '') { ++ const runRow = bucket(runs, run) ++ if (runRow[task.status] !== undefined) runRow[task.status] += 1 ++ if (task.status === 'accepted') runRow.deliveries += 1 ++ } ++ } ++ return { roles, runs } ++ } ++ // The workflow plugin reads a card's lease through the host (#1729), never a copy of its own. ++ rt.taskLease = function taskLease(id) { ++ const task = find(String(id ?? '')) ++ if (task === undefined) return undefined ++ const lease = leaseOf(task) ++ if (lease === undefined) return undefined ++ const now = Date.now() ++ return { ...lease, live: lease.expiresAt > now, heldBy: holderName(lease.holder) } ++ } + const leaseOf = task => (task?.lease !== null && typeof task?.lease === 'object' ? task.lease : undefined) + const leaseLive = (task, now = Date.now()) => { + const lease = leaseOf(task) +@@ -494,16 +565,28 @@ + for (const task of Object.values(store())) { + const lease = leaseOf(task) + if (lease === undefined || !Number.isFinite(lease.expiresAt) || lease.expiresAt > now) continue +- if (!task.queueRole || (task.status !== 'claimed' && task.status !== 'working')) { +- delete task.lease +- continue +- } +- requeue(task, 'lease expired', 0) ++ // Read once and clear before the requeue: a second read in the same tick sees no lease, so ++ // two expired reads return the card once, not twice (#1729). The recovery is recorded on the ++ // card itself, and the dead worker is dropped so the next claim starts with no state of it. ++ const heldBy = holderName(lease.holder) + delete task.lease ++ dropWorker(task.worker) ++ delete task.worker ++ if (!task.queueRole || (task.status !== 'claimed' && task.status !== 'working')) continue ++ const entry = { check: `lease expired (held by ${heldBy})`, at: now } ++ task.evidence = [...(Array.isArray(task.evidence) ? task.evidence : []), entry].slice(-EVIDENCE_MAX) ++ requeue(task, 'lease expired', 0) + expired += 1 + } + return expired + } ++ // The one owner of lease recovery (#1729): a read here is the same read the claim path makes, ++ // so a restart and a claim cannot both return the same expired card. ++ rt.taskExpireLeases = async (now = Date.now()) => { ++ const expired = expireLeases(now) ++ if (expired > 0) await save() ++ return expired ++ } + // A rejection is a handover, not a status flip (#1720): the original's `handoff` names the + // roles, and the rework is one child card linked by `reworkOf` — never `parentId`, which + // means "part of an overall card" and must never make `settleOverall` fire. A replay of the +@@ -642,6 +725,8 @@ + // blocked → open is the queue's own edge: a member that cannot finish returns the item, and + // no other path sends a card back to the queue, so the edge is not on TRANSITIONS. + function requeue(task, reason, retryInSec) { ++ dropWorker(task.worker) ++ delete task.worker + delete task.waitingFor + delete task.inputRequest + delete task.inputApprovalIds +@@ -828,6 +913,8 @@ + delete live.queueReason + const claimedAt = Date.now() + live.lease = { holder: botId, takenAt: claimedAt, expiresAt: claimedAt + LEASE_TTL_SEC * 1000 } ++ // Every claim gets a worker nobody else has started (#1729). ++ live.worker = newWorker() + transition(live, 'claimed', null, { by: 'claimed from the queue' }) + announceClaim(live) + await save() +@@ -972,8 +1059,17 @@ + if (overall) task.kind = 'group' + if (queue) { + if (task.claimedBy) throw invalid('That queued item is already claimed; enqueue a new one') ++ // A graph may state the role's ceiling; otherwise the one constant does (#1729). The ++ // count leaves this card out, so re-offering a queued item keeps its id and is never ++ // refused for the place it already holds. ++ const cap = readCap(item.cap) ?? (Number.isInteger(task.queueCap) ? task.queueCap : ROLE_QUEUE_CAP) ++ const queued = Object.values(store()).filter(other => other.id !== task.id ++ && other.origin?.roomId === roomId && unclaimed(other) && other.status === 'open' ++ && other.queueRole && sameRole(other.queueRole, role)) ++ if (queued.length >= cap) throw refuse('queue-full', `the ${role} queue already holds ${queued.length} items, its cap is ${cap}`) + task.queueRole = role + task.queueBy = botId ++ task.queueCap = cap + delete task.retryAt + delete task.queueReason + } +diff --git a/src/host/tools.js b/src/host/tools.js +--- a/src/host/tools.js ++++ b/src/host/tools.js +@@ -930,6 +930,7 @@ + assignee: { type: 'string', description: 'Name of the Bot that owns this subtask. Only the room\'s task lead may name another member. The card is claimed, and that Bot starts the work without waiting for a confirmation.' }, + queue: { type: 'boolean', description: 'Enqueue this item for a role instead of naming an assignee. Only the room\'s task lead may enqueue. Pass role with it: the item belongs to the room until a member of that role claims it, and enqueuing wakes one idle member of that role.' }, + role: { type: 'string', description: 'The role a queued item is for: the Bot role label, matched case-insensitively. It is not a group membership role.' }, ++ cap: { type: 'integer', description: 'The role queue ceiling this graph runs with. Omit for the default of 8. An enqueue over it is refused, naming the role and the cap; re-offering an item already in the queue is counted once.' }, + ask: { type: 'string', description: 'The question you need before this task can continue. The task waits. Call ask_user in your own chat. For a group card, post one short question in the group only when a person has to decide.' }, + answer: { type: 'string', description: 'What the user or the group answered. Pass it to resume a task that was waiting.' }, + target: { type: 'string', description: 'Where a decision on this task is written back: a GitHub issue or pull request, as https://github.com/owner/repo/issues/12 or owner/repo#12. Optional.' }, +diff --git a/src/host/hooks.js b/src/host/hooks.js +--- a/src/host/hooks.js ++++ b/src/host/hooks.js +@@ -1336,6 +1336,14 @@ + onProjectProvider(fn) { + return rt.onProjectProvider?.(fn) ?? (() => {}) + }, ++ // #1729: a workflow reads a card's lease, and the delivery/resource counts, through the ++ // host — never a copy of its own, so the two cannot disagree about who holds a card. ++ lease(taskId) { ++ return rt.taskLease?.(taskId) ++ }, ++ counters(query) { ++ return rt.taskCounters?.(query) ?? { roles: {}, runs: {} } ++ }, + }, + tools, + api, diff --git a/patches/ds-bot/files/test/task-queue-lease.test.js b/patches/ds-bot/files/test/task-queue-lease.test.js new file mode 100644 index 000000000..b664851ac --- /dev/null +++ b/patches/ds-bot/files/test/task-queue-lease.test.js @@ -0,0 +1,219 @@ +// Role queues, isolated workers and lease recovery (#1729). Stage 3.2 adds a stated per-role +// ceiling, one worker per claim that no other claim can read, and one owner for an expired lease +// that recovers a card exactly once and records the recovery on the card. This file pins those +// four behaviors through the host surface the workflow plugin reads. +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createRuntime } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +// The room's roles are the Bot labels (`roster.js`: free text, matched case-insensitively — +// `Dev2` below carries `role: 'dev'` so a case-sensitive match would leave it in the queue). +const team = () => ({ + version: 2, + revision: 1, + mainBotIds: ['a'], + members: { 'owner-1': { id: 'owner-1', createdAt: 1, designatedChiefId: 'a' } }, + bots: { + a: { id: 'a', name: 'Chief', role: 'Lead', sessionId: 'a', createdAt: 1, ownerId: 'owner-1' }, + d1: { id: 'd1', name: 'Dev1', role: 'Dev', sessionId: 'd1', createdAt: 2, ownerId: 'owner-1', createdBy: 'a' }, + d2: { id: 'd2', name: 'Dev2', role: 'dev', sessionId: 'd2', createdAt: 3, ownerId: 'owner-1', createdBy: 'a' }, + r1: { id: 'r1', name: 'Rev1', role: 'Reviewer', sessionId: 'r1', createdAt: 4, ownerId: 'owner-1', createdBy: 'a' }, + }, + rooms: { + war: { + id: 'war', name: 'War', members: ['a', 'd1', 'd2', 'r1'], admin: 'a', mode: 'everyone', named: true, + sessions: {}, log: [], seq: 0, seen: {}, + participants: [ + { kind: 'human', id: 'owner-1', role: 'member', displayName: 'Alice', joinedSeq: 0, joinedAt: 1 }, + { kind: 'bot', id: 'a', role: 'admin', displayName: 'Chief', joinedSeq: 0, joinedAt: 1 }, + { kind: 'bot', id: 'd1', role: 'member', displayName: 'Dev1', joinedSeq: 0, joinedAt: 1 }, + { kind: 'bot', id: 'd2', role: 'member', displayName: 'Dev2', joinedSeq: 0, joinedAt: 1 }, + { kind: 'bot', id: 'r1', role: 'member', displayName: 'Rev1', joinedSeq: 0, joinedAt: 1 }, + ], + }, + }, + grants: {}, + agents: {}, exchanges: [], exchangeTurns: {}, questions: {}, prefs: {}, +}) + +async function start(t) { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-queue-')) + writeFileSync(join(home, 'state.json'), JSON.stringify(team())) + const ctx = fakeContext() + const agents = new Map() + const agentOf = (sessionId) => { + if (!agents.has(sessionId)) { + const inbox = { + nextStep: [], + nextTurn: [], + remove(id) { + const keep = message => message.id !== id + const before = this.nextTurn.length + this.nextStep.length + this.nextTurn = this.nextTurn.filter(keep) + this.nextStep = this.nextStep.filter(keep) + return this.nextTurn.length + this.nextStep.length !== before + }, + } + agents.set(sessionId, { + session: { id: sessionId }, id: sessionId, calls: [], status: 'idle', inbox, + followup(message) { this.calls.push(message); this.inbox.nextTurn.push(message) }, + steer() {}, inject() {}, cancel() {}, async whenIdle() {}, + }) + } + return agents.get(sessionId) + } + let created = 0 + Object.assign(ctx, { + llm: { listProviders: () => [], listModels: async () => [], resolveModelInfo: async () => ({}) }, + workspaceRegistry: { get: () => undefined, resolveByPath: async () => undefined, create: async path => ({ id: 'w1', path }), archiveSession: async () => {} }, + sessionController: { + create: async () => ({ sessionId: `made-${++created}` }), + resolveAgent: async id => ({ agent: agentOf(id) }), + selectModel: async () => {}, + page: async () => ({ records: [], hasMore: false }), + }, + sessionTitle: { rename: async () => {} }, + sessions: { flush: async () => {} }, + }) + const rt = createRuntime(ctx, { home }) + await rt.load() + t.after(async () => { + await ctx.dispose() + rmSync(home, { recursive: true, force: true, maxRetries: 5 }) + }) + const track = (sessionId, args) => ctx.registeredTools.get('track_tasks').execute(args ?? {}, { agent: { id: sessionId } }) + const text = (sessionId) => (agents.get(sessionId)?.calls ?? []).map(message => message.content.map(block => block.text ?? '').join('\n')).join('\n') + const calls = (sessionId) => agents.get(sessionId)?.calls ?? [] + const emit = (sessionId, type, data) => ctx.emitArgs('session/event', { id: sessionId }, { type, data, time: Date.now() }) + const idOf = (body, title) => { + const line = body.split('\n').find(item => item.startsWith('- ') && item.includes(title)) + const id = line?.match(/task-[0-9a-f]+/)?.[0] + assert.ok(id, body) + return id + } + const enqueue = async (spec = {}) => { + const body = await track('a', { tasks: [{ group: 'War', queue: true, ...spec }] }) + return { body, id: idOf(body, spec.title) } + } + // DSH claims the queued follow-up when the turn starts, so the fake inbox drops it too. The turn + // that ends here is a member's own work: the next pass sees a member with a card and no wake. + const finishWake = async (sessionId) => { + const message = calls(sessionId).at(-1) + assert.ok(message, `${sessionId} has a wake to consume`) + agents.get(sessionId).inbox.remove(message.id) + emit(sessionId, 'user/message', { content: message.content, source: message.source }) + emit(sessionId, 'turn/start', { turn: 1 }) + emit(sessionId, 'turn/end', { reason: { kind: 'completed' } }) + await ctx.taskDrive.settled() + } + return { ctx, rt, track, text, calls, emit, idOf, enqueue, finishWake, home } +} + + + +test('a queue over the role cap is refused, naming the role and the cap (#1729)', async (t) => { + const app = await start(t) + const enqueueError = async (args) => { + try { return String(await app.track('a', { tasks: [args] })) } catch (error) { return String(error?.message ?? error) } + } + for (let n = 0; n < 8; n += 1) await app.enqueue({ title: `Queue fill ${n}`, role: 'Dev' }) + const refused = await enqueueError({ group: 'War', queue: true, title: 'One too many', role: 'Dev' }) + assert.match(refused, /queue-full/) + assert.match(refused, /Dev/) + assert.match(refused, /cap is 8/) + // A graph that knows how many workers it runs may state a smaller ceiling on the card. + const small = await start(t) + await small.enqueue({ title: 'A', role: 'Dev', cap: 2 }) + await small.enqueue({ title: 'B', role: 'Dev', cap: 2 }) + const over = String(await small.track('a', { tasks: [{ group: 'War', queue: true, title: 'C', role: 'Dev', cap: 2 }] })) + assert.match(over, /queue-full/) + assert.match(over, /cap is 2/) +}) + +test('re-offering a queued item keeps its id and is never its own refusal (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev', cap: 1 }) + const again = app.rt.state.tasks[id] + const body = await app.track('a', { tasks: [{ id, group: 'War', queue: true, role: 'Dev', title: 'Fix the flaky test' }] }) + assert.match(body, new RegExp(id)) + assert.equal(app.rt.state.tasks[id].id, id) + assert.equal(app.rt.state.tasks[id].queueCap, 1) + assert.equal(again.claimedBy, undefined) +}) + +test('two concurrent claims run in workers that share no state (#1729)', async (t) => { + const app = await start(t) + const one = await app.enqueue({ title: 'First', role: 'Dev' }) + const two = await app.enqueue({ title: 'Second', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id: one.id }) + await app.track('d2', { claim: true, group: 'War', id: two.id }) + const w1 = app.rt.state.tasks[one.id].worker + const w2 = app.rt.state.tasks[two.id].worker + assert.ok(w1 && w2, 'each claim names a worker') + assert.notEqual(w1, w2, 'two claims never share one worker') + app.rt.taskWorkerState(w1).vars.set('scratch', 'first') + app.rt.taskWorkerState(w1).files.set('tmp.txt', 'first') + assert.equal(app.rt.taskWorkerState(w2).vars.get('scratch'), undefined) + assert.equal(app.rt.taskWorkerState(w2).files.get('tmp.txt'), undefined) +}) + +test('two expired reads requeue the card once and record the recovery (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Lease me', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id }) + const task = app.rt.state.tasks[id] + task.lease = { ...task.lease, expiresAt: Date.now() - 1000 } + assert.equal(await app.rt.taskExpireLeases(), 1) + assert.equal(await app.rt.taskExpireLeases(), 0, 'the second read finds nothing to recover') + assert.equal(app.rt.state.tasks[id].status, 'open') + assert.equal(app.rt.state.tasks[id].lease, undefined) + assert.equal(app.rt.state.tasks[id].worker, undefined, 'the dead worker is dropped') + assert.equal(app.rt.taskWorkerState(task.worker), undefined) + const evidence = app.rt.state.tasks[id].evidence ?? [] + assert.equal(evidence.length, 1) + assert.match(evidence[0].check, /lease expired/) + assert.match(evidence[0].check, /Dev1/, 'the recovery names the holder it recovered from') +}) + +test('a claim on a live lease is refused and names the holder (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Held', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id }) + const refused = await app.track('d2', { claim: true, group: 'War', id }) + assert.match(refused, /lease-held/) + assert.match(refused, /Dev1/) + assert.equal(app.rt.state.tasks[id].claimedBy, 'd1') + // The host reads that same lease for the workflow plugin. + assert.equal(app.rt.taskLease(id).live, true) + assert.equal(app.rt.taskLease(id).heldBy, 'Dev1') +}) + +test('a restart with a live lease changes nothing (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Still running', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id }) + const before = app.rt.state.tasks[id] + assert.equal(await app.rt.taskExpireLeases(), 0) + assert.equal(app.rt.state.tasks[id].status, 'claimed') + assert.equal(app.rt.state.tasks[id].claimedBy, 'd1') + assert.equal(app.rt.state.tasks[id].lease.expiresAt, before.lease.expiresAt) + assert.deepEqual(app.rt.state.tasks[id].evidence ?? [], []) +}) + +test('deliveries and held resources are counted from the rows, per role (#1729)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Count me', role: 'Dev' }) + await app.track('d1', { claim: true, group: 'War', id }) + await app.track('d1', { tasks: [{ id, column: 'done', progress: 100 }] }) + const counts = app.rt.taskCounters() + assert.equal(counts.roles.Dev.queued, 0) + assert.equal(counts.roles.Dev.accepted, 1) + assert.equal(counts.roles.Dev.deliveries, 1) + assert.ok(counts.roles.Dev.workerSeconds >= 0) + const onlyDev = app.rt.taskCounters({ role: 'Dev' }) + assert.deepEqual(Object.keys(onlyDev.roles), ['Dev']) +}) diff --git a/plugins/workflow/src/graph.js b/plugins/workflow/src/graph.js index 6f34f6591..f6109da70 100644 --- a/plugins/workflow/src/graph.js +++ b/plugins/workflow/src/graph.js @@ -260,7 +260,8 @@ export function normalizeGraph(graph) { } const { layers } = ranksOf(graph.nodes, edges) const nodes = layoutNodes(graph.nodes.map(node => ({ ...node })), layers) - return { ...graph, nodes, edges, layers } + // A graph the operator marked strict runs its nodes in declared order with no fan-out (#1729). + return { ...graph, strict: graph.strict === true, nodes, edges, layers } } export function validateGraph(graph) { @@ -535,6 +536,12 @@ export function startFrontier(graph, status, fired) { next[id] = 'running' started.push(id) } + // Strict order (#1729): one node at a time, in declared order. A sibling that was ready is left + // queued and starts once the one before it finishes. + if (graph.strict === true && started.length > 1) { + for (const id of started.slice(1)) next[id] = 'queued' + started.length = 1 + } const finished = graph.nodes.every(node => settled(next[node.id])) return { status: next, started, finished } } diff --git a/plugins/workflow/test/strict-order.test.js b/plugins/workflow/test/strict-order.test.js new file mode 100644 index 000000000..e957b4de3 --- /dev/null +++ b/plugins/workflow/test/strict-order.test.js @@ -0,0 +1,53 @@ +// Strict order (#1729): a graph the operator marks strict walks its nodes in the declared order +// with no parallel fan-out, and the release-shaped graph keeps working unchanged. +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { completeNode, freshStatus, normalizeGraph, startFrontier } from '../src/graph.js' +import { deliveryWorkflow } from '../src/template.js' + +const parallel = { + nodes: [ + { id: 'planner', name: 'Planner' }, + { id: 'writer', name: 'Writer' }, + { id: 'coder', name: 'Coder' }, + ], + edges: [ + { from: 'planner', to: 'writer' }, + { from: 'planner', to: 'coder' }, + ], +} + +test('a strict graph starts one node at a time in declared order (#1729)', () => { + const graph = normalizeGraph({ ...parallel, strict: true }) + assert.equal(graph.strict, true, 'the option survives normalization') + const first = startFrontier(graph, freshStatus(graph)) + assert.deepEqual(first.started, ['planner'], 'one node, not every root') + assert.equal(first.status.writer, 'queued') + assert.equal(first.status.coder, 'queued') + const afterPlanner = completeNode(graph, first.status, 'planner') + assert.deepEqual(afterPlanner.started, ['writer'], 'declared order decides the next one') + assert.equal(afterPlanner.status.writer, 'running') + assert.equal(afterPlanner.status.coder, 'queued', 'the sibling waits its turn') + const afterWriter = completeNode(graph, afterPlanner.status, 'writer') + assert.deepEqual(afterWriter.started, ['coder']) + const done = completeNode(graph, afterWriter.status, 'coder') + assert.equal(done.finished, true) +}) + +test('the same graph without strict still fans out (#1729)', () => { + const graph = normalizeGraph(parallel) + assert.equal(graph.strict, false) + const first = startFrontier(graph, freshStatus(graph)) + assert.deepEqual(first.started, ['planner']) + const afterPlanner = completeNode(graph, first.status, 'planner') + assert.deepEqual(afterPlanner.started, ['writer', 'coder'], 'the default stays parallel') +}) + +test('the release-shaped delivery graph passes unchanged (#1729)', () => { + const graph = normalizeGraph(deliveryWorkflow()) + assert.equal(graph.strict, false) + const first = startFrontier(graph, freshStatus(graph)) + assert.deepEqual(first.started, ['planner']) + const afterPlanner = completeNode(graph, first.status, 'planner') + assert.ok(afterPlanner.started.length >= 1) +})