From 9864f03730873f3c6f3052a0397a78a0bfb5ad8d Mon Sep 17 00:00:00 2001
From: Danyang Chen <birfychen@gmail.com>
Date: Fri, 30 Oct 2026 03:17:37 +0800
Subject: [PATCH 1/2] Stage 3.2: role queue caps, isolated workers and lease
 recovery (#1729)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A per-role queue cap (8 by default, capped per graph) refuses an enqueue over it
naming the role and the cap, and a re-offer keeps the card's id. Every claim gets
its own worker, and requeueing or an expired lease drops it, so no claim reads
another's state. Expired leases are recovered by one read that records the
recovery on the card's evidence and requeues it exactly once; a live lease still
refuses a claim naming its holder and survives a restart. Deliveries and worker
seconds are counted from the rows, per role and per run. A graph marked strict
starts one node at a time in declared order. The workflow plugin reads the lease
and the counters through the host.

Docs-not-needed: no — docs/subsystems/core-host.md and workflow.md updated.
Written-by: Claim 0f76
---
 changelog.d/1729.added.md                     |   1 +
 core/src/host/hooks.js                        |   8 +
 core/src/host/tasks.js                        | 106 ++++++++-
 core/src/host/tools.js                        |   1 +
 core/test/task-queue-lease.test.js            | 219 ++++++++++++++++++
 docs/subsystems/core-host.md                  |  10 +
 docs/subsystems/workflow.md                   |   2 +
 .../ds-bot/0017-task-pool-role-queues.patch   | 189 +++++++++++++++
 .../files/test/task-queue-lease.test.js       | 219 ++++++++++++++++++
 plugins/workflow/src/graph.js                 |   9 +-
 plugins/workflow/test/strict-order.test.js    |  53 +++++
 11 files changed, 811 insertions(+), 6 deletions(-)
 create mode 100644 changelog.d/1729.added.md
 create mode 100644 core/test/task-queue-lease.test.js
 create mode 100644 patches/ds-bot/0017-task-pool-role-queues.patch
 create mode 100644 patches/ds-bot/files/test/task-queue-lease.test.js
 create mode 100644 plugins/workflow/test/strict-order.test.js

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-<hex>`, 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 <name>)`) 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 <name>; 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)
+})

From a6dc7c94b4b046533da62b0f902432612ccbd86e Mon Sep 17 00:00:00 2001
From: Danyang Chen <birfychen@gmail.com>
Date: Fri, 30 Oct 2026 03:20:29 +0800
Subject: [PATCH 2/2] Stage 3.2: role queue caps, isolated workers and lease
 recovery (#1729)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A per-role queue cap (8 by default, capped per graph) refuses an enqueue over it
naming the role and the cap, and a re-offer keeps the card's id. Every claim gets
its own worker, and requeueing or an expired lease drops it, so no claim reads
another's state. Expired leases are recovered by one read that records the
recovery on the card's evidence and requeues it exactly once; a live lease still
refuses a claim naming its holder and survives a restart. Deliveries and worker
seconds are counted from the rows, per role and per run. A graph marked strict
starts one node at a time in declared order. The workflow plugin reads the lease
and the counters through the host. The patch-series guard now expects the new
0017 entry.

Docs-not-needed: no — docs/subsystems/core-host.md and workflow.md updated.
Written-by: Claim 0f76
---
 .github/scripts/ds-bot-patches.test.mjs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

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 {