A lead enqueues work for a role, and an idle member claims it (#1247)

changelog.d/1247-pull-queue.added.md

@@ -0,0 +1 @@
+A group's board can now be a pull queue: the lead enqueues an item for a role (`track_tasks` with `queue` and `role`, no assignee), an idle member of that role claims it with `claim` and one item is handed to exactly one member, and a member that cannot finish returns it with a one-line reason and a `retryAt`; the room board shows one line per role with its depth and oldest wait ([#1247](https://github.com/ThinkFlowLab/FlyBot/issues/1247)).

core/client.js

@@ -8040,6 +8040,16 @@ function compareForWork(a, b) {
   if (byWorking !== 0) return byWorking;
   return (b?.updatedAt ?? 0) - (a?.updatedAt ?? 0);
 }
+function sameRole(a, b) {
+  const left = String(a ?? "").trim().toLowerCase();
+  const right = String(b ?? "").trim().toLowerCase();
+  return left !== "" && left === right;
+}
+function compareForQueue(a, b) {
+  const byPriority = (PRIORITY_RANK[priorityOf(a)] ?? 2) - (PRIORITY_RANK[priorityOf(b)] ?? 2);
+  if (byPriority !== 0) return byPriority;
+  return (a?.createdAt ?? 0) - (b?.createdAt ?? 0);
+}
 function progressOf(task) {
   const value = task?.progress;
   if (typeof value !== "number" || !Number.isFinite(value)) return columnOf(task) === "done" && task?.status !== "cancelled" && task?.status !== "failed" ? 100 : 0;

core/src/host/room-context.js

@@ -134,19 +134,54 @@ export function fitRoomLines(lines, { budget = ROOM_CONTEXT_BUDGET, render, prep
   }
 }
 
-function cardLine(task, nameOf) {
-  const who = oneLine(typeof nameOf === 'function' ? nameOf(task) : '', 40) || 'unassigned'
+// How long an item has waited (#1247). Same shape as the presence age: minutes, then hours,
+// then days — one compact token, because it rides inside the card's bracket.
+const ageLabel = (ms) => {
+  const minutes = Math.max(0, Math.round(ms / 60000))
+  if (minutes < 60) return `${minutes}m`
+  const hours = Math.round(minutes / 60)
+  if (hours < 48) return `${hours}h`
+  return `${Math.round(hours / 24)}d`
+}
+
+// An unclaimed queue item has no holder yet, so the card says `unassigned` and carries which
+// role it is queued for and how long it has waited (#1247) — the same card line the board
+// already had, with the queue's two extra facts in the bracket.
+function cardLine(task, nameOf, now) {
+  const queued = Boolean(task.queueRole) && !task.claimedBy
+  const who = queued ? 'unassigned' : oneLine(typeof nameOf === 'function' ? nameOf(task) : '', 40) || 'unassigned'
   const title = oneLine(task.title || '(untitled)', 80)
   const status = oneLine(task.status || 'open', 24)
   const pct = typeof task.progress === 'number' && Number.isFinite(task.progress) ? ` ${progressOf(task)}%` : ''
   const wait = task.waitingFor === 'input' ? ' waiting' : ''
+  const created = Number.isFinite(task.createdAt) ? task.createdAt : now
+  const queue = queued ? ` ${ageLabel(now - created)} ${oneLine(task.queueRole, 24)}` : ''
   const note = task.note ? ` — ${oneLine(task.note, 80)}` : ''
-  return `${who} [${status}${pct}${wait}] ${title}${note}`
+  return `${who} [${status}${pct}${wait}${queue}] ${title}${note}`
+}
+
+// One line per role that has something waiting: depth and the age of the oldest item, so
+// starvation is visible without reading every card (#1247). A role with nothing waiting is
+// omitted; the count is unclaimed items only, because a claimed one is already a card above.
+function queueLines(open, now) {
+  const roles = new Map()
+  for (const task of open) {
+    if (!task.queueRole || task.claimedBy) continue
+    const role = oneLine(task.queueRole, 24)
+    const key = role.toLowerCase()
+    const entry = roles.get(key) ?? { role, count: 0, oldest: now }
+    entry.count += 1
+    entry.oldest = Math.min(entry.oldest, Number.isFinite(task.createdAt) ? task.createdAt : now)
+    roles.set(key, entry)
+  }
+  return [...roles.values()]
+    .sort((a, b) => a.role.localeCompare(b.role))
+    .map(entry => `- Queue ${entry.role}: ${entry.count} waiting, oldest ${ageLabel(now - entry.oldest)}`)
 }
 
 // Open cards for one room: the newest overall task, then its subtasks and any
 // room card that is not under a different overall task. Done cards stay off.
-export function formatRoomBoard(tasks, { roomId, nameOf, limit = 8 } = {}) {
+export function formatRoomBoard(tasks, { roomId, nameOf, limit = 8, now = Date.now() } = {}) {
   if (!roomId) return ''
   const open = Object.values(tasks ?? {}).filter(task => task && task.origin?.roomId === roomId && !TERMINAL.has(task.status))
   if (open.length === 0) return ''
@@ -160,9 +195,10 @@ export function formatRoomBoard(tasks, { roomId, nameOf, limit = 8 } = {}) {
   }
   if (cards.length === 0) return ''
   const lines = ['Task board (assign, claim, and progress live here; update it with track_tasks, do not restate it in chat):']
-  for (const { task, indent } of cards) lines.push(`${indent}- ${cardLine(task, nameOf)}`)
+  for (const { task, indent } of cards) lines.push(`${indent}- ${cardLine(task, nameOf, now)}`)
   const hidden = open.length - cards.length
   if (hidden > 0) lines.push(`- ${hidden} more open card(s) not shown`)
+  lines.push(...queueLines(open, now))
   return lines.join('\n')
 }
 

core/src/host/task-drive.js

@@ -4,7 +4,7 @@
 // subtask, is a follow-up in that Bot's own chat (#1205). The driver does not
 // open a group Session and does not post the reply into the room. Assigning,
 // progress, and the next assignee are board writes; chat is not the scheduler.
-import { compareForWork } from '../shared/task-board.js'
+import { columnOf, compareForQueue, compareForWork, sameRole } from '../shared/task-board.js'
 import { formatRoomBoard } from './room-context.js'
 
 // A no-op track_tasks call still bumps updatedAt. The hash leaves that stamp out,
@@ -30,6 +30,25 @@ const fingerprint = (tasks) => tasks.map(task => [
   task.writeBack ? `${task.writeBack.state}:${task.writeBack.target}` : '',
 ].join('\u0000')).sort().join('\u0001')
 
+// A queue item is offered, not owned: this cue names the item and the claim, never its note —
+// the note is the body, and reading it is the turn's job, not the wake's (#1247).
+const queueCueFor = (task) => {
+  const title = String(task.title ?? '').replace(/[\r\n]+/g, ' ').trim().slice(0, 200)
+  const priority = task.priority ?? 'normal'
+  return `Take ${task.id} [${priority}] ${title} — queued for ${task.queueRole}. Claim it with track_tasks {claim: true, id: '${task.id}'}, then set its column to in-progress. Do not post a status in the group. Stop when it is done or waiting on an answer. Do not invent other work.`
+}
+
+// What the queue pass compares between passes: an item nobody claimed is still the same offer.
+const queueFingerprint = (task) => (task === undefined ? '' : [
+  task.id,
+  task.status,
+  task.priority ?? 'normal',
+  task.claimedBy ?? '',
+  task.retryAt ?? '',
+  task.note ?? '',
+  task.title ?? '',
+].join('\u0000'))
+
 const cueFor = (task) => {
   const title = String(task.title ?? '').replace(/[\r\n]+/g, ' ').trim().slice(0, 200)
   const priority = task.priority ?? 'normal'
@@ -73,6 +92,10 @@ export function install(rt) {
     if (!bot || bot.remote !== undefined) return []
     return Object.values(rt.state.tasks ?? {}).filter((task) => {
       if (task.ownerBotId !== botId || task.waitingFor === 'input') return false
+      // An item with a role and no claimer belongs to the room, not to this Bot: the enqueuer
+      // must not be woken for its own item (#1247). A claimed one names the member and is
+      // driven above like any other card.
+      if (task.queueRole && !task.claimedBy) return false
       // A decision waiting to be posted is work even when the task is accepted: the Bot holds the
       // connector, so the comment goes out on its next turn (#1154). The reminder stops after two
       // wakes (#1172) — a Bot that never posts must not be woken for ever. The status path below is
@@ -111,6 +134,17 @@ export function install(rt) {
   // say running for that moment; a follow-up queues behind it, so status is not a gate.
   const busy = (sessionId) => rt.liveOf?.(sessionId)?.turn !== undefined
 
+  // WIP = 1, the rule `claimQueued` enforces (#1247): a member already holding a card in progress
+  // cannot take a queued item, and a blocked card waits on a person so it does not count. Waking
+  // that member only spends a turn on the refusal that names its own card, and leaves the item
+  // behind that card, so the offer goes to another member of the role. Same test, one board.
+  const holdsCard = (botId) => Object.values(rt.state.tasks ?? {})
+    .some(task => task.ownerBotId === botId && columnOf(task) === 'in-progress' && task.status !== 'blocked')
+
+  // A wake this Session already has: one the pass counted (`inflight`), or one still waiting in
+  // the inbox — after a restart the count is gone and the message is not (#1247).
+  const wakeWaiting = (sessionId, agent) => inflight.has(sessionId) || taskMessages(agent).length > 0
+
   const dropQueued = (agent) => {
     let kept = false
     for (const message of taskMessages(agent)) {
@@ -128,24 +162,36 @@ export function install(rt) {
     const owner = rt.groupOwners.get(sessionId)
     const botId = owner?.botId ?? rt.selfOf(sessionId)
     const round = owner !== undefined && roundLive(owner.roomId)
-    const still = !round && drivingTasks(botId).some(task => ownsSession(task, sessionId))
+    const queued = job.queue !== undefined && (() => {
+      const task = rt.state.tasks?.[job.queue]
+      return Boolean(task && task.queueRole && !task.claimedBy && task.status === 'open')
+    })()
+    const still = !round && (drivingTasks(botId).some(task => ownsSession(task, sessionId)) || queued)
     if (still) return
     const agent = await rt.resolveAgent(sessionId).catch(() => undefined)
     if (!dropQueued(agent)) return
     inflight.delete(sessionId)
   }
 
-  const noteMiss = (sessionId, fp) => {
+  const noteMiss = (sessionId, fp, kind = 'work') => {
     const count = (misses.get(sessionId) ?? 0) + 1
     misses.set(sessionId, count)
     const delay = TASK_DRIVE_BACKOFF_MS[Math.min(count - 1, TASK_DRIVE_BACKOFF_MS.length - 1)]
-    held.set(sessionId, { until: clock.now() + delay, fingerprint: fp })
+    held.set(sessionId, { until: clock.now() + delay, fingerprint: fp, kind })
   }
 
-  const clearMiss = (sessionId) => {
+  // Only the pass that set the hold may clear it: the queue pass and the work pass compare
+  // different fingerprints, so without the kind each would wipe the other's wait (#1247).
+  const clearMiss = (sessionId, kind) => {
+    const hold = held.get(sessionId)
+    if (kind !== undefined && hold !== undefined && hold.kind !== kind) return
     misses.delete(sessionId)
     held.delete(sessionId)
   }
+  const heldFor = (sessionId, kind) => {
+    const hold = held.get(sessionId)
+    return hold !== undefined && hold.kind === kind ? hold : undefined
+  }
 
   // Only a wake that actually left for a pending decision counts (#1172): a hold that never delivered
   // the reminder must not spend the Bot's two wakes. The count lives on the record so it survives a
@@ -196,8 +242,8 @@ export function install(rt) {
       if (!sessionId) continue
       if (inflight.get(sessionId)?.started) continue
       const fp = fingerprint(list)
-      const hold = held.get(sessionId)
-      if (hold && hold.fingerprint !== fp) clearMiss(sessionId)
+      const hold = heldFor(sessionId, 'work')
+      if (hold && hold.fingerprint !== fp) clearMiss(sessionId, 'work')
       else if (hold && clock.now() < hold.until) continue
       let agent
       try {
@@ -239,6 +285,65 @@ export function install(rt) {
         rt.warn('ds-bot: task drive could not wake %s: %s', botId, error)
       }
     }
+    await offerQueue()
+  }
+
+  // The pull wake (#1247): an unclaimed item wakes one idle member of its role in that member's
+  // own chat. One offer per room and role per pass — a queue of five must not start five turns —
+  // and one member, so the item is pulled by one person instead of broadcast to the whole role.
+  // The item's age is on the board; this pass adds no second presence block.
+  async function offerQueue() {
+    const byRoomRole = new Map()
+    for (const task of Object.values(rt.state.tasks ?? {})) {
+      if (!task?.queueRole || task.claimedBy || task.status !== 'open') continue
+      const roomId = task.origin?.roomId
+      const room = roomId ? rt.roomOf(roomId) : undefined
+      if (!room || room.origin === 'face-to-face') continue
+      if (Number.isFinite(task.retryAt) && task.retryAt > clock.now()) continue
+      const key = `${roomId}\u0000${String(task.queueRole).toLowerCase()}`
+      const list = byRoomRole.get(key) ?? []
+      list.push(task)
+      byRoomRole.set(key, list)
+    }
+    for (const list of byRoomRole.values()) {
+      const item = list.slice().sort(compareForQueue)[0]
+      const room = rt.roomOf(item.origin.roomId)
+      const carriers = (room?.members ?? [])
+        .filter(id => { const bot = rt.botOf(id); return Boolean(bot) && bot.remote === undefined && sameRole(bot.role, item.queueRole) })
+      for (const botId of carriers) {
+        const sessionId = rt.chatOf(botId)
+        if (!sessionId || rt.roomOf(sessionId)) continue
+        // A member that cannot claim is not the carrier: the offer moves down the role.
+        if (holdsCard(botId)) continue
+        if (busy(sessionId)) continue
+        const fp = queueFingerprint(item)
+        const hold = heldFor(sessionId, 'queue')
+        if (hold && hold.fingerprint !== fp) clearMiss(sessionId, 'queue')
+        else if (hold && clock.now() < hold.until) continue
+        let agent
+        try {
+          agent = await rt.resolveAgent(sessionId)
+        } catch (error) {
+          rt.warn('ds-bot: task drive could not reach %s: %s', sessionId, error)
+          continue
+        }
+        if (busy(sessionId)) continue
+        // One Session, one task wake per pass (#1247): the work pass above may already have woken
+        // this Session, and a wake may still wait in its inbox. A second `[Continue task]` in the
+        // same turn contradicts the first, and the item is offered next pass instead.
+        if (wakeWaiting(sessionId, agent)) continue
+        inflight.set(sessionId, { fingerprint: fp, started: false, queue: item.id })
+        try {
+          await rt.deliver({ toId: botId, sessionId, role: 'task', text: queueCueFor(item) })
+        } catch (error) {
+          inflight.delete(sessionId)
+          rt.warn('ds-bot: task drive could not offer %s to %s: %s', item.id, botId, error)
+          // The cue did not leave, so this member is not the carrier: try the next one.
+          continue
+        }
+        break
+      }
+    }
   }
 
   function consider() {
@@ -260,9 +365,10 @@ export function install(rt) {
     if (retry && job?.started) return undefined
     if (job?.started) {
       inflight.delete(sessionId)
-      const fp = fingerprint(drivingTasks(rt.selfOf(sessionId)))
-      if (fp === job.fingerprint) noteMiss(sessionId, fp)
-      else clearMiss(sessionId)
+      const kind = job.queue !== undefined ? 'queue' : 'work'
+      const fp = kind === 'queue' ? queueFingerprint(rt.state.tasks?.[job.queue]) : fingerprint(drivingTasks(rt.selfOf(sessionId)))
+      if (fp === job.fingerprint) noteMiss(sessionId, fp, kind)
+      else clearMiss(sessionId, kind)
     }
     return consider()
   }