diff --git a/changelog.d/2597.fixed.md b/changelog.d/2597.fixed.md new file mode 100644 index 000000000..94f20e158 --- /dev/null +++ b/changelog.d/2597.fixed.md @@ -0,0 +1 @@ +A `track_tasks` batch that is refused no longer leaves the cards it had already created on the board: the whole list is validated on a staged copy of the board and only committed once every item passed, so a refused batch leaves `state.tasks` exactly as it was and posts no claim line for a card that never landed ([#2597](https://github.com/ThinkFlowLab/FlyBot/issues/2597)). diff --git a/core/src/host/tasks.js b/core/src/host/tasks.js index a320355be..e1dea3545 100644 --- a/core/src/host/tasks.js +++ b/core/src/host/tasks.js @@ -177,9 +177,13 @@ export function install(rt) { const { state } = rt state.tasks ??= {} + // #2597: a batch is one write. `trackBotTasks` stages its whole list on a copy of the board + // and commits the copy only after the last item passed, so a refusal anywhere in the list + // leaves `state.tasks` byte-identical; while it runs, every reader and writer sees the copy. + let staging = null const store = () => { state.tasks ??= {} - return state.tasks + return staging ?? state.tasks } // A workflow run names a board task, never a second one (#1696). Core never reads // `workflows.json`, so the workflow plugin registers its run refs here and the host only @@ -1140,137 +1144,152 @@ export function install(rt) { } const saved = [] const kinds = [] + const claims = [] const batch = new Map() - for (const item of items) { - if (item === null || typeof item !== 'object' || Array.isArray(item)) throw invalid('Each task must be an object') - const column = readColumn(item.column) - const title = item.title !== undefined ? readTitle(item.title) : undefined - const ask = readOptionalLine(item.ask, 'A question') - const answer = readOptionalLine(item.answer, 'An answer') - const overall = item.overall === true - const assigneeName = item.assignee === undefined || item.assignee === null ? '' : String(item.assignee).trim() - const parentRef = item.parent === undefined || item.parent === null ? '' : String(item.parent).trim() - if (ask && column === 'done') throw invalid('A task that needs an answer cannot be done') - const roomId = resolveRoom(botId, item.group, origin.roomId) - const room = roomId ? state.rooms?.[roomId] : undefined - if ((overall || assigneeName || parentRef) && !room) throw invalid('A group task needs a group') - if (room?.origin === 'face-to-face' && (overall || assigneeName || parentRef)) throw invalid('A face-to-face room has no group task board') - if (overall && assigneeName) throw invalid('The overall task is the group\'s, not a subtask') - 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) - if (ownerBot.id !== botId && !parentRef) throw invalid('A subtask needs the overall task (parent)') - } - const parent = parentRef ? resolveParent(roomId, parentRef, batch) : undefined - let task - let isNew = false - let frozen = false - const givenId = item.id !== undefined && item.id !== null && String(item.id) !== '' - if (givenId) { - task = find(item.id) - if (!task || task.ownerBotId !== botId) throw forbidden('That task is not yours') - if (overall && task.kind !== 'group') throw invalid('That task is not the overall task') - } else if (overall) { - task = openOverall(roomId) - if (task && task.ownerBotId !== botId) throw forbidden('That overall task belongs to another Bot') - if (!task) { - if (!title) throw invalid('A new task needs a title') - isNew = true - task = makeTask(bot, title, roomId) - task.kind = 'group' - } - } else if (ownerBot.id !== botId) { - if (!title) throw invalid('A new task needs a title') - task = matchChild(ownerBot.id, title, roomId, parent.id) - if (task) frozen = true - else { - isNew = true - task = makeTask(ownerBot, title, roomId) - task.parentId = parent.id - } - } else { - if (!title) throw invalid('A new task needs a title') - task = matchTitle(botId, title, roomId) - if (!task) { - // A member still records its own room card, parented or not. Dividing - // the board (overall, or assignee) is refused above (#1182). - isNew = true - task = makeTask(bot, title, roomId) - } - } - if (!frozen) { - if (parent && task.kind !== 'group') { - task.parentId = parent.id - 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') - // A graph may state the role's ceiling; otherwise the one constant does (#1729). The - // count leaves this card out, so re-offering a queued item keeps its id and is never - // refused for the place it already holds. - const cap = readCap(item.cap) ?? (Number.isInteger(task.queueCap) ? task.queueCap : ROLE_QUEUE_CAP) - const queued = Object.values(store()).filter(other => other.id !== task.id - && other.origin?.roomId === roomId && unclaimed(other) && other.status === 'open' - && other.queueRole && sameRole(other.queueRole, role)) - if (queued.length >= cap) throw refuse('queue-full', `the ${role} queue already holds ${queued.length} items, its cap is ${cap}`) - task.queueRole = role - task.queueBy = botId - task.queueCap = cap - delete task.retryAt - delete task.queueReason + const board = store() + // The list is one write: it runs against a staged copy of the board, and the copy is + // committed only after the last item passed. A refusal anywhere in the list therefore + // leaves `state.tasks` byte-identical and posts no claim line for a card that never landed + // (#2597). Every card the accepted batch created lands on the board once, below. + const staged = structuredClone(board) + staging = staged + try { + for (const item of items) { + if (item === null || typeof item !== 'object' || Array.isArray(item)) throw invalid('Each task must be an object') + const column = readColumn(item.column) + const title = item.title !== undefined ? readTitle(item.title) : undefined + const ask = readOptionalLine(item.ask, 'A question') + const answer = readOptionalLine(item.answer, 'An answer') + const overall = item.overall === true + const assigneeName = item.assignee === undefined || item.assignee === null ? '' : String(item.assignee).trim() + const parentRef = item.parent === undefined || item.parent === null ? '' : String(item.parent).trim() + if (ask && column === 'done') throw invalid('A task that needs an answer cannot be done') + const roomId = resolveRoom(botId, item.group, origin.roomId) + const room = roomId ? state.rooms?.[roomId] : undefined + if ((overall || assigneeName || parentRef) && !room) throw invalid('A group task needs a group') + if (room?.origin === 'face-to-face' && (overall || assigneeName || parentRef)) throw invalid('A face-to-face room has no group task board') + if (overall && assigneeName) throw invalid('The overall task is the group\'s, not a subtask') + 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) + if (ownerBot.id !== botId && !parentRef) throw invalid('A subtask needs the overall task (parent)') } - 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.') + const parent = parentRef ? resolveParent(roomId, parentRef, batch) : undefined + let task + let isNew = false + let frozen = false + const givenId = item.id !== undefined && item.id !== null && String(item.id) !== '' + if (givenId) { + task = find(item.id) + if (!task || task.ownerBotId !== botId) throw forbidden('That task is not yours') + if (overall && task.kind !== 'group') throw invalid('That task is not the overall task') + } else if (overall) { + task = openOverall(roomId) + if (task && task.ownerBotId !== botId) throw forbidden('That overall task belongs to another Bot') + if (!task) { + if (!title) throw invalid('A new task needs a title') + isNew = true + task = makeTask(bot, title, roomId) + task.kind = 'group' + } + } else if (ownerBot.id !== botId) { + if (!title) throw invalid('A new task needs a title') + task = matchChild(ownerBot.id, title, roomId, parent.id) + if (task) frozen = true + else { + isNew = true + task = makeTask(ownerBot, title, roomId) + task.parentId = parent.id + } + } else { + if (!title) throw invalid('A new task needs a title') + task = matchTitle(botId, title, roomId) + if (!task) { + // A member still records its own room card, parented or not. Dividing + // the board (overall, or assignee) is refused above (#1182). + isNew = true + task = makeTask(bot, title, roomId) + } } - if (title && task.title !== title) task.title = title - // A refused value must not leave a half-created card behind: the writer validates before - // it assigns, and a new row it refuses is removed again (#1720). - try { - writeContract(task, item) - } catch (error) { - if (isNew) delete store()[task.id] - throw error + if (!frozen) { + if (parent && task.kind !== 'group') { + task.parentId = parent.id + 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') + // A graph may state the role's ceiling; otherwise the one constant does (#1729). The + // count leaves this card out, so re-offering a queued item keeps its id and is never + // refused for the place it already holds. + const cap = readCap(item.cap) ?? (Number.isInteger(task.queueCap) ? task.queueCap : ROLE_QUEUE_CAP) + const queued = Object.values(store()).filter(other => other.id !== task.id + && other.origin?.roomId === roomId && unclaimed(other) && other.status === 'open' + && other.queueRole && sameRole(other.queueRole, role)) + if (queued.length >= cap) throw refuse('queue-full', `the ${role} queue already holds ${queued.length} items, its cap is ${cap}`) + task.queueRole = role + task.queueBy = botId + task.queueCap = cap + delete task.retryAt + delete task.queueReason + } + 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.') + } + if (title && task.title !== title) task.title = title + // A refused value must not leave a half-created card behind: the writer validates before + // it assigns, and a new row it refuses is removed again (#1720). + try { + writeContract(task, item) + } catch (error) { + if (isNew) delete store()[task.id] + throw error + } + const { progress } = applyFields(task, item) + if (!task.source) task.source = 'conversation' + const before = task.updatedAt + // #1253: only the move out of the queue/not-started is a claim, so the progress ticks that + // follow it on the same card never reach the room. + const wasOpen = task.status === 'open' + if (answer) resumeFromAnswer(task, answer) + const moved = applyColumn(task, column, null, 'conversation') + if (column === 'done' && progress === undefined) task.progress = 100 + if (ask) { + if (task.ownerBotId !== botId) throw forbidden('Only that Bot can mark its task waiting') + blockForInput(task, ask) + } else if (!answer && !isNew && !moved && task.updatedAt === before) record(task, null, { by: 'conversation' }) + // Assigning claims the card. The assignee starts the work; nobody confirms first (#1182). + if (isNew && parent && task.ownerBotId !== botId && task.status === 'open') transition(task, 'claimed', null, { by: 'assigned' }) + if (wasOpen && (task.status === 'claimed' || task.status === 'working')) claims.push(task) } - const { progress } = applyFields(task, item) - if (!task.source) task.source = 'conversation' - const before = task.updatedAt - // #1253: only the move out of the queue/not-started is a claim, so the progress ticks that - // follow it on the same card never reach the room. - const wasOpen = task.status === 'open' - if (answer) resumeFromAnswer(task, answer) - const moved = applyColumn(task, column, null, 'conversation') - if (column === 'done' && progress === undefined) task.progress = 100 - if (ask) { - if (task.ownerBotId !== botId) throw forbidden('Only that Bot can mark its task waiting') - blockForInput(task, ask) - } else if (!answer && !isNew && !moved && task.updatedAt === before) record(task, null, { by: 'conversation' }) - // Assigning claims the card. The assignee starts the work; nobody confirms first (#1182). - if (isNew && parent && task.ownerBotId !== botId && task.status === 'open') transition(task, 'claimed', null, { by: 'assigned' }) - if (wasOpen && (task.status === 'claimed' || task.status === 'working')) announceClaim(task) + if (task.title) batch.set(task.title, task) + batch.set(task.id, task) + saved.push(task) + kinds.push(isNew ? 'create' : 'update') } - if (task.title) batch.set(task.title, task) - batch.set(task.id, task) - saved.push(task) - kinds.push(isNew ? 'create' : 'update') + prune() + } finally { + staging = null } - prune() + for (const id of Object.keys(board)) if (!(id in staged)) delete board[id] + for (const [id, task] of Object.entries(staged)) board[id] = task + for (const task of claims) announceClaim(task) await save() // The conversation path is a board write like task-create and task-update, so a plugin // has to see it: a card a Bot records itself with `workflow:` starts nothing diff --git a/core/test/track-tasks-batch-refusal.test.js b/core/test/track-tasks-batch-refusal.test.js new file mode 100644 index 000000000..d47798c00 --- /dev/null +++ b/core/test/track-tasks-batch-refusal.test.js @@ -0,0 +1,124 @@ +// #2597: `track_tasks` created the first card of a batch while the later items were still +// being validated, and the batch `save()` only ran after the loop. An item two refusal — the +// `ask` + `column: 'done'` pair — unwound with the first card already in `state.tasks`, so the +// caller was told the batch failed while a card nobody asked for stayed on the board. +// The whole store is compared, not one field, so any other early write shows up too. +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createRuntime } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +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' }, + }, + rooms: { + war: { + id: 'war', name: 'War', members: ['a', 'd1'], admin: 'a', mode: 'everyone', named: true, sessions: {}, log: [], seq: 0, seen: {}, + participants: [ + { kind: 'human', id: 'owner-1', role: 'admin', 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 }, + ], + }, + }, + grants: {}, + agents: {}, exchanges: [], exchangeTurns: {}, questions: {}, prefs: {}, +}) + +async function start(t) { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-batch-refusal-')) + writeFileSync(join(home, 'state.json'), JSON.stringify(team())) + const ctx = fakeContext() + const agents = new Map() + const agentOf = (sessionId) => { + if (!agents.has(sessionId)) { + agents.set(sessionId, { + session: { id: sessionId }, id: sessionId, calls: [], status: 'idle', + followup() {}, steer() {}, inject() {}, cancel() {}, async whenIdle() {}, inbox: { nextStep: [], nextTurn: [], remove() {} }, + }) + } + return agents.get(sessionId) + } + 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-1' }), 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 () => { + try { await ctx.dispose() } catch {} + rmSync(home, { recursive: true, force: true, maxRetries: 5 }) + }) + const saves = [] + const save = rt.save + rt.save = async () => { saves.push(1); return save() } + return { rt, saves } +} + +const snapshot = rt => JSON.stringify(rt.state.tasks) + +test('a refused batch leaves state.tasks byte-identical (#2597)', async (t) => { + const { rt, saves } = await start(t) + const before = snapshot(rt) + assert.equal(before, '{}') + + await assert.rejects( + () => rt.trackBotTasks('d1', { + tasks: [ + { title: 'First card' }, + { title: 'Second card', column: 'done', ask: 'Which one?' }, + ], + }), + error => error.code === 'bot/invalid' && /A task that needs an answer cannot be done/.test(error.message), + ) + assert.equal(snapshot(rt), before, 'the refused batch left the store untouched') + assert.equal(saves.length, 0, 'a refused batch is never saved') +}) + +test('a refused batch does not keep an earlier item\'s update to an existing card (#2597)', async (t) => { + const { rt } = await start(t) + const [card] = await rt.trackBotTasks('d1', { tasks: [{ title: 'Existing card' }] }) + const before = snapshot(rt) + + // Item one is accepted and would rename the card; item two is refused, so neither may land. + await assert.rejects( + () => rt.trackBotTasks('d1', { + tasks: [ + { id: card.id, title: 'Renamed', note: 'A note' }, + { title: 'Second card', column: 'done', ask: 'Which one?' }, + ], + }), + error => error.code === 'bot/invalid' && /A task that needs an answer cannot be done/.test(error.message), + ) + assert.equal(snapshot(rt), before, 'the refused batch left the existing card untouched') + assert.equal(rt.state.tasks[card.id].title, 'Existing card') +}) + +test('an accepted batch creates every card once (#2597)', async (t) => { + const { rt } = await start(t) + const cards = await rt.trackBotTasks('d1', { + tasks: [ + { title: 'First card' }, + { title: 'Second card', column: 'in-progress' }, + ], + }) + assert.equal(cards.length, 2) + assert.equal(cards[0].title, 'First card') + assert.equal(cards[1].title, 'Second card') + assert.equal(cards[1].status, 'working') + assert.deepEqual(Object.keys(rt.state.tasks).sort(), [cards[0].id, cards[1].id].sort(), 'both cards landed once') +}) diff --git a/patches/ds-bot/files/src/host/tasks.js b/patches/ds-bot/files/src/host/tasks.js index a320355be..e1dea3545 100644 --- a/patches/ds-bot/files/src/host/tasks.js +++ b/patches/ds-bot/files/src/host/tasks.js @@ -177,9 +177,13 @@ export function install(rt) { const { state } = rt state.tasks ??= {} + // #2597: a batch is one write. `trackBotTasks` stages its whole list on a copy of the board + // and commits the copy only after the last item passed, so a refusal anywhere in the list + // leaves `state.tasks` byte-identical; while it runs, every reader and writer sees the copy. + let staging = null const store = () => { state.tasks ??= {} - return state.tasks + return staging ?? state.tasks } // A workflow run names a board task, never a second one (#1696). Core never reads // `workflows.json`, so the workflow plugin registers its run refs here and the host only @@ -1140,137 +1144,152 @@ export function install(rt) { } const saved = [] const kinds = [] + const claims = [] const batch = new Map() - for (const item of items) { - if (item === null || typeof item !== 'object' || Array.isArray(item)) throw invalid('Each task must be an object') - const column = readColumn(item.column) - const title = item.title !== undefined ? readTitle(item.title) : undefined - const ask = readOptionalLine(item.ask, 'A question') - const answer = readOptionalLine(item.answer, 'An answer') - const overall = item.overall === true - const assigneeName = item.assignee === undefined || item.assignee === null ? '' : String(item.assignee).trim() - const parentRef = item.parent === undefined || item.parent === null ? '' : String(item.parent).trim() - if (ask && column === 'done') throw invalid('A task that needs an answer cannot be done') - const roomId = resolveRoom(botId, item.group, origin.roomId) - const room = roomId ? state.rooms?.[roomId] : undefined - if ((overall || assigneeName || parentRef) && !room) throw invalid('A group task needs a group') - if (room?.origin === 'face-to-face' && (overall || assigneeName || parentRef)) throw invalid('A face-to-face room has no group task board') - if (overall && assigneeName) throw invalid('The overall task is the group\'s, not a subtask') - 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) - if (ownerBot.id !== botId && !parentRef) throw invalid('A subtask needs the overall task (parent)') - } - const parent = parentRef ? resolveParent(roomId, parentRef, batch) : undefined - let task - let isNew = false - let frozen = false - const givenId = item.id !== undefined && item.id !== null && String(item.id) !== '' - if (givenId) { - task = find(item.id) - if (!task || task.ownerBotId !== botId) throw forbidden('That task is not yours') - if (overall && task.kind !== 'group') throw invalid('That task is not the overall task') - } else if (overall) { - task = openOverall(roomId) - if (task && task.ownerBotId !== botId) throw forbidden('That overall task belongs to another Bot') - if (!task) { - if (!title) throw invalid('A new task needs a title') - isNew = true - task = makeTask(bot, title, roomId) - task.kind = 'group' - } - } else if (ownerBot.id !== botId) { - if (!title) throw invalid('A new task needs a title') - task = matchChild(ownerBot.id, title, roomId, parent.id) - if (task) frozen = true - else { - isNew = true - task = makeTask(ownerBot, title, roomId) - task.parentId = parent.id - } - } else { - if (!title) throw invalid('A new task needs a title') - task = matchTitle(botId, title, roomId) - if (!task) { - // A member still records its own room card, parented or not. Dividing - // the board (overall, or assignee) is refused above (#1182). - isNew = true - task = makeTask(bot, title, roomId) - } - } - if (!frozen) { - if (parent && task.kind !== 'group') { - task.parentId = parent.id - 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') - // A graph may state the role's ceiling; otherwise the one constant does (#1729). The - // count leaves this card out, so re-offering a queued item keeps its id and is never - // refused for the place it already holds. - const cap = readCap(item.cap) ?? (Number.isInteger(task.queueCap) ? task.queueCap : ROLE_QUEUE_CAP) - const queued = Object.values(store()).filter(other => other.id !== task.id - && other.origin?.roomId === roomId && unclaimed(other) && other.status === 'open' - && other.queueRole && sameRole(other.queueRole, role)) - if (queued.length >= cap) throw refuse('queue-full', `the ${role} queue already holds ${queued.length} items, its cap is ${cap}`) - task.queueRole = role - task.queueBy = botId - task.queueCap = cap - delete task.retryAt - delete task.queueReason + const board = store() + // The list is one write: it runs against a staged copy of the board, and the copy is + // committed only after the last item passed. A refusal anywhere in the list therefore + // leaves `state.tasks` byte-identical and posts no claim line for a card that never landed + // (#2597). Every card the accepted batch created lands on the board once, below. + const staged = structuredClone(board) + staging = staged + try { + for (const item of items) { + if (item === null || typeof item !== 'object' || Array.isArray(item)) throw invalid('Each task must be an object') + const column = readColumn(item.column) + const title = item.title !== undefined ? readTitle(item.title) : undefined + const ask = readOptionalLine(item.ask, 'A question') + const answer = readOptionalLine(item.answer, 'An answer') + const overall = item.overall === true + const assigneeName = item.assignee === undefined || item.assignee === null ? '' : String(item.assignee).trim() + const parentRef = item.parent === undefined || item.parent === null ? '' : String(item.parent).trim() + if (ask && column === 'done') throw invalid('A task that needs an answer cannot be done') + const roomId = resolveRoom(botId, item.group, origin.roomId) + const room = roomId ? state.rooms?.[roomId] : undefined + if ((overall || assigneeName || parentRef) && !room) throw invalid('A group task needs a group') + if (room?.origin === 'face-to-face' && (overall || assigneeName || parentRef)) throw invalid('A face-to-face room has no group task board') + if (overall && assigneeName) throw invalid('The overall task is the group\'s, not a subtask') + 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) + if (ownerBot.id !== botId && !parentRef) throw invalid('A subtask needs the overall task (parent)') } - 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.') + const parent = parentRef ? resolveParent(roomId, parentRef, batch) : undefined + let task + let isNew = false + let frozen = false + const givenId = item.id !== undefined && item.id !== null && String(item.id) !== '' + if (givenId) { + task = find(item.id) + if (!task || task.ownerBotId !== botId) throw forbidden('That task is not yours') + if (overall && task.kind !== 'group') throw invalid('That task is not the overall task') + } else if (overall) { + task = openOverall(roomId) + if (task && task.ownerBotId !== botId) throw forbidden('That overall task belongs to another Bot') + if (!task) { + if (!title) throw invalid('A new task needs a title') + isNew = true + task = makeTask(bot, title, roomId) + task.kind = 'group' + } + } else if (ownerBot.id !== botId) { + if (!title) throw invalid('A new task needs a title') + task = matchChild(ownerBot.id, title, roomId, parent.id) + if (task) frozen = true + else { + isNew = true + task = makeTask(ownerBot, title, roomId) + task.parentId = parent.id + } + } else { + if (!title) throw invalid('A new task needs a title') + task = matchTitle(botId, title, roomId) + if (!task) { + // A member still records its own room card, parented or not. Dividing + // the board (overall, or assignee) is refused above (#1182). + isNew = true + task = makeTask(bot, title, roomId) + } } - if (title && task.title !== title) task.title = title - // A refused value must not leave a half-created card behind: the writer validates before - // it assigns, and a new row it refuses is removed again (#1720). - try { - writeContract(task, item) - } catch (error) { - if (isNew) delete store()[task.id] - throw error + if (!frozen) { + if (parent && task.kind !== 'group') { + task.parentId = parent.id + 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') + // A graph may state the role's ceiling; otherwise the one constant does (#1729). The + // count leaves this card out, so re-offering a queued item keeps its id and is never + // refused for the place it already holds. + const cap = readCap(item.cap) ?? (Number.isInteger(task.queueCap) ? task.queueCap : ROLE_QUEUE_CAP) + const queued = Object.values(store()).filter(other => other.id !== task.id + && other.origin?.roomId === roomId && unclaimed(other) && other.status === 'open' + && other.queueRole && sameRole(other.queueRole, role)) + if (queued.length >= cap) throw refuse('queue-full', `the ${role} queue already holds ${queued.length} items, its cap is ${cap}`) + task.queueRole = role + task.queueBy = botId + task.queueCap = cap + delete task.retryAt + delete task.queueReason + } + 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.') + } + if (title && task.title !== title) task.title = title + // A refused value must not leave a half-created card behind: the writer validates before + // it assigns, and a new row it refuses is removed again (#1720). + try { + writeContract(task, item) + } catch (error) { + if (isNew) delete store()[task.id] + throw error + } + const { progress } = applyFields(task, item) + if (!task.source) task.source = 'conversation' + const before = task.updatedAt + // #1253: only the move out of the queue/not-started is a claim, so the progress ticks that + // follow it on the same card never reach the room. + const wasOpen = task.status === 'open' + if (answer) resumeFromAnswer(task, answer) + const moved = applyColumn(task, column, null, 'conversation') + if (column === 'done' && progress === undefined) task.progress = 100 + if (ask) { + if (task.ownerBotId !== botId) throw forbidden('Only that Bot can mark its task waiting') + blockForInput(task, ask) + } else if (!answer && !isNew && !moved && task.updatedAt === before) record(task, null, { by: 'conversation' }) + // Assigning claims the card. The assignee starts the work; nobody confirms first (#1182). + if (isNew && parent && task.ownerBotId !== botId && task.status === 'open') transition(task, 'claimed', null, { by: 'assigned' }) + if (wasOpen && (task.status === 'claimed' || task.status === 'working')) claims.push(task) } - const { progress } = applyFields(task, item) - if (!task.source) task.source = 'conversation' - const before = task.updatedAt - // #1253: only the move out of the queue/not-started is a claim, so the progress ticks that - // follow it on the same card never reach the room. - const wasOpen = task.status === 'open' - if (answer) resumeFromAnswer(task, answer) - const moved = applyColumn(task, column, null, 'conversation') - if (column === 'done' && progress === undefined) task.progress = 100 - if (ask) { - if (task.ownerBotId !== botId) throw forbidden('Only that Bot can mark its task waiting') - blockForInput(task, ask) - } else if (!answer && !isNew && !moved && task.updatedAt === before) record(task, null, { by: 'conversation' }) - // Assigning claims the card. The assignee starts the work; nobody confirms first (#1182). - if (isNew && parent && task.ownerBotId !== botId && task.status === 'open') transition(task, 'claimed', null, { by: 'assigned' }) - if (wasOpen && (task.status === 'claimed' || task.status === 'working')) announceClaim(task) + if (task.title) batch.set(task.title, task) + batch.set(task.id, task) + saved.push(task) + kinds.push(isNew ? 'create' : 'update') } - if (task.title) batch.set(task.title, task) - batch.set(task.id, task) - saved.push(task) - kinds.push(isNew ? 'create' : 'update') + prune() + } finally { + staging = null } - prune() + for (const id of Object.keys(board)) if (!(id in staged)) delete board[id] + for (const [id, task] of Object.entries(staged)) board[id] = task + for (const task of claims) announceClaim(task) await save() // The conversation path is a board write like task-create and task-update, so a plugin // has to see it: a card a Bot records itself with `workflow:` starts nothing diff --git a/patches/ds-bot/files/test/track-tasks-batch-refusal.test.js b/patches/ds-bot/files/test/track-tasks-batch-refusal.test.js new file mode 100644 index 000000000..d47798c00 --- /dev/null +++ b/patches/ds-bot/files/test/track-tasks-batch-refusal.test.js @@ -0,0 +1,124 @@ +// #2597: `track_tasks` created the first card of a batch while the later items were still +// being validated, and the batch `save()` only ran after the loop. An item two refusal — the +// `ask` + `column: 'done'` pair — unwound with the first card already in `state.tasks`, so the +// caller was told the batch failed while a card nobody asked for stayed on the board. +// The whole store is compared, not one field, so any other early write shows up too. +import assert from 'node:assert/strict' +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' +import { createRuntime } from '../index.js' +import { fakeContext } from './fake-ctx.js' + +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' }, + }, + rooms: { + war: { + id: 'war', name: 'War', members: ['a', 'd1'], admin: 'a', mode: 'everyone', named: true, sessions: {}, log: [], seq: 0, seen: {}, + participants: [ + { kind: 'human', id: 'owner-1', role: 'admin', 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 }, + ], + }, + }, + grants: {}, + agents: {}, exchanges: [], exchangeTurns: {}, questions: {}, prefs: {}, +}) + +async function start(t) { + const home = mkdtempSync(join(tmpdir(), 'ds-bot-batch-refusal-')) + writeFileSync(join(home, 'state.json'), JSON.stringify(team())) + const ctx = fakeContext() + const agents = new Map() + const agentOf = (sessionId) => { + if (!agents.has(sessionId)) { + agents.set(sessionId, { + session: { id: sessionId }, id: sessionId, calls: [], status: 'idle', + followup() {}, steer() {}, inject() {}, cancel() {}, async whenIdle() {}, inbox: { nextStep: [], nextTurn: [], remove() {} }, + }) + } + return agents.get(sessionId) + } + 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-1' }), 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 () => { + try { await ctx.dispose() } catch {} + rmSync(home, { recursive: true, force: true, maxRetries: 5 }) + }) + const saves = [] + const save = rt.save + rt.save = async () => { saves.push(1); return save() } + return { rt, saves } +} + +const snapshot = rt => JSON.stringify(rt.state.tasks) + +test('a refused batch leaves state.tasks byte-identical (#2597)', async (t) => { + const { rt, saves } = await start(t) + const before = snapshot(rt) + assert.equal(before, '{}') + + await assert.rejects( + () => rt.trackBotTasks('d1', { + tasks: [ + { title: 'First card' }, + { title: 'Second card', column: 'done', ask: 'Which one?' }, + ], + }), + error => error.code === 'bot/invalid' && /A task that needs an answer cannot be done/.test(error.message), + ) + assert.equal(snapshot(rt), before, 'the refused batch left the store untouched') + assert.equal(saves.length, 0, 'a refused batch is never saved') +}) + +test('a refused batch does not keep an earlier item\'s update to an existing card (#2597)', async (t) => { + const { rt } = await start(t) + const [card] = await rt.trackBotTasks('d1', { tasks: [{ title: 'Existing card' }] }) + const before = snapshot(rt) + + // Item one is accepted and would rename the card; item two is refused, so neither may land. + await assert.rejects( + () => rt.trackBotTasks('d1', { + tasks: [ + { id: card.id, title: 'Renamed', note: 'A note' }, + { title: 'Second card', column: 'done', ask: 'Which one?' }, + ], + }), + error => error.code === 'bot/invalid' && /A task that needs an answer cannot be done/.test(error.message), + ) + assert.equal(snapshot(rt), before, 'the refused batch left the existing card untouched') + assert.equal(rt.state.tasks[card.id].title, 'Existing card') +}) + +test('an accepted batch creates every card once (#2597)', async (t) => { + const { rt } = await start(t) + const cards = await rt.trackBotTasks('d1', { + tasks: [ + { title: 'First card' }, + { title: 'Second card', column: 'in-progress' }, + ], + }) + assert.equal(cards.length, 2) + assert.equal(cards[0].title, 'First card') + assert.equal(cards[1].title, 'Second card') + assert.equal(cards[1].status, 'working') + assert.deepEqual(Object.keys(rt.state.tasks).sort(), [cards[0].id, cards[1].id].sort(), 'both cards landed once') +})