diff --git a/changelog.d/1247-pull-queue.added.md b/changelog.d/1247-pull-queue.added.md new file mode 100644 index 00000000..2e65f88f --- /dev/null +++ b/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)). diff --git a/core/client.js b/core/client.js index 8bc26a1b..7a2adeb7 100644 --- a/core/client.js +++ b/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; diff --git a/core/src/host/room-context.js b/core/src/host/room-context.js index 5a76144a..3fe75806 100644 --- a/core/src/host/room-context.js +++ b/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') } diff --git a/core/src/host/task-drive.js b/core/src/host/task-drive.js index bc244e67..28c520ca 100644 --- a/core/src/host/task-drive.js +++ b/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() } diff --git a/core/src/host/tasks.js b/core/src/host/tasks.js index 2705b527..3ec3baa3 100644 --- a/core/src/host/tasks.js +++ b/core/src/host/tasks.js @@ -16,7 +16,7 @@ // are terminal, and accepted is reachable only from result-ready. A cancelled or // failed task never becomes accepted. Records keep pointers and a summary, not transcripts. import { randomBytes } from 'node:crypto' -import { columnOf, compareForWork } from '../shared/task-board.js' +import { columnOf, compareForQueue, compareForWork, sameRole } from '../shared/task-board.js' import { currentParticipants, deriveParticipants } from './room-membership.js' const ACTIVE = new Set(['open', 'claimed', 'working', 'blocked']) @@ -40,6 +40,20 @@ const invalid = message => Object.assign(new Error(message), { code: 'bot/invali const waitingForInput = task => task.status === 'blocked' && task.waitingFor === 'input' const canLinkSend = task => task.status === 'working' || waitingForInput(task) const PRIORITY = new Set(['urgent', 'high', 'normal', 'low']) +// A queue item names the role that may claim it (#1247). The label is the Bot's own role text +// (`roster.js`: free text, trimmed to 24 characters), never a group membership role. +const ROLE_MAX = 24 +const QUEUE_RETRY_DEFAULT_SEC = 60 +const QUEUE_RETRY_MAX_SEC = 7 * 24 * 60 * 60 +const readRole = (value) => { + if (value === undefined || value === null) return undefined + const role = oneLine(value, ROLE_MAX) + if (role === '') throw invalid('A queued item needs a role') + return role +} +// A refusal is named, never a silent null (#1247): the code and the message both carry the +// reason, so the tool's own error text says which refusal it was. +const refuse = (code, message) => Object.assign(new Error(`${code}: ${message}`), { code: `bot/${code}` }) const WALK = ['open', 'claimed', 'working', 'result-ready', 'accepted'] const TERMINAL_MATCH = new Set(['accepted', 'cancelled', 'failed']) @@ -357,6 +371,32 @@ export function install(rt) { return matches[0] } const isOverall = (task, roomId) => task?.kind === 'group' && task.origin?.roomId === roomId && openTask(task) + // ---------------------------------------------------------------- the pull queue (#1247) + // An item belongs to the room, not to a Bot: `queueRole` names who may claim it, `queueBy` is + // the lead that enqueued it, and `claimedBy` names the member that took it. `ownerBotId` stays + // the enqueuer until a claim, so every existing reader (owns, mayAct, prune, view) keeps + // working; task-drive skips an item that has a role and no claimer, so the enqueuer is not + // woken for its own item. + const unclaimed = task => Boolean(task?.queueRole) && !task?.claimedBy + const roleCarriers = (room, role) => liveParticipants(room) + .filter(person => person.kind === 'bot' && !person.remote) + .map(person => botRecord(person.id)) + .filter(bot => Boolean(bot) && sameRole(bot.role, role)) + // 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) { + delete task.waitingFor + delete task.inputRequest + delete task.inputApprovalIds + delete task.approvalId + delete task.claimedBy + task.status = 'open' + task.ownerBotId = task.queueBy ?? task.ownerBotId + task.progress = 0 + task.retryAt = Date.now() + retryInSec * 1000 + task.queueReason = reason + record(task, null, { by: 'returned to the queue', retryInSec }) + } const resolveParent = (roomId, ref, batch) => { const text = String(ref).trim() const direct = [find(text), batch.get(text)].find(task => isOverall(task, roomId)) @@ -408,12 +448,82 @@ export function install(rt) { return view(task) } + // A member claims from the room's queue (#1247). The claim is a board write: it sets the + // owner and moves the card to claimed, so the delivery view and the room board read the same + // card with no second row. + async function claimQueued(botId, body, roomId) { + const room = resolveRoom(botId, body.group, roomId) + const entry = room ? state.rooms?.[room] : undefined + if (!entry) throw invalid('A claim needs a group') + const myRole = oneLine(botRecord(botId)?.role ?? '', ROLE_MAX) + if (myRole === '') throw refuse('not-in-role', 'this Bot has no role to claim with') + // WIP = 1: one card in progress fills the member's slot. A blocked card waits on a person, + // so counting it would let one unanswered question freeze the member out of the queue. + const holding = Object.values(store()) + .filter(task => task.ownerBotId === botId && columnOf(task) === 'in-progress' && task.status !== 'blocked') + if (holding.length > 0) return { item: null, reason: 'wip', held: holding.sort(compareForWork)[0].id } + const now = Date.now() + const offered = task => !(Number.isFinite(task.retryAt) && task.retryAt > now) + const wantedId = body.id !== undefined && body.id !== null && String(body.id) !== '' ? String(body.id) : undefined + let pick + if (wantedId !== undefined) { + pick = find(wantedId) + if (pick === undefined) throw stale(`no task ${wantedId}`) + if (!unclaimed(pick) || pick.origin?.roomId !== entry.id) throw stale(`no queued item ${wantedId} in this group`) + if (!sameRole(pick.queueRole, myRole)) throw refuse('not-in-role', `that item is queued for ${pick.queueRole}, and this Bot is ${myRole}`) + if (!offered(pick)) return { item: null, reason: 'waiting' } + } else { + pick = Object.values(store()) + .filter(task => unclaimed(task) && task.status === 'open' && task.origin?.roomId === entry.id && sameRole(task.queueRole, myRole) && offered(task)) + .sort(compareForQueue)[0] + } + if (pick === undefined) return { item: null, reason: 'empty' } + // Pick and write are two steps: the tool's own load and the save both yield, so a second + // member can pick the same item inside the first claim's stack. The item is read again + // after the yield and only one still open is taken — compare-and-set, not a read (#1247). + await rt.load() + const live = find(pick.id) + if (live === undefined || live.status !== 'open' || live.claimedBy !== undefined) return { item: null, reason: 'taken' } + live.claimedBy = botId + live.ownerBotId = botId + delete live.retryAt + delete live.queueReason + transition(live, 'claimed', null, { by: 'claimed from the queue' }) + await save() + return { item: view(live) } + } + + // A member that cannot finish returns the item with one line of why; it is not offered + // again before `retryAt` (#1247). + async function releaseQueued(botId, spec) { + const task = find(spec?.task ?? spec?.id) + if (task === undefined) throw stale(`no task ${String(spec?.task ?? spec?.id ?? '')}`) + if (task.claimedBy !== botId) throw forbidden('Only the member that claimed this item may return it') + if (task.status !== 'claimed' && task.status !== 'working' && task.status !== 'blocked') { + throw stale(`a task that is ${task.status} cannot return to the queue`) + } + const reason = readOptionalLine(spec?.reason, 'A reason') + if (reason === undefined) throw invalid('Returning an item to the queue needs a one-line reason') + const raw = spec?.retryInSec + const seconds = raw === undefined || raw === null || raw === '' ? QUEUE_RETRY_DEFAULT_SEC : Number(raw) + if (!Number.isInteger(seconds) || seconds < 0 || seconds > QUEUE_RETRY_MAX_SEC) { + throw invalid(`retryInSec must be a whole number of seconds from 0 to ${QUEUE_RETRY_MAX_SEC}`) + } + requeue(task, reason, seconds) + await save() + return view(task) + } + // The Bot's own list, written from the conversation. The caller is the Bot. // The requester of a task is that task's Bot's owner, so a group admin who // does not own the Bot can plan a subtask and still cannot move the card. rt.trackBotTasks = async function trackBotTasks(botId, body, origin = {}) { const bot = botRecord(botId) if (!bot || bot.remote !== undefined) throw forbidden('No such Bot') + // Both come before the read path: a claim arrives with `tasks` omitted, so reading first + // would return the list instead of the item (#1247). + if (body?.release !== undefined) return releaseQueued(botId, body.release) + if (body?.claim === true) return claimQueued(botId, body, origin.roomId) const items = body?.tasks if (items === undefined) return listed(botId) if (!Array.isArray(items)) throw invalid('tasks must be a list') @@ -456,6 +566,19 @@ export function install(rt) { if (overall && parentRef) throw invalid('The overall task has no parent') if (overall && !plansRoom(botId, room)) planDenied(room, 'record the overall task') if (assigneeName && !plansRoom(botId, room)) planDenied(room, 'assign a subtask') + // A queued item names a role instead of an assignee (#1247). Only the room's lead + // enqueues, and an item whose role no live member carries is refused here: left open it + // would sit for ever and the age on the board would name a wait nobody can take. + const queue = item.queue === true + const role = readRole(item.role) + if (queue && !room) throw invalid('A queued item needs a group') + if (queue && room?.origin === 'face-to-face') throw invalid('A face-to-face room has no group task board') + if (queue && overall) throw invalid('A queued item is not the overall task') + if (queue && assigneeName) throw invalid('A queued item has no assignee; a member claims it from the queue') + if (queue && !plansRoom(botId, room)) planDenied(room, 'enqueue work') + if (queue && role === undefined) throw invalid('A queued item needs a role') + if (!queue && role !== undefined) throw invalid('role belongs to a queued item; pass queue: true with it') + if (queue && roleCarriers(room, role).length === 0) throw refuse('no-member-with-role', `no member of this group has the role ${role}`) let ownerBot = bot if (assigneeName) { ownerBot = memberBot(room, assigneeName) @@ -504,6 +627,13 @@ export function install(rt) { task.origin = { ...task.origin, roomId: parent.origin?.roomId ?? roomId } } if (overall) task.kind = 'group' + if (queue) { + if (task.claimedBy) throw invalid('That queued item is already claimed; enqueue a new one') + task.queueRole = role + task.queueBy = botId + delete task.retryAt + delete task.queueReason + } if (!answer && waitingForInput(task) && (column !== undefined || item.progress !== undefined)) { throw stale('This task is waiting on an answer. Ask, then pass answer with what was said.') } diff --git a/core/src/host/tools.js b/core/src/host/tools.js index 7dc83322..471e5180 100644 --- a/core/src/host/tools.js +++ b/core/src/host/tools.js @@ -831,8 +831,19 @@ ${task.writeBack.body}`) ctx.tools.register(defineTool({ name: 'track_tasks', - description: 'Read or update tasks on the board. Assigning work, claiming it, and recording progress are this tool, not a chat message. Omit tasks to read your list: the first lines name the unfinished task to continue (higher priority first) and any task waiting on an answer. Pass tasks to create or update your own. In a group, the task lead (the Bot picked for the room, or the admin when none is picked) records one overall task (overall: true) and a subtask per Bot (parent is the overall task\'s id or title, assignee is that Bot\'s name). Another Main Bot does not plan. A member claims the card assigned to it by setting its column to in-progress, updates that card\'s column and progress as it moves, or links its own subtask with parent. Do not record a second plan. Match an id, or the title of a task still open. Each task has a title, a column (not-started, in-progress, done), progress from 0 to 100, a priority (low, normal, high, urgent) and a note. Pass ask when you need an answer before the task can continue, and answer when the user or the group has answered. A group card is worked in your own chat. Post one short question in the group only when a person has to decide. In a group the task is tracked for that group.', + description: 'Read or update tasks on the board. Assigning work, claiming it, and recording progress are this tool, not a chat message. Omit tasks to read your list: the first lines name the unfinished task to continue (higher priority first) and any task waiting on an answer. Pass tasks to create or update your own. In a group, the task lead (the Bot picked for the room, or the admin when none is picked) records one overall task (overall: true) and a subtask per Bot (parent is the overall task\'s id or title, assignee is that Bot\'s name). Another Main Bot does not plan. A member claims the card assigned to it by setting its column to in-progress, updates that card\'s column and progress as it moves, or links its own subtask with parent. A lead can also enqueue work: pass queue with a role and no assignee, and an idle member of that role takes it with claim, which hands one item to exactly one member; release returns an item you claimed, with a one-line reason and a retryAt. Do not record a second plan. Match an id, or the title of a task still open. Each task has a title, a column (not-started, in-progress, done), progress from 0 to 100, a priority (low, normal, high, urgent) and a note. Pass ask when you need an answer before the task can continue, and answer when the user or the group has answered. A group card is worked in your own chat. Post one short question in the group only when a person has to decide. In a group the task is tracked for that group.', parameters: { + claim: { type: 'boolean', description: 'Take one queued item in this group. The highest priority first, then the one that has waited longest, for a role that matches yours; pass id to claim a named item. One member holds one card in progress, so a member already holding one gets nothing. A role mismatch is refused by name. Returns the item.' }, + release: { + type: 'object', + description: 'Return a queued item you claimed to the group queue with a one-line reason. It is not offered again before retryAt.', + properties: { + task: { type: 'string', description: 'The queued item to return.' }, + reason: { type: 'string', description: 'One line on why you could not finish it.' }, + retryInSec: { type: 'integer', description: 'Seconds before the item is offered again. Defaults to 60.' }, + }, + additionalProperties: false, + }, tasks: { type: 'array', description: 'Tasks to create or update. Omit to read your list.', @@ -850,6 +861,8 @@ ${task.writeBack.body}`) overall: { type: 'boolean', description: 'Record this as the group\'s one overall task. Only the room\'s task lead may set it (the group admin, when no lead is picked). A later call updates that same open task.' }, parent: { type: 'string', description: 'Id or title of the overall group task this subtask belongs to.' }, 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.' }, 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.' }, @@ -875,6 +888,21 @@ ${task.writeBack.body}`) const roomId = rt.groupOwners.get(exec.agent?.id)?.roomId try { if (args?.postResult !== undefined) await rt.recordTaskPostResult(self, args.postResult) + // A claim arrives with tasks omitted, so it is taken before the read path (#1247). + if (args?.claim === true) { + const result = await rt.trackBotTasks(self, args ?? {}, { sessionId: exec.agent?.id, roomId }) + if (result?.item) { + return `Claimed ${result.item.id} [${result.item.priority ?? 'normal'}] ${result.item.title}\nSet its column to in-progress with track_tasks as you start, and update its note as it moves. Do not post a status in the group.` + } + const role = rt.botOf(self)?.role ?? '' + const why = result?.reason === 'wip' ? ` You already hold an in-progress card (${result.held}).` : '' + return `No queued item for ${role === '' ? 'your role' : role} right now.${why}` + } + if (args?.release !== undefined) { + const returned = await rt.trackBotTasks(self, args ?? {}, { sessionId: exec.agent?.id, roomId }) + const seconds = Math.max(0, Math.round(((returned.retryAt ?? 0) - Date.now()) / 1000)) + return `Returned ${returned.id} to the queue: ${returned.queueReason}. Not offered again for ${seconds} s.` + } const tasks = await rt.trackBotTasks(self, args ?? {}, { sessionId: exec.agent?.id, roomId }) if (!Array.isArray(tasks) || tasks.length === 0) return 'No tasks.' return formatTasks(tasks, { inGroup: Boolean(roomId), withNext: args?.tasks === undefined }) diff --git a/core/src/shared/task-board.js b/core/src/shared/task-board.js index c8151ea0..f9c73ab5 100644 --- a/core/src/shared/task-board.js +++ b/core/src/shared/task-board.js @@ -44,6 +44,23 @@ export function compareForWork(a, b) { return (b?.updatedAt ?? 0) - (a?.updatedAt ?? 0) } +// A role label, matched the way a person reads it (#1247): free text, trimmed, case-insensitive. +// The label is not a membership role (`admin` / `member`), so nothing here reads that field. +export function sameRole(a, b) { + const left = String(a ?? '').trim().toLowerCase() + const right = String(b ?? '').trim().toLowerCase() + return left !== '' && left === right +} + +// Queue order (#1247): the highest priority first, then the item that has waited longest. +// compareForWork cannot be reused — its tiebreak is newest first (`task-board.js`), which +// starves the oldest item and would make the "oldest wait" on the board a lie. +export 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) +} + export 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 diff --git a/core/test/task-queue.test.js b/core/test/task-queue.test.js new file mode 100644 index 00000000..64bbfe01 --- /dev/null +++ b/core/test/task-queue.test.js @@ -0,0 +1,310 @@ +// A pull queue (#1247): the lead enqueues an item that names a role and no assignee, an idle +// member of that role claims it, and one item is handed to exactly one member. The queue is +// the same board — a claimed item is the card the room board and the delivery view already +// read — so this file covers only what the pull path adds: the single-winner claim, the named +// role refusals, WIP = 1, the wake list, the cheap wake, the blocked round-trip and the queue +// lines on the room board. +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 { ROOM_CONTEXT_BUDGET, formatRoomBoard } from '../src/host/room-context.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() + } + const board = () => formatRoomBoard(rt.state.tasks, { roomId: 'war', nameOf: task => rt.botOf(task.ownerBotId)?.name }) + return { ctx, rt, track, text, calls, emit, idOf, enqueue, finishWake, board, home } +} + +test('two claims for one queued item leave exactly one owner (#1247)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev', priority: 'high' }) + const [first, second] = await Promise.all([ + app.track('d1', { claim: true, group: 'War' }), + app.track('d2', { claim: true, group: 'War' }), + ]) + const claimed = [first, second].filter(text => text.startsWith('Claimed')) + assert.equal(claimed.length, 1, `one winner, got:\n${first}\n---\n${second}`) + assert.match(claimed[0], new RegExp(`Claimed ${id} \\[high\\] Fix the flaky test`)) + const loser = [first, second].find(text => !text.startsWith('Claimed')) + assert.match(loser, /^No queued item for /, 'the loser is told there is nothing, not handed the board') + const task = app.rt.state.tasks[id] + const winner = task.ownerBotId + assert.equal(task.status, 'claimed') + assert.equal(task.claimedBy, winner) + assert.equal(claimed[0].includes(app.rt.botOf(winner).name), false, 'the card names no owner in its wake text') +}) + +test('a member of another role is refused by name, not silently (#1247)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Review the diff', role: 'Reviewer' }) + const refused = await app.track('d1', { claim: true, group: 'War', id }) + assert.match(refused, /not-in-role/) + assert.doesNotMatch(refused, /^No queued item/, 'a role mismatch is named, not a silent null') + assert.equal(app.rt.state.tasks[id].claimedBy, undefined) + // Picking blind is a silent null: nothing in the queue is for a Dev. + assert.match(await app.track('d1', { claim: true, group: 'War' }), /^No queued item for Dev/) + // And the item is still there for the role it was queued for. + assert.match(await app.track('r1', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `)) +}) + +test('a member that already holds an in-progress card claims nothing (#1247)', async (t) => { + const app = await start(t) + const held = await app.track('d1', { tasks: [{ group: 'War', title: 'Held card', column: 'in-progress', progress: 30 }] }) + const heldId = app.idOf(held, 'Held card') + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev' }) + const refused = await app.track('d1', { claim: true, group: 'War' }) + assert.match(refused, /^No queued item for Dev/) + assert.match(refused, new RegExp(`in-progress card \\(${heldId}\\)`)) + assert.equal(app.rt.state.tasks[heldId].status, 'working') + assert.equal(app.rt.state.tasks[heldId].progress, 30) + assert.equal(app.rt.state.tasks[id].claimedBy, undefined) + // WIP is per member, not a room-wide lock: the other Dev takes it. + assert.match(await app.track('d2', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `)) +}) + +test('enqueuing wakes the idle member of the role and nobody else (#1247)', async (t) => { + const app = await start(t) + app.emit('d2', 'turn/start', { turn: 1 }) // Dev2 is mid-turn, so it is not idle + await app.enqueue({ title: 'Fix the flaky test', role: 'Dev' }) + assert.equal(app.calls('d1').length, 1, 'the idle Dev is woken') + assert.equal(app.calls('d2').length, 0, 'a Dev mid-turn is not woken') + assert.equal(app.calls('r1').length, 0, 'another role is not woken') + assert.equal(app.calls('a').length, 0, 'the enqueuer is not woken for its own item') + assert.match(app.text('d1'), /queued for Dev/) + await app.enqueue({ title: 'Review the diff', role: 'Reviewer' }) + assert.equal(app.calls('r1').length, 1, 'the second role gets its own wake') + assert.equal(app.calls('d1').length, 1, 'one wake per item: the Dev is not woken for a Reviewer item') +}) + +// The member that already holds a card in progress is refused by `claimQueued` (WIP = 1), so a queue +// cue for it is a turn spent on that refusal while the item waits behind its card. The offer goes to +// the idle member of the role instead (#1247). +test('an idle member behind one holding a card is the one woken (#1247)', async (t) => { + const app = await start(t) + const held = await app.track('d1', { tasks: [{ group: 'War', title: 'Held card', column: 'in-progress', progress: 30 }] }) + const heldId = app.idOf(held, 'Held card') + // That wake was the member's own turn: it ended, and the card is the only thing left on it. + await app.finishWake('d1') + assert.equal(app.calls('d1').length, 1) + + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev' }) + assert.equal(app.calls('d1').length, 1, 'the member holding a card is not offered the queue item') + assert.doesNotMatch(app.text('d1'), /Take /, 'a queue cue for a member that cannot claim it') + assert.equal(app.calls('d2').length, 1, 'the idle member of the role is woken') + assert.match(app.text('d2'), new RegExp(`Take ${id} `)) + assert.equal(app.rt.state.tasks[id].claimedBy, undefined, 'a wake is not a claim') + // WIP = 1 still refuses the claim itself, naming the card the member holds. + assert.match(await app.track('d1', { claim: true, group: 'War' }), new RegExp(`in-progress card \\(${heldId}\\)`)) +}) + +// One pass, one Session, one task wake (#1247): the work pass wakes the member its card names, and the +// queue pass must not stack a second cue on the same Session — one turn would read two contradictions, +// and the queue item would stay behind the first cue's card. +test('one pass gives a Session at most one task wake (#1247)', async (t) => { + const app = await start(t) + const body = await app.track('a', { tasks: [ + { title: 'Ship the note', overall: true, group: 'War' }, + { title: 'Write the draft', parent: 'Ship the note', assignee: 'Dev1', group: 'War', priority: 'high' }, + { title: 'Fix the flaky test', group: 'War', queue: true, role: 'Dev' }, + ] }) + const written = app.idOf(body, 'Write the draft') + const queued = app.idOf(body, 'Fix the flaky test') + assert.equal(app.calls('d1').length, 1, 'the assigned card wakes its member once') + assert.match(app.text('d1'), new RegExp(`Continue ${written} `)) + assert.doesNotMatch(app.text('d1'), /Take /, 'no second task wake on a Session that already has one') + assert.equal(app.calls('d2').length, 1, 'the idle member of the role takes the queue item instead') + assert.match(app.text('d2'), new RegExp(`Take ${queued} `)) + assert.equal(app.rt.state.tasks[queued].claimedBy, undefined, 'a wake is not a claim') +}) + +test('a queue wake names the title and id, not the item body (#1247)', async (t) => { + const app = await start(t) + const note = 'BODY_START the migration needs a backfill, a rollback plan and two reviewers before it moves BODY_END' + const { id } = await app.enqueue({ title: 'Backfill the table', role: 'Dev', note }) + const wake = app.text('d1') + assert.ok(wake.includes(id), wake) + assert.match(wake, /Backfill the table/) + assert.doesNotMatch(wake, /BODY_START/, 'the body is not carried into the wake') + assert.ok(wake.length < 400, `a wake is a short turn, got ${wake.length} chars:\n${wake}`) +}) + +test('a blocked item comes back to the queue with a retryAt (#1247)', async (t) => { + const app = await start(t) + t.mock.timers.enable({ apis: ['Date'], now: 1_700_000_000_000 }) + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev' }) + assert.match(await app.track('d1', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `)) + await app.track('d1', { tasks: [{ group: 'War', id, column: 'in-progress', progress: 10 }] }) + await app.track('d1', { tasks: [{ group: 'War', id, ask: 'Which key should it use?' }] }) + assert.equal(app.rt.state.tasks[id].status, 'blocked') + const returned = await app.track('d1', { release: { task: id, reason: 'needs the owner key', retryInSec: 120 } }) + assert.match(returned, new RegExp(`Returned ${id} to the queue`)) + const task = app.rt.state.tasks[id] + assert.equal(task.status, 'open') + assert.equal(task.claimedBy, undefined) + assert.equal(task.ownerBotId, 'a', 'an open item belongs to the room again, not to the member') + assert.equal(task.waitingFor, undefined) + assert.equal(task.retryAt, 1_700_000_120_000) + assert.equal(task.queueReason, 'needs the owner key') + assert.match(await app.track('d1', { claim: true, group: 'War' }), /^No queued item for Dev/, 'not offered again before retryAt') + t.mock.timers.tick(121_000) + assert.match(await app.track('d1', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `), 'offered again once retryAt passed') +}) + +test('an item whose role nobody in the room carries is refused at enqueue (#1247)', async (t) => { + const app = await start(t) + const body = await app.track('a', { tasks: [{ group: 'War', queue: true, role: 'Ops', title: 'Restart the box' }] }) + assert.match(body, /no-member-with-role/) + assert.equal(app.rt.state.tasks['task-' + '0'.repeat(16)], undefined) + assert.equal(Object.keys(app.rt.state.tasks ?? {}).length, 0, 'no item is left sitting for a role nobody has') +}) + +test('the room board carries one queue line per role, with depth and oldest wait (#1247)', () => { + const now = 1_700_000_000_000 + const minutes = 60_000 + const tasks = { + a1: { id: 'a1', status: 'open', queueRole: 'Dev', title: 'Old Dev item', ownerBotId: 'lead', origin: { roomId: 'room' }, createdAt: now - 180 * minutes, updatedAt: now - 180 * minutes }, + a2: { id: 'a2', status: 'open', queueRole: 'dev', title: 'New Dev item', ownerBotId: 'lead', origin: { roomId: 'room' }, createdAt: now - 10 * minutes, updatedAt: now - 10 * minutes }, + a3: { id: 'a3', status: 'open', queueRole: 'Reviewer', title: 'Review item', ownerBotId: 'lead', origin: { roomId: 'room' }, createdAt: now - 12 * minutes, updatedAt: now - 12 * minutes }, + a4: { id: 'a4', status: 'claimed', queueRole: 'Dev', claimedBy: 'd1', title: 'Taken Dev item', ownerBotId: 'd1', origin: { roomId: 'room' }, createdAt: now - 300 * minutes, updatedAt: now - 5 * minutes }, + } + const board = formatRoomBoard(tasks, { roomId: 'room', now, nameOf: task => (task.ownerBotId === 'd1' ? 'Dev1' : 'Chief') }) + assert.match(board, /- Queue Dev: 2 waiting, oldest 3h/, board) + assert.match(board, /- Queue Reviewer: 1 waiting, oldest 12m/, board) + assert.doesNotMatch(board, /Queue Ops/, 'a role with nothing waiting is omitted') + assert.match(board, /unassigned \[open 3h Dev\] Old Dev item/) + assert.match(board, /Dev1 \[claimed\] Taken Dev item/, 'a claimed item names its member, not the queue') +}) + +test('eight cards and three role lines still fit the room budget (#1247)', () => { + const now = 1_700_000_000_000 + const tasks = {} + const roles = ['Dev', 'Reviewer', 'Docs'] + for (let index = 0; index < 24; index += 1) { + const id = `q${index}` + tasks[id] = { + id, status: 'open', queueRole: roles[index % 3], ownerBotId: 'lead', origin: { roomId: 'room' }, + title: `A queued item with a long enough name to matter ${index} ${'detail '.repeat(6)}`, + note: 'note '.repeat(12), + createdAt: now - index * 60_000, updatedAt: now - index * 60_000, + } + } + const board = formatRoomBoard(tasks, { roomId: 'room', now, nameOf: () => 'Chief' }) + assert.ok(board.length <= ROOM_CONTEXT_BUDGET, `board is ${board.length} chars, budget ${ROOM_CONTEXT_BUDGET}`) + assert.match(board, /- Queue Dev: 8 waiting/) + assert.match(board, /- Queue Reviewer: 8 waiting/) + assert.match(board, /- Queue Docs: 8 waiting/) +}) + +test('a claimed item is the same card the room board already shows (#1247)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Ship the note', role: 'Dev' }) + assert.match(await app.track('d1', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `)) + const board = app.board() + const rows = board.split('\n').filter(line => line.includes('Ship the note')) + assert.equal(rows.length, 1, `one row for one card, got:\n${board}`) + assert.match(rows[0], /Dev1 \[claimed\] Ship the note/) + assert.doesNotMatch(board, /- Queue Dev/, 'the claimed item is not still counted in the queue') + assert.equal(Object.values(app.rt.state.tasks).filter(task => task.title === 'Ship the note').length, 1) +}) diff --git a/docs/subsystems/core-host.md b/docs/subsystems/core-host.md index 20716ac5..8ef8a749 100644 --- a/docs/subsystems/core-host.md +++ b/docs/subsystems/core-host.md @@ -105,7 +105,7 @@ A member's group Session reads the group's fixed rules from a system-prompt sect A group turn does not hand the model the raw ring buffer. `src/host/room-context.js` `fitRoomLines` takes the lines the count caps already chose (`ROOM_FIRST_LOOK` 12, `ROOM_DELTA_MAX` 24, and `ROOM_EARLIER_MAX` 8 of those from before the round) and keeps a newer-first slice of at most `ROOM_CONTEXT_BUDGET` characters (4000). The budget is characters because the measured essays are mostly CJK, about one token per character. A long bot line that reads as CI, merge-queue, or review narration (`statusShaped`: at least 400 characters and two of the status hints) is collapsed to a one-line stub that keeps the opening and up to four `@` names. The stored log line is not rewritten, and a person's line is not collapsed. When lines are left out, the turn says so and points at `read_group_chat` for an exact earlier line. `read_group_chat` still returns the log, not the stub. -The open task board for that room is a compact snapshot in the same turn (`formatRoomBoard`): the newest overall task, then its subtasks and any room card that is not under a different overall task, up to eight cards. Done cards stay off. Titles and notes are one line, capped, with `<` and `>` removed. A face-to-face room has no snapshot and keeps `@Name` as how its lead gives a part. The standing section does not carry the snapshot or the transcript, so it still changes only when the group settings change. In a group that has a board, `ROOM_REPLY_BOARD` and the task-lead note say assigning, claiming, and progress are `track_tasks` writes (overall task, subtask, assignee, column, progress), `@Name` only wakes, and a status the board already shows is `[PASS]`. A task-drive wake in the group says to set that card's column and progress, and appends the same snapshot after its cue. Covered by `core/test/room-context.test.js`, `core/test/rounds.test.js`, and `core/test/task-drive.test.js`. +The open task board for that room is a compact snapshot in the same turn (`formatRoomBoard`): the newest overall task, then its subtasks and any room card that is not under a different overall task, up to eight cards. Done cards stay off. Titles and notes are one line, capped, with `<` and `>` removed. An unclaimed queue item reads `unassigned [open 3h Dev]` — its role and its wait on the card itself — and the snapshot ends with one line per role that has something waiting (`- Queue Dev: 2 waiting, oldest 3h`), so starvation is visible without reading every card (#1247). A role with nothing waiting is omitted. A face-to-face room has no snapshot and keeps `@Name` as how its lead gives a part. The standing section does not carry the snapshot or the transcript, so it still changes only when the group settings change. In a group that has a board, `ROOM_REPLY_BOARD` and the task-lead note say assigning, claiming, and progress are `track_tasks` writes (overall task, subtask, assignee, column, progress), `@Name` only wakes, and a status the board already shows is `[PASS]`. A task-drive wake in the group says to set that card's column and progress, and appends the same snapshot after its cue. Covered by `core/test/room-context.test.js`, `core/test/rounds.test.js`, and `core/test/task-drive.test.js`. ### Who is working (#1213) @@ -192,7 +192,9 @@ Each Bot's work sits on the same `state.tasks` records. A column is read from th `track_tasks` is the conversation path. The calling Bot creates or updates only its own tasks (`source: 'conversation'`, `requesterId` the Bot's owner, `origin.sessionId` the turn, `origin.roomId` the group when the turn is in one). The same title in the same group updates the open task instead of adding another. The prompt tells the Bot to record work the user or a group actually handed it, to continue the unfinished task the tool names next, and to pass `ask` when it cannot continue without an answer. -A group job is one overall task plus a subtask per Bot (#1099). The overall task is `kind: 'group'`, owned by the Bot that records it, and tagged with `origin.roomId`. A group keeps one open overall task; recording another updates that one. A subtask sets `parentId` to that task. One Bot divides the board (#1182): `room.taskLead` when that Bot is still a member, otherwise the group admin. A face-to-face room has no board, so it has no task lead. Another Main Bot does not plan. The user, a Main Bot, or the group's admin sets the lead with `update_group` (`task_lead`, `none` clears it) or `update-room` (`taskLead`); group settings shows **Make task lead** / **Use the admin** on a group that lives here. Assigning a subtask to another member claims it (`open` → `claimed`) so the assignee starts without a confirmation, and wakes that assignee in its own chat (#1205); the assignment does not open a group round. The assignee then sets the column to in-progress with `track_tasks`; a chat line does not assign, claim, or record progress. The group board shows that above the columns. The requester stays that Bot's owner, so the lead still cannot move the card through `task-update`. A member updates the card assigned to it, records its own room task, or links its own subtask to the overall task. Only the lead records the overall task or assigns a subtask to someone else. Reading the list orders unfinished work by priority (urgent, high, normal, low) and prefers an in-progress task when the priority is the same. `ask` walks the task to working and then to the existing input wait (`waitingFor: 'input'`); the reply tells the Bot to call `ask_user` for the user, and to post one short question in the group only when a person there has to decide. `answer` resumes that wait. The question is the same record the inbox already shows for a blocked task. +A group job is one overall task plus a subtask per Bot (#1099). The overall task is `kind: 'group'`, owned by the Bot that records it, and tagged with `origin.roomId`. A group keeps one open overall task; recording another updates that one. A subtask sets `parentId` to that task. One Bot divides the board (#1182): `room.taskLead` when that Bot is still a member, otherwise the group admin. A face-to-face room has no board, so it has no task lead. Another Main Bot does not plan. The user, a Main Bot, or the group's admin sets the lead with `update_group` (`task_lead`, `none` clears it) or `update-room` (`taskLead`); group settings shows **Make task lead** / **Use the admin** on a group that lives here. Assigning a subtask to another member claims it (`open` → `claimed`) so the assignee starts without a confirmation, and wakes that assignee in its own chat (#1205); the assignment does not open a group round. The assignee then sets the column to in-progress with `track_tasks`; a chat line does not assign, claim, or record progress. The group board shows that above the columns. The requester stays that Bot's owner, so the lead still cannot move the card through `task-update`. A member updates the card assigned to it, records its own room task, or links its own subtask to the overall task. Only the lead records the overall task or assigns a subtask to someone else. + +A lead can put work in a pull queue instead of naming an assignee (#1247). `track_tasks { tasks: [{ queue: true, role: 'Dev', priority, title }] }` records one item for that role: `queueRole` names it, `queueBy` is the lead, and `ownerBotId` stays the enqueuer until a claim, so an unclaimed item wakes nobody — `task-drive` skips it. Only the room's lead enqueues, and an item whose role no live member of that room carries is refused (`no-member-with-role`) instead of sitting for ever for a role nobody has. `track_tasks { claim: true }` takes the highest-priority item of the caller's own role, oldest first on a tie (`compareForQueue`, not `compareForWork`, whose tiebreak is newest first), and `id` claims a named item; a member of another role is refused by name (`not-in-role`), never silently. One member holds one card in progress, so a member already holding one gets nothing — a blocked card waits on a person and does not count. The claim is compare-and-set: the item is read again after the pick yields, and only one still open is taken, so two claims leave exactly one owner. `track_tasks { release: { task, reason, retryInSec } }` returns a claimed item (`blocked → open` included) with `retryAt`, and it is not offered again before then. A claim sets `ownerBotId` and `claimedBy` to the member and keeps `queueRole`, so the card the board, the delivery view and room presence read is the same card: claiming is a board write, not a second row. Tests: `core/test/task-queue.test.js`. Reading the list orders unfinished work by priority (urgent, high, normal, low) and prefers an in-progress task when the priority is the same. `ask` walks the task to working and then to the existing input wait (`waitingFor: 'input'`); the reply tells the Bot to call `ask_user` for the user, and to post one short question in the group only when a person there has to decide. `answer` resumes that wait. The question is the same record the inbox already shows for a blocked task. A task may name where it came from (`target`, #1154): `task-create` and `track_tasks` take `https://github.com///issues|pull/` or `/#` and store `{ owner, repo, number, href }` with the href always on the `/issues/` form; anything else is refused (`bot/invalid`). The target is optional, and a task without one behaves exactly as before: no `writeBack`, nothing extra on the card, no extra wake. @@ -208,11 +210,11 @@ The module is read-only by construction: every request is a `GET`, and the paylo ### Task-driven turns (#1109) -Unfinished conversation tasks wake the owning Bot. `src/host/task-drive.js` runs after a board write and after a turn ends. For each owner, the next task is the one `compareForWork` would name: `open`, `claimed`, or `working`, not `waitingFor: 'input'`, and not an overall group card (`kind: 'group'`). A decision whose comment is still `pending` counts as work even when the task is already `accepted`, and the wake says where to post it (#1154). That card is the plan; the subtasks are the work. One drive pass wakes every such owner, so a group assignment starts each member's own subtask in that same pass. One Bot still receives only its own next task. A task with no group wakes the Bot's own chat. The wake is a follow-up whose header is `[Continue task]` (`delivery.js`, role `task`). It names the task id, priority, and title. It is not a group post, so the reply is not posted back and does not start a round. A group subtask wakes that same own chat (#1205). It does not open the Bot's group Session and it does not start a round. The cue says to set the card's column and progress with `track_tasks`, not to post a status in the group, and not to wait for a confirmation. A question a person has to decide is one short post in the group; the wake's own reply stays in the chat and is not posted into the room. Completing or updating the card is what wakes the next assignee. The wake text also carries that room's compact board snapshot (#1203), the same one a round turn shows, so the Bot reads progress from the board. A Bot @ in the group whose open card already shows that line is a pass (`boardCoversCue` in `room-context.js`, applied in the round before a member turn): the round records it as passed and does not start a turn. A question, a line the card does not already say, and a person's @ still get a turn. +Unfinished conversation tasks wake the owning Bot. `src/host/task-drive.js` runs after a board write and after a turn ends. For each owner, the next task is the one `compareForWork` would name: `open`, `claimed`, or `working`, not `waitingFor: 'input'`, and not an overall group card (`kind: 'group'`). A decision whose comment is still `pending` counts as work even when the task is already `accepted`, and the wake says where to post it (#1154). That card is the plan; the subtasks are the work. One drive pass wakes every such owner, so a group assignment starts each member's own subtask in that same pass. An unclaimed queue item is offered, not driven (#1247): one idle member of its role that does not already hold a card in progress is woken in that member's own chat, one item per room and role per pass, one Session gets at most one task wake in a pass, and the cue carries the item's id, priority and title — not its note, which is the body. A claimed item is then a normal card above. One Bot still receives only its own next task. A task with no group wakes the Bot's own chat. The wake is a follow-up whose header is `[Continue task]` (`delivery.js`, role `task`). It names the task id, priority, and title. It is not a group post, so the reply is not posted back and does not start a round. A group subtask wakes that same own chat (#1205). It does not open the Bot's group Session and it does not start a round. The cue says to set the card's column and progress with `track_tasks`, not to post a status in the group, and not to wait for a confirmation. A question a person has to decide is one short post in the group; the wake's own reply stays in the chat and is not posted into the room. Completing or updating the card is what wakes the next assignee. The wake text also carries that room's compact board snapshot (#1203), the same one a round turn shows, so the Bot reads progress from the board. A Bot @ in the group whose open card already shows that line is a pass (`boardCoversCue` in `room-context.js`, applied in the round before a member turn): the round records it as passed and does not start a turn. A question, a line the card does not already say, and a person's @ still get a turn. Tests reach the driver through `ctx.taskDrive` (`now`, `settled`, `consider`). That handle is a Cordis service registered with `provide`, the same path as `ctx.flybot`. A live context refuses a bare property write, which stopped startup before any Bot loaded (#1120). -The driver skips a Session that already has a turn, and a second wake while the first is still queued or in progress. A queued wake that has not started is replaced when a different task becomes that Bot's next one. A live group round does not defer that wake: it is not a group turn, so it does not sit beside the round. The parallel wake is the assignment pass. A wake that ends with the same board (status, progress, priority, note, title, who is waiting, and the state and target of a pending comment — not `updatedAt`, which a no-op `track_tasks` call still bumps) waits 30s, then 2min, 10min, and 30min before trying that Session again. A change on the board, or a person speaking in that Session, clears the wait. A task that is waiting on an answer does not wake: the turn that called `ask` asks once, and the driver does not loop. The idle rule still forbids inventing work; unfinished work the user or a group already handed the Bot counts as that task. Tests: `core/test/task-drive.test.js`. +The driver skips a Session that already has a turn, and a second wake while the first is still queued or in progress. A queued wake that has not started is replaced when a different task becomes that Bot's next one. A live group round does not defer that wake: it is not a group turn, so it does not sit beside the round. The parallel wake is the assignment pass. A wake that ends with the same board (status, progress, priority, note, title, who is waiting, and the state and target of a pending comment — not `updatedAt`, which a no-op `track_tasks` call still bumps) waits 30s, then 2min, 10min, and 30min before trying that Session again. A change on the board, or a person speaking in that Session, clears the wait. A queue offer keeps its own wait: the work pass and the queue pass compare different fingerprints, so one does not clear the other's backoff. A task that is waiting on an answer does not wake: the turn that called `ask` asks once, and the driver does not loop. The idle rule still forbids inventing work; unfinished work the user or a group already handed the Bot counts as that task. Tests: `core/test/task-drive.test.js`. ### Work view step (#546) diff --git a/patches/ds-bot/0002-flybot-api-host.patch b/patches/ds-bot/0002-flybot-api-host.patch index 7f41c26a..e6fbbbd9 100644 --- a/patches/ds-bot/0002-flybot-api-host.patch +++ b/patches/ds-bot/0002-flybot-api-host.patch @@ -1404,7 +1404,7 @@ diff --git a/src/host/tools.js b/src/host/tools.js .join('') return `${verb} ${args.name} ${args.bot ? `to ${result.bot}` : 'to yourself'}. Changes apply from ${result.bot}'s next message.${endpoint === 'attach' ? approvals : ''}` } catch (error) { return error.message } -@@ -595,5 +595,93 @@ export function install(rt) { +@@ -595,5 +595,121 @@ export function install(rt) { })) } @@ -1443,8 +1443,19 @@ diff --git a/src/host/tools.js b/src/host/tools.js + + ctx.tools.register(defineTool({ + name: 'track_tasks', -+ description: 'Read or update tasks on the board. Assigning work, claiming it, and recording progress are this tool, not a chat message. Omit tasks to read your list: the first lines name the unfinished task to continue (higher priority first) and any task waiting on an answer. Pass tasks to create or update your own. In a group, the task lead (the Bot picked for the room, or the admin when none is picked) records one overall task (overall: true) and a subtask per Bot (parent is the overall task\'s id or title, assignee is that Bot\'s name). Another Main Bot does not plan. A member claims the card assigned to it by setting its column to in-progress, updates that card\'s column and progress as it moves, or links its own subtask with parent. Do not record a second plan. Match an id, or the title of a task still open. Each task has a title, a column (not-started, in-progress, done), progress from 0 to 100, a priority (low, normal, high, urgent) and a note. Pass ask when you need an answer before the task can continue, and answer when the user or the group has answered. A group card is worked in your own chat. Post one short question in the group only when a person has to decide. In a group the task is tracked for that group.', ++ description: 'Read or update tasks on the board. Assigning work, claiming it, and recording progress are this tool, not a chat message. Omit tasks to read your list: the first lines name the unfinished task to continue (higher priority first) and any task waiting on an answer. Pass tasks to create or update your own. In a group, the task lead (the Bot picked for the room, or the admin when none is picked) records one overall task (overall: true) and a subtask per Bot (parent is the overall task\'s id or title, assignee is that Bot\'s name). Another Main Bot does not plan. A member claims the card assigned to it by setting its column to in-progress, updates that card\'s column and progress as it moves, or links its own subtask with parent. A lead can also enqueue work: pass queue with a role and no assignee, and an idle member of that role takes it with claim, which hands one item to exactly one member; release returns an item you claimed, with a one-line reason and a retryAt. Do not record a second plan. Match an id, or the title of a task still open. Each task has a title, a column (not-started, in-progress, done), progress from 0 to 100, a priority (low, normal, high, urgent) and a note. Pass ask when you need an answer before the task can continue, and answer when the user or the group has answered. A group card is worked in your own chat. Post one short question in the group only when a person has to decide. In a group the task is tracked for that group.', + parameters: { ++ claim: { type: 'boolean', description: 'Take one queued item in this group. The highest priority first, then the one that has waited longest, for a role that matches yours; pass id to claim a named item. One member holds one card in progress, so a member already holding one gets nothing. A role mismatch is refused by name. Returns the item.' }, ++ release: { ++ type: 'object', ++ description: 'Return a queued item you claimed to the group queue with a one-line reason. It is not offered again before retryAt.', ++ properties: { ++ task: { type: 'string', description: 'The queued item to return.' }, ++ reason: { type: 'string', description: 'One line on why you could not finish it.' }, ++ retryInSec: { type: 'integer', description: 'Seconds before the item is offered again. Defaults to 60.' }, ++ }, ++ additionalProperties: false, ++ }, + tasks: { + type: 'array', + description: 'Tasks to create or update. Omit to read your list.', @@ -1462,6 +1473,8 @@ diff --git a/src/host/tools.js b/src/host/tools.js + overall: { type: 'boolean', description: 'Record this as the group\'s one overall task. Only the room\'s task lead may set it (the group admin, when no lead is picked). A later call updates that same open task.' }, + parent: { type: 'string', description: 'Id or title of the overall group task this subtask belongs to.' }, + 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.' }, + 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.' }, @@ -1487,6 +1500,21 @@ diff --git a/src/host/tools.js b/src/host/tools.js + const roomId = rt.groupOwners.get(exec.agent?.id)?.roomId + try { + if (args?.postResult !== undefined) await rt.recordTaskPostResult(self, args.postResult) ++ // A claim arrives with tasks omitted, so it is taken before the read path (#1247). ++ if (args?.claim === true) { ++ const result = await rt.trackBotTasks(self, args ?? {}, { sessionId: exec.agent?.id, roomId }) ++ if (result?.item) { ++ return `Claimed ${result.item.id} [${result.item.priority ?? 'normal'}] ${result.item.title}\nSet its column to in-progress with track_tasks as you start, and update its note as it moves. Do not post a status in the group.` ++ } ++ const role = rt.botOf(self)?.role ?? '' ++ const why = result?.reason === 'wip' ? ` You already hold an in-progress card (${result.held}).` : '' ++ return `No queued item for ${role === '' ? 'your role' : role} right now.${why}` ++ } ++ if (args?.release !== undefined) { ++ const returned = await rt.trackBotTasks(self, args ?? {}, { sessionId: exec.agent?.id, roomId }) ++ const seconds = Math.max(0, Math.round(((returned.retryAt ?? 0) - Date.now()) / 1000)) ++ return `Returned ${returned.id} to the queue: ${returned.queueReason}. Not offered again for ${seconds} s.` ++ } + const tasks = await rt.trackBotTasks(self, args ?? {}, { sessionId: exec.agent?.id, roomId }) + if (!Array.isArray(tasks) || tasks.length === 0) return 'No tasks.' + return formatTasks(tasks, { inGroup: Boolean(roomId), withNext: args?.tasks === undefined }) diff --git a/patches/ds-bot/0008-client-bundle.patch b/patches/ds-bot/0008-client-bundle.patch index fc87177e..b59eced6 100644 --- a/patches/ds-bot/0008-client-bundle.patch +++ b/patches/ds-bot/0008-client-bundle.patch @@ -3190,7 +3190,7 @@ diff --git a/client.js b/client.js .bt-account{flex:none;width:36px;height:36px;border:0;padding:0;background:none;border-radius:50%;cursor:pointer;display:inline-flex} .bt-account:hover .bt-you,.bt-account[aria-expanded=true] .bt-you{background:color-mix(in srgb,var(--bt-ink) 9%,var(--bt-sidebar))} /* Popover keyframes move translate and scale, which compose with a transform the popover -@@ -2574,2206 +2574,4862 @@ body[data-ds-dark-theme] .bt-theme-chip i{background:var(--da)} +@@ -2574,2206 +2574,4872 @@ body[data-ds-dark-theme] .bt-theme-chip i{background:var(--da)} @media (max-width:640px){.bt-mem{width:100vw;height:100%;border-radius:0;border:0}.bt-mem-body{padding:18px 18px 20px}.bt-mem-foot{padding:8px 12px 12px}} @media (prefers-reduced-motion:reduce){.bt-mem,.bt-mem-scrim,.bt-mem-summary>*,.bt-mem-empty,.bt-mem-reply,.bt-mem-skel-block span,.bt-mem-entries,.bt-mem-send>*,.bt-mem-changes li svg{animation:none}} @@ -8374,6 +8374,16 @@ diff --git a/client.js b/client.js + 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; diff --git a/patches/ds-bot/files/src/host/room-context.js b/patches/ds-bot/files/src/host/room-context.js index 5a76144a..3fe75806 100644 --- a/patches/ds-bot/files/src/host/room-context.js +++ b/patches/ds-bot/files/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') } diff --git a/patches/ds-bot/files/src/host/task-drive.js b/patches/ds-bot/files/src/host/task-drive.js index bc244e67..28c520ca 100644 --- a/patches/ds-bot/files/src/host/task-drive.js +++ b/patches/ds-bot/files/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() } diff --git a/patches/ds-bot/files/src/host/tasks.js b/patches/ds-bot/files/src/host/tasks.js index 2705b527..3ec3baa3 100644 --- a/patches/ds-bot/files/src/host/tasks.js +++ b/patches/ds-bot/files/src/host/tasks.js @@ -16,7 +16,7 @@ // are terminal, and accepted is reachable only from result-ready. A cancelled or // failed task never becomes accepted. Records keep pointers and a summary, not transcripts. import { randomBytes } from 'node:crypto' -import { columnOf, compareForWork } from '../shared/task-board.js' +import { columnOf, compareForQueue, compareForWork, sameRole } from '../shared/task-board.js' import { currentParticipants, deriveParticipants } from './room-membership.js' const ACTIVE = new Set(['open', 'claimed', 'working', 'blocked']) @@ -40,6 +40,20 @@ const invalid = message => Object.assign(new Error(message), { code: 'bot/invali const waitingForInput = task => task.status === 'blocked' && task.waitingFor === 'input' const canLinkSend = task => task.status === 'working' || waitingForInput(task) const PRIORITY = new Set(['urgent', 'high', 'normal', 'low']) +// A queue item names the role that may claim it (#1247). The label is the Bot's own role text +// (`roster.js`: free text, trimmed to 24 characters), never a group membership role. +const ROLE_MAX = 24 +const QUEUE_RETRY_DEFAULT_SEC = 60 +const QUEUE_RETRY_MAX_SEC = 7 * 24 * 60 * 60 +const readRole = (value) => { + if (value === undefined || value === null) return undefined + const role = oneLine(value, ROLE_MAX) + if (role === '') throw invalid('A queued item needs a role') + return role +} +// A refusal is named, never a silent null (#1247): the code and the message both carry the +// reason, so the tool's own error text says which refusal it was. +const refuse = (code, message) => Object.assign(new Error(`${code}: ${message}`), { code: `bot/${code}` }) const WALK = ['open', 'claimed', 'working', 'result-ready', 'accepted'] const TERMINAL_MATCH = new Set(['accepted', 'cancelled', 'failed']) @@ -357,6 +371,32 @@ export function install(rt) { return matches[0] } const isOverall = (task, roomId) => task?.kind === 'group' && task.origin?.roomId === roomId && openTask(task) + // ---------------------------------------------------------------- the pull queue (#1247) + // An item belongs to the room, not to a Bot: `queueRole` names who may claim it, `queueBy` is + // the lead that enqueued it, and `claimedBy` names the member that took it. `ownerBotId` stays + // the enqueuer until a claim, so every existing reader (owns, mayAct, prune, view) keeps + // working; task-drive skips an item that has a role and no claimer, so the enqueuer is not + // woken for its own item. + const unclaimed = task => Boolean(task?.queueRole) && !task?.claimedBy + const roleCarriers = (room, role) => liveParticipants(room) + .filter(person => person.kind === 'bot' && !person.remote) + .map(person => botRecord(person.id)) + .filter(bot => Boolean(bot) && sameRole(bot.role, role)) + // 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) { + delete task.waitingFor + delete task.inputRequest + delete task.inputApprovalIds + delete task.approvalId + delete task.claimedBy + task.status = 'open' + task.ownerBotId = task.queueBy ?? task.ownerBotId + task.progress = 0 + task.retryAt = Date.now() + retryInSec * 1000 + task.queueReason = reason + record(task, null, { by: 'returned to the queue', retryInSec }) + } const resolveParent = (roomId, ref, batch) => { const text = String(ref).trim() const direct = [find(text), batch.get(text)].find(task => isOverall(task, roomId)) @@ -408,12 +448,82 @@ export function install(rt) { return view(task) } + // A member claims from the room's queue (#1247). The claim is a board write: it sets the + // owner and moves the card to claimed, so the delivery view and the room board read the same + // card with no second row. + async function claimQueued(botId, body, roomId) { + const room = resolveRoom(botId, body.group, roomId) + const entry = room ? state.rooms?.[room] : undefined + if (!entry) throw invalid('A claim needs a group') + const myRole = oneLine(botRecord(botId)?.role ?? '', ROLE_MAX) + if (myRole === '') throw refuse('not-in-role', 'this Bot has no role to claim with') + // WIP = 1: one card in progress fills the member's slot. A blocked card waits on a person, + // so counting it would let one unanswered question freeze the member out of the queue. + const holding = Object.values(store()) + .filter(task => task.ownerBotId === botId && columnOf(task) === 'in-progress' && task.status !== 'blocked') + if (holding.length > 0) return { item: null, reason: 'wip', held: holding.sort(compareForWork)[0].id } + const now = Date.now() + const offered = task => !(Number.isFinite(task.retryAt) && task.retryAt > now) + const wantedId = body.id !== undefined && body.id !== null && String(body.id) !== '' ? String(body.id) : undefined + let pick + if (wantedId !== undefined) { + pick = find(wantedId) + if (pick === undefined) throw stale(`no task ${wantedId}`) + if (!unclaimed(pick) || pick.origin?.roomId !== entry.id) throw stale(`no queued item ${wantedId} in this group`) + if (!sameRole(pick.queueRole, myRole)) throw refuse('not-in-role', `that item is queued for ${pick.queueRole}, and this Bot is ${myRole}`) + if (!offered(pick)) return { item: null, reason: 'waiting' } + } else { + pick = Object.values(store()) + .filter(task => unclaimed(task) && task.status === 'open' && task.origin?.roomId === entry.id && sameRole(task.queueRole, myRole) && offered(task)) + .sort(compareForQueue)[0] + } + if (pick === undefined) return { item: null, reason: 'empty' } + // Pick and write are two steps: the tool's own load and the save both yield, so a second + // member can pick the same item inside the first claim's stack. The item is read again + // after the yield and only one still open is taken — compare-and-set, not a read (#1247). + await rt.load() + const live = find(pick.id) + if (live === undefined || live.status !== 'open' || live.claimedBy !== undefined) return { item: null, reason: 'taken' } + live.claimedBy = botId + live.ownerBotId = botId + delete live.retryAt + delete live.queueReason + transition(live, 'claimed', null, { by: 'claimed from the queue' }) + await save() + return { item: view(live) } + } + + // A member that cannot finish returns the item with one line of why; it is not offered + // again before `retryAt` (#1247). + async function releaseQueued(botId, spec) { + const task = find(spec?.task ?? spec?.id) + if (task === undefined) throw stale(`no task ${String(spec?.task ?? spec?.id ?? '')}`) + if (task.claimedBy !== botId) throw forbidden('Only the member that claimed this item may return it') + if (task.status !== 'claimed' && task.status !== 'working' && task.status !== 'blocked') { + throw stale(`a task that is ${task.status} cannot return to the queue`) + } + const reason = readOptionalLine(spec?.reason, 'A reason') + if (reason === undefined) throw invalid('Returning an item to the queue needs a one-line reason') + const raw = spec?.retryInSec + const seconds = raw === undefined || raw === null || raw === '' ? QUEUE_RETRY_DEFAULT_SEC : Number(raw) + if (!Number.isInteger(seconds) || seconds < 0 || seconds > QUEUE_RETRY_MAX_SEC) { + throw invalid(`retryInSec must be a whole number of seconds from 0 to ${QUEUE_RETRY_MAX_SEC}`) + } + requeue(task, reason, seconds) + await save() + return view(task) + } + // The Bot's own list, written from the conversation. The caller is the Bot. // The requester of a task is that task's Bot's owner, so a group admin who // does not own the Bot can plan a subtask and still cannot move the card. rt.trackBotTasks = async function trackBotTasks(botId, body, origin = {}) { const bot = botRecord(botId) if (!bot || bot.remote !== undefined) throw forbidden('No such Bot') + // Both come before the read path: a claim arrives with `tasks` omitted, so reading first + // would return the list instead of the item (#1247). + if (body?.release !== undefined) return releaseQueued(botId, body.release) + if (body?.claim === true) return claimQueued(botId, body, origin.roomId) const items = body?.tasks if (items === undefined) return listed(botId) if (!Array.isArray(items)) throw invalid('tasks must be a list') @@ -456,6 +566,19 @@ export function install(rt) { if (overall && parentRef) throw invalid('The overall task has no parent') if (overall && !plansRoom(botId, room)) planDenied(room, 'record the overall task') if (assigneeName && !plansRoom(botId, room)) planDenied(room, 'assign a subtask') + // A queued item names a role instead of an assignee (#1247). Only the room's lead + // enqueues, and an item whose role no live member carries is refused here: left open it + // would sit for ever and the age on the board would name a wait nobody can take. + const queue = item.queue === true + const role = readRole(item.role) + if (queue && !room) throw invalid('A queued item needs a group') + if (queue && room?.origin === 'face-to-face') throw invalid('A face-to-face room has no group task board') + if (queue && overall) throw invalid('A queued item is not the overall task') + if (queue && assigneeName) throw invalid('A queued item has no assignee; a member claims it from the queue') + if (queue && !plansRoom(botId, room)) planDenied(room, 'enqueue work') + if (queue && role === undefined) throw invalid('A queued item needs a role') + if (!queue && role !== undefined) throw invalid('role belongs to a queued item; pass queue: true with it') + if (queue && roleCarriers(room, role).length === 0) throw refuse('no-member-with-role', `no member of this group has the role ${role}`) let ownerBot = bot if (assigneeName) { ownerBot = memberBot(room, assigneeName) @@ -504,6 +627,13 @@ export function install(rt) { task.origin = { ...task.origin, roomId: parent.origin?.roomId ?? roomId } } if (overall) task.kind = 'group' + if (queue) { + if (task.claimedBy) throw invalid('That queued item is already claimed; enqueue a new one') + task.queueRole = role + task.queueBy = botId + delete task.retryAt + delete task.queueReason + } if (!answer && waitingForInput(task) && (column !== undefined || item.progress !== undefined)) { throw stale('This task is waiting on an answer. Ask, then pass answer with what was said.') } diff --git a/patches/ds-bot/files/src/shared/task-board.js b/patches/ds-bot/files/src/shared/task-board.js index c8151ea0..f9c73ab5 100644 --- a/patches/ds-bot/files/src/shared/task-board.js +++ b/patches/ds-bot/files/src/shared/task-board.js @@ -44,6 +44,23 @@ export function compareForWork(a, b) { return (b?.updatedAt ?? 0) - (a?.updatedAt ?? 0) } +// A role label, matched the way a person reads it (#1247): free text, trimmed, case-insensitive. +// The label is not a membership role (`admin` / `member`), so nothing here reads that field. +export function sameRole(a, b) { + const left = String(a ?? '').trim().toLowerCase() + const right = String(b ?? '').trim().toLowerCase() + return left !== '' && left === right +} + +// Queue order (#1247): the highest priority first, then the item that has waited longest. +// compareForWork cannot be reused — its tiebreak is newest first (`task-board.js`), which +// starves the oldest item and would make the "oldest wait" on the board a lie. +export 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) +} + export 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 diff --git a/patches/ds-bot/files/test/task-queue.test.js b/patches/ds-bot/files/test/task-queue.test.js new file mode 100644 index 00000000..64bbfe01 --- /dev/null +++ b/patches/ds-bot/files/test/task-queue.test.js @@ -0,0 +1,310 @@ +// A pull queue (#1247): the lead enqueues an item that names a role and no assignee, an idle +// member of that role claims it, and one item is handed to exactly one member. The queue is +// the same board — a claimed item is the card the room board and the delivery view already +// read — so this file covers only what the pull path adds: the single-winner claim, the named +// role refusals, WIP = 1, the wake list, the cheap wake, the blocked round-trip and the queue +// lines on the room board. +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 { ROOM_CONTEXT_BUDGET, formatRoomBoard } from '../src/host/room-context.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() + } + const board = () => formatRoomBoard(rt.state.tasks, { roomId: 'war', nameOf: task => rt.botOf(task.ownerBotId)?.name }) + return { ctx, rt, track, text, calls, emit, idOf, enqueue, finishWake, board, home } +} + +test('two claims for one queued item leave exactly one owner (#1247)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev', priority: 'high' }) + const [first, second] = await Promise.all([ + app.track('d1', { claim: true, group: 'War' }), + app.track('d2', { claim: true, group: 'War' }), + ]) + const claimed = [first, second].filter(text => text.startsWith('Claimed')) + assert.equal(claimed.length, 1, `one winner, got:\n${first}\n---\n${second}`) + assert.match(claimed[0], new RegExp(`Claimed ${id} \\[high\\] Fix the flaky test`)) + const loser = [first, second].find(text => !text.startsWith('Claimed')) + assert.match(loser, /^No queued item for /, 'the loser is told there is nothing, not handed the board') + const task = app.rt.state.tasks[id] + const winner = task.ownerBotId + assert.equal(task.status, 'claimed') + assert.equal(task.claimedBy, winner) + assert.equal(claimed[0].includes(app.rt.botOf(winner).name), false, 'the card names no owner in its wake text') +}) + +test('a member of another role is refused by name, not silently (#1247)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Review the diff', role: 'Reviewer' }) + const refused = await app.track('d1', { claim: true, group: 'War', id }) + assert.match(refused, /not-in-role/) + assert.doesNotMatch(refused, /^No queued item/, 'a role mismatch is named, not a silent null') + assert.equal(app.rt.state.tasks[id].claimedBy, undefined) + // Picking blind is a silent null: nothing in the queue is for a Dev. + assert.match(await app.track('d1', { claim: true, group: 'War' }), /^No queued item for Dev/) + // And the item is still there for the role it was queued for. + assert.match(await app.track('r1', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `)) +}) + +test('a member that already holds an in-progress card claims nothing (#1247)', async (t) => { + const app = await start(t) + const held = await app.track('d1', { tasks: [{ group: 'War', title: 'Held card', column: 'in-progress', progress: 30 }] }) + const heldId = app.idOf(held, 'Held card') + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev' }) + const refused = await app.track('d1', { claim: true, group: 'War' }) + assert.match(refused, /^No queued item for Dev/) + assert.match(refused, new RegExp(`in-progress card \\(${heldId}\\)`)) + assert.equal(app.rt.state.tasks[heldId].status, 'working') + assert.equal(app.rt.state.tasks[heldId].progress, 30) + assert.equal(app.rt.state.tasks[id].claimedBy, undefined) + // WIP is per member, not a room-wide lock: the other Dev takes it. + assert.match(await app.track('d2', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `)) +}) + +test('enqueuing wakes the idle member of the role and nobody else (#1247)', async (t) => { + const app = await start(t) + app.emit('d2', 'turn/start', { turn: 1 }) // Dev2 is mid-turn, so it is not idle + await app.enqueue({ title: 'Fix the flaky test', role: 'Dev' }) + assert.equal(app.calls('d1').length, 1, 'the idle Dev is woken') + assert.equal(app.calls('d2').length, 0, 'a Dev mid-turn is not woken') + assert.equal(app.calls('r1').length, 0, 'another role is not woken') + assert.equal(app.calls('a').length, 0, 'the enqueuer is not woken for its own item') + assert.match(app.text('d1'), /queued for Dev/) + await app.enqueue({ title: 'Review the diff', role: 'Reviewer' }) + assert.equal(app.calls('r1').length, 1, 'the second role gets its own wake') + assert.equal(app.calls('d1').length, 1, 'one wake per item: the Dev is not woken for a Reviewer item') +}) + +// The member that already holds a card in progress is refused by `claimQueued` (WIP = 1), so a queue +// cue for it is a turn spent on that refusal while the item waits behind its card. The offer goes to +// the idle member of the role instead (#1247). +test('an idle member behind one holding a card is the one woken (#1247)', async (t) => { + const app = await start(t) + const held = await app.track('d1', { tasks: [{ group: 'War', title: 'Held card', column: 'in-progress', progress: 30 }] }) + const heldId = app.idOf(held, 'Held card') + // That wake was the member's own turn: it ended, and the card is the only thing left on it. + await app.finishWake('d1') + assert.equal(app.calls('d1').length, 1) + + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev' }) + assert.equal(app.calls('d1').length, 1, 'the member holding a card is not offered the queue item') + assert.doesNotMatch(app.text('d1'), /Take /, 'a queue cue for a member that cannot claim it') + assert.equal(app.calls('d2').length, 1, 'the idle member of the role is woken') + assert.match(app.text('d2'), new RegExp(`Take ${id} `)) + assert.equal(app.rt.state.tasks[id].claimedBy, undefined, 'a wake is not a claim') + // WIP = 1 still refuses the claim itself, naming the card the member holds. + assert.match(await app.track('d1', { claim: true, group: 'War' }), new RegExp(`in-progress card \\(${heldId}\\)`)) +}) + +// One pass, one Session, one task wake (#1247): the work pass wakes the member its card names, and the +// queue pass must not stack a second cue on the same Session — one turn would read two contradictions, +// and the queue item would stay behind the first cue's card. +test('one pass gives a Session at most one task wake (#1247)', async (t) => { + const app = await start(t) + const body = await app.track('a', { tasks: [ + { title: 'Ship the note', overall: true, group: 'War' }, + { title: 'Write the draft', parent: 'Ship the note', assignee: 'Dev1', group: 'War', priority: 'high' }, + { title: 'Fix the flaky test', group: 'War', queue: true, role: 'Dev' }, + ] }) + const written = app.idOf(body, 'Write the draft') + const queued = app.idOf(body, 'Fix the flaky test') + assert.equal(app.calls('d1').length, 1, 'the assigned card wakes its member once') + assert.match(app.text('d1'), new RegExp(`Continue ${written} `)) + assert.doesNotMatch(app.text('d1'), /Take /, 'no second task wake on a Session that already has one') + assert.equal(app.calls('d2').length, 1, 'the idle member of the role takes the queue item instead') + assert.match(app.text('d2'), new RegExp(`Take ${queued} `)) + assert.equal(app.rt.state.tasks[queued].claimedBy, undefined, 'a wake is not a claim') +}) + +test('a queue wake names the title and id, not the item body (#1247)', async (t) => { + const app = await start(t) + const note = 'BODY_START the migration needs a backfill, a rollback plan and two reviewers before it moves BODY_END' + const { id } = await app.enqueue({ title: 'Backfill the table', role: 'Dev', note }) + const wake = app.text('d1') + assert.ok(wake.includes(id), wake) + assert.match(wake, /Backfill the table/) + assert.doesNotMatch(wake, /BODY_START/, 'the body is not carried into the wake') + assert.ok(wake.length < 400, `a wake is a short turn, got ${wake.length} chars:\n${wake}`) +}) + +test('a blocked item comes back to the queue with a retryAt (#1247)', async (t) => { + const app = await start(t) + t.mock.timers.enable({ apis: ['Date'], now: 1_700_000_000_000 }) + const { id } = await app.enqueue({ title: 'Fix the flaky test', role: 'Dev' }) + assert.match(await app.track('d1', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `)) + await app.track('d1', { tasks: [{ group: 'War', id, column: 'in-progress', progress: 10 }] }) + await app.track('d1', { tasks: [{ group: 'War', id, ask: 'Which key should it use?' }] }) + assert.equal(app.rt.state.tasks[id].status, 'blocked') + const returned = await app.track('d1', { release: { task: id, reason: 'needs the owner key', retryInSec: 120 } }) + assert.match(returned, new RegExp(`Returned ${id} to the queue`)) + const task = app.rt.state.tasks[id] + assert.equal(task.status, 'open') + assert.equal(task.claimedBy, undefined) + assert.equal(task.ownerBotId, 'a', 'an open item belongs to the room again, not to the member') + assert.equal(task.waitingFor, undefined) + assert.equal(task.retryAt, 1_700_000_120_000) + assert.equal(task.queueReason, 'needs the owner key') + assert.match(await app.track('d1', { claim: true, group: 'War' }), /^No queued item for Dev/, 'not offered again before retryAt') + t.mock.timers.tick(121_000) + assert.match(await app.track('d1', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `), 'offered again once retryAt passed') +}) + +test('an item whose role nobody in the room carries is refused at enqueue (#1247)', async (t) => { + const app = await start(t) + const body = await app.track('a', { tasks: [{ group: 'War', queue: true, role: 'Ops', title: 'Restart the box' }] }) + assert.match(body, /no-member-with-role/) + assert.equal(app.rt.state.tasks['task-' + '0'.repeat(16)], undefined) + assert.equal(Object.keys(app.rt.state.tasks ?? {}).length, 0, 'no item is left sitting for a role nobody has') +}) + +test('the room board carries one queue line per role, with depth and oldest wait (#1247)', () => { + const now = 1_700_000_000_000 + const minutes = 60_000 + const tasks = { + a1: { id: 'a1', status: 'open', queueRole: 'Dev', title: 'Old Dev item', ownerBotId: 'lead', origin: { roomId: 'room' }, createdAt: now - 180 * minutes, updatedAt: now - 180 * minutes }, + a2: { id: 'a2', status: 'open', queueRole: 'dev', title: 'New Dev item', ownerBotId: 'lead', origin: { roomId: 'room' }, createdAt: now - 10 * minutes, updatedAt: now - 10 * minutes }, + a3: { id: 'a3', status: 'open', queueRole: 'Reviewer', title: 'Review item', ownerBotId: 'lead', origin: { roomId: 'room' }, createdAt: now - 12 * minutes, updatedAt: now - 12 * minutes }, + a4: { id: 'a4', status: 'claimed', queueRole: 'Dev', claimedBy: 'd1', title: 'Taken Dev item', ownerBotId: 'd1', origin: { roomId: 'room' }, createdAt: now - 300 * minutes, updatedAt: now - 5 * minutes }, + } + const board = formatRoomBoard(tasks, { roomId: 'room', now, nameOf: task => (task.ownerBotId === 'd1' ? 'Dev1' : 'Chief') }) + assert.match(board, /- Queue Dev: 2 waiting, oldest 3h/, board) + assert.match(board, /- Queue Reviewer: 1 waiting, oldest 12m/, board) + assert.doesNotMatch(board, /Queue Ops/, 'a role with nothing waiting is omitted') + assert.match(board, /unassigned \[open 3h Dev\] Old Dev item/) + assert.match(board, /Dev1 \[claimed\] Taken Dev item/, 'a claimed item names its member, not the queue') +}) + +test('eight cards and three role lines still fit the room budget (#1247)', () => { + const now = 1_700_000_000_000 + const tasks = {} + const roles = ['Dev', 'Reviewer', 'Docs'] + for (let index = 0; index < 24; index += 1) { + const id = `q${index}` + tasks[id] = { + id, status: 'open', queueRole: roles[index % 3], ownerBotId: 'lead', origin: { roomId: 'room' }, + title: `A queued item with a long enough name to matter ${index} ${'detail '.repeat(6)}`, + note: 'note '.repeat(12), + createdAt: now - index * 60_000, updatedAt: now - index * 60_000, + } + } + const board = formatRoomBoard(tasks, { roomId: 'room', now, nameOf: () => 'Chief' }) + assert.ok(board.length <= ROOM_CONTEXT_BUDGET, `board is ${board.length} chars, budget ${ROOM_CONTEXT_BUDGET}`) + assert.match(board, /- Queue Dev: 8 waiting/) + assert.match(board, /- Queue Reviewer: 8 waiting/) + assert.match(board, /- Queue Docs: 8 waiting/) +}) + +test('a claimed item is the same card the room board already shows (#1247)', async (t) => { + const app = await start(t) + const { id } = await app.enqueue({ title: 'Ship the note', role: 'Dev' }) + assert.match(await app.track('d1', { claim: true, group: 'War' }), new RegExp(`^Claimed ${id} `)) + const board = app.board() + const rows = board.split('\n').filter(line => line.includes('Ship the note')) + assert.equal(rows.length, 1, `one row for one card, got:\n${board}`) + assert.match(rows[0], /Dev1 \[claimed\] Ship the note/) + assert.doesNotMatch(board, /- Queue Dev/, 'the claimed item is not still counted in the queue') + assert.equal(Object.values(app.rt.state.tasks).filter(task => task.title === 'Ship the note').length, 1) +})