Stage 3.2: role queue caps, isolated workers and lease recovery

.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 {

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.

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,

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
         }

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.' },