diff --git a/docs/plans/2026-08-18-peer-session-collaboration.md b/docs/plans/2026-08-18-peer-session-collaboration.md new file mode 100644 index 00000000000..6c5972ff28a --- /dev/null +++ b/docs/plans/2026-08-18-peer-session-collaboration.md @@ -0,0 +1,213 @@ +# Peer session collaboration for Qwen Code + +> Status: Board layer implemented by #9402; not a standalone collaboration milestone +> Tracking: [#8724](https://github.com/QwenLM/qwen-code/issues/8724) + +## Decision + +The product path and the portable board are separate layers. The original #8724 milestone is +Qwen-to-Qwen collaboration: already-running Qwen Code sessions discover each other and use +`send_message` with a receiver-side consent gate. Agent Team already owns the simpler case +where one session spawns and coordinates an in-process Qwen teammate. + +Independently started or non-Qwen agents can additionally collaborate through a durable board +on disk. Every participant uses the same pull-based CLI. Qwen-specific delivery, terminal +hosting, and UI do not define the board contract, but delivery or orchestration is required +before the project can claim a user-visible delegation milestone. + +The first release has three rules: + +1. **No membership.** There is no join, leave, participant record, roster, heartbeat, or name + claim. Reading or writing a named board is participation. +2. **No implicit identity or scope.** Commands require `--board ` and mutations require + `--as `. These values label records; they do not authenticate the caller. +3. **Pull is the contract.** Work becomes visible on the board. Nothing is delivered into a + running agent process. + +This is the smallest storage model shared by Qwen Code, Codex, shell scripts, scheduled jobs, +and other tools that can run a command. It is not by itself a scheduler, launcher, or inbox. + +## Product sequence + +The user-visible collaboration route is: + +1. **Spawned Qwen teammates.** Existing Agent and Agent Team tools cover delegation to a + Qwen worker managed by the current session. +2. **Already-running Qwen sessions.** The session registry discovers peers; the inbound gate + lets a peer refuse or hold work; sender addressing completes the original #8724 flow. +3. **Independent process launch.** A launcher is needed only when a separate long-lived Qwen + Code process is required instead of an in-process teammate. +4. **Foreign runtimes.** A concrete runner, such as a Codex runner, turns one board task into + a real process invocation and returns its result. + +Stages 1 and 2 of the board implementation below prove the portable storage contract. They +must not be presented as completion of this product sequence, and the board should not block +the Qwen-to-Qwen sender path. + +## Why a board + +An already-running process can receive unsolicited input only if it voluntarily exposes an +inbound channel. Foreign CLIs do not expose a common one. They can all run a command, so +fetching shared state is the portable operation. + +Qwen Code already has the pieces needed for a filesystem-backed implementation: + +- atomic JSON writes; +- `proper-lockfile` for cross-process locking; +- task ownership and state-transition patterns; +- process-liveness discovery for features that need it later. + +The board reuses the locking and state-transition patterns. It does not reuse Agent Team's +storage or scheduler because those are scoped to a spawned, in-process team. + +## MVP contract + +### Scope and identity + +`--board` is required on every command. It is a logical name, not a path. The implementation +validates it before resolving the directory under `~/.qwen/boards/`. + +`--as` is required when creating or changing an item. Read-only commands may omit it unless +they request an identity-filtered view. The value is written into fields such as `createdBy`, +`owner`, or `from`. + +There is deliberately no default derived from the current directory, environment variable, +global process state, or live-session registry. Those shortcuts can be added after one +unambiguous contract ships. + +### Storage + +``` +~/.qwen/boards/{board}/ + tasks/{id}.json + asks/{id}.json +``` + +Directories use mode `0700`; files use `0600`. Each item is one versioned JSON object. +Identifiers have a type prefix and UUID suffix. Creating an item never scans for or reuses a +numeric id. + +All read-modify-write transitions and pruning use the same per-item lock discipline. Creation +uses exclusive semantics. A command must not decide that a target is stale and then mutate +it after releasing the lock. + +Readers validate each record. A list operation reports and skips malformed records so one +bad file does not hide the rest of the board. A mutation targeting a malformed record fails +without rewriting it. + +### Items + +| Item | Purpose | Terminal states | +| ------ | ---------------------------------------- | --------------------------------- | +| `task` | Work with an owner, status, and notes | pending / in_progress / completed | +| `ask` | A question addressed to a declared label | answered / declined / timeout | + +An `ask` is addressed to a label, not a registered session. A receiver chooses the same +label with `--as` and answers it. If nobody does, its deadline determines `timeout`; no +background sweeper is required. + +There is no generic message. Status belongs on a task and information requests are asks. + +### CLI + +The MVP surface is non-interactive and machine-readable: + +```text +qwen board show --board [--as ] [--json] +qwen board task --board --as [--owner ] +qwen board claim --board --as +qwen board done --board --as [--note ] +qwen board ask --board --as [--wait] [--timeout ] +qwen board answer --board --as +qwen board decline --board --as +qwen board prune --board --as --older-than +``` + +`--json` produces stable data without ANSI output. Human output may be formatted but must not +truncate identifiers or state needed to act. `--wait` uses bounded polling, returns a +distinct timeout exit code, and does not start a daemon or socket. + +Unknown ids, invalid transitions, invalid names, malformed target records, and lock failures +are errors. An absent board is an empty result for `show` and an error for mutations that +need an existing item. + +## Authority and security boundary + +The local OS account is the access boundary. File permissions prevent other local users from +reading the board. `--as` is self-declared and must never authorize filesystem access, +dangerous tools, approval mode, or sandbox changes. + +Board text is untrusted input. Consumers display or summarize it as data; they do not inject +it as a user message or automatically execute instructions from it. + +Push delivery, if added, requires a receiver-side consent gate before the first send path. +That later gate must fail closed and cannot trust the sender's declared `--as` value. + +## Board implementation stages + +### Stage 1 — storage primitives + +Add versioned task and ask records with validation, secure permissions, one lock discipline, +random ids, and focused transition tests. + +Observable result: two processes can safely create, claim, answer, and prune items on the +same named board without lost updates or id reuse. + +### Stage 2 — CLI + +Expose the storage primitives through the explicit `--board` / `--as` commands above. + +Observable result: two independently started agents, including a non-Qwen agent, can share +work by running commands and can distinguish completed, declined, and timed-out outcomes +from exit status and JSON. + +### Stage 3 — Qwen-native consumer + +After the CLI contract is stable, a Qwen-native tool, slash command, footer indicator, or +turn-boundary poll may consume it. Every native action must map exactly to an existing CLI +operation. + +Observable result: Qwen users get lower-friction access without changing board semantics or +excluding foreign agents. + +### Stage 4 — orchestration or push + +Fleet/tmux startup, a Codex runner, and Qwen-to-Qwen wake delivery are separate features. +Fleet may pass explicit board and identity arguments to child commands; it does not create +membership. Push lands only with its receiver-side consent gate and remains a latency +optimization for board correctness, while still being necessary for low-latency delegation +to an already-running Qwen session. + +Observable result: a user can ask a lead agent to start or address a worker and receive its +result without manually operating the board CLI. Removing this stage leaves the Stage 2 +storage contract correct, but removes the user-visible delegation experience. + +## Explicit non-goals for the MVP + +- membership, participant records, join/leave, roster, name claiming, or liveness coupling; +- implicit project boards, ambient identity, or process-global board context; +- `/board`, footer badges, background polling, or native agent tools; +- fleet/tmux orchestration, PTY attachment, or lifecycle management; +- push, broadcast, remote access, or cross-machine synchronization; +- multiple agents writing the same checkout; +- a public compatibility promise for the on-disk format. + +The existing agent-view PR series (#7799–#7803) owns supervised terminals and roster UI. It +is independent of this board and is neither removed nor extended by the MVP. + +## Board implementation acceptance gate + +The MVP is ready only when: + +- its implementation contains no participant or join/leave subsystem; +- every mutation has explicit board and actor arguments; +- concurrent creators cannot reuse ids or overwrite each other; +- pruning cannot delete an item changed after eligibility was checked; +- malformed records cannot crash or hide a healthy board; +- human output preserves actionable ids and JSON output remains parseable; +- focused unit tests cover the above, followed by package typecheck and build. + +Passing this gate means the board implementation is technically sound. It does not by itself +mean the cross-session collaboration product is complete or that this layer must land before +the Qwen-to-Qwen sender path. A standalone merge requires an explicit maintainer decision to +ship the low-level CLI; otherwise it should land with a concrete native consumer or runner. diff --git a/docs/users/features/_meta.ts b/docs/users/features/_meta.ts index ed7d9ef7260..65892942185 100644 --- a/docs/users/features/_meta.ts +++ b/docs/users/features/_meta.ts @@ -6,6 +6,7 @@ export default { 'markdown-rendering': 'Markdown Rendering', 'sub-agents': 'SubAgents', 'multi-agent-coordination': 'Multi-Agent Coordination', + 'agent-board': 'Agent Board', arena: 'Agent Arena', skills: 'Skills', memory: 'Memory', diff --git a/docs/users/features/agent-board.md b/docs/users/features/agent-board.md new file mode 100644 index 00000000000..a692fdcfb04 --- /dev/null +++ b/docs/users/features/agent-board.md @@ -0,0 +1,79 @@ +# Agent Board + +Agent Board lets independently started agents share work through files on the +same machine. It does not start, join, monitor, or send input to agent processes. + +It is a low-level interoperability surface, not the Qwen Agent Team scheduler or +the cross-session messaging transport. A task owner is only a recorded label; +it does not start or wake a Qwen Code, Codex, or other agent process. + +> Experimental. The on-disk format may change between releases. + +## Use a board + +Every command names the board explicitly. Every command that changes the board +also declares the actor with `--as`. + +```bash +qwen board task "check the API response" --board orders --as api +qwen board show --board orders +``` + +The first command prints a task id. Another agent can claim and complete it: + +```bash +qwen board claim --board orders --as web +qwen board done --board orders --as web --note "status is numeric" +``` + +`--as` is a label recorded with the action, not authentication. There is no +membership list, join command, heartbeat, or reserved participant name. + +## Ask a question + +```bash +qwen board ask web "does the client parse status as text?" \ + --board orders --as api --wait +``` + +The receiver uses the same label when answering or declining: + +```bash +qwen board answer "yes" --board orders --as web +qwen board decline "not my area" --board orders --as web +``` + +With `--wait`, exit code `0` means answered, `2` declined, `3` the ask's TTL +expired, and `4` the local wait ended while the ask was still open. `--timeout` +sets the local wait in seconds; `--ttl` sets the ask lifetime in seconds. + +## Machine-readable output + +Add `--json` to receive JSON without ANSI formatting: + +```bash +qwen board show --board orders --as web --json +``` + +Passing `--as` to `show` filters tasks to that owner and asks to or from that +actor. + +## Housekeeping + +Settled records remain until explicitly pruned: + +```bash +qwen board prune --board orders --as human --older-than 7 +``` + +The cutoff is in days. Pruning rechecks each record while holding its lock, so +an item changed after the scan is not deleted from stale information. + +## Limits + +- Boards live under `~/.qwen/boards/` and are scoped to the current OS user. +- Nothing is pushed into an agent. Each participant chooses when to read. +- Board text is untrusted data and is never automatically executed. +- Multiple agents writing the same checkout is not supported. +- Slash commands, footer polling, fleet/tmux orchestration, and remote boards + are not part of this first version. diff --git a/packages/cli/src/cli.test.ts b/packages/cli/src/cli.test.ts index 44bf2b2ba3b..b367ec1241f 100644 --- a/packages/cli/src/cli.test.ts +++ b/packages/cli/src/cli.test.ts @@ -1061,6 +1061,7 @@ describe('bootstrap import boundaries', () => { const configSource = readFileSync('src/config/config.ts', 'utf8'); const commandNameByIdentifier = new Map([ ['authCommand', 'auth'], + ['boardCommand', 'board'], ['channelCommand', 'channel'], ['extensionsCommand', 'extensions'], ['hooksCommand', 'hooks'], diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts index 7e2929bce21..29a07acad8d 100644 --- a/packages/cli/src/cli.ts +++ b/packages/cli/src/cli.ts @@ -31,6 +31,7 @@ type BootstrapRoute = 'serve' | 'mcp' | 'help' | 'version' | 'default'; export const TOP_LEVEL_COMMANDS = [ ['auth', 'Configure authentication (removed)'], + ['board ', 'Share work with other agents through a board'], ['channel ', 'Manage messaging channels (Telegram, Discord, etc.)'], ['extensions ', 'Manage Qwen Code extensions.'], ['hooks', 'Manage Qwen Code hooks (use /hooks in interactive mode).'], diff --git a/packages/cli/src/commands/board.ts b/packages/cli/src/commands/board.ts new file mode 100644 index 00000000000..204e804fce7 --- /dev/null +++ b/packages/cli/src/commands/board.ts @@ -0,0 +1,309 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { Argv, CommandModule } from 'yargs'; +import { + answerAsk, + claimBoardTask, + completeBoardTask, + createAsk, + createBoardTask, + declineAsk, + getAsk, + listAsks, + listBoardTasks, + pruneAsks, + pruneBoardTasks, +} from '@qwen-code/qwen-code-core/board'; +import { sanitizeTerminalText } from '../ui/utils/textUtils.js'; +import { requireActorName, requireBoardName } from './board/context.js'; +import { oneLine, renderBoard, type BoardSnapshot } from './board/render.js'; + +interface CommonArgs { + board?: string; + as?: string; + json?: boolean; +} + +async function emit( + argv: CommonArgs, + value: unknown, + human: string, +): Promise { + // Sanitize without flattening: the `show` panel and `ask --wait` answers + // are genuinely multi-line, and `sanitizeTerminalText` keeps LF/TAB while + // neutralizing dangerous control sequences. + const text = `${argv.json ? JSON.stringify(value) : sanitizeTerminalText(human)}\n`; + // `parseArguments` calls `process.exit` as soon as this handler resolves. A + // write to a pipe is asynchronous, so resolving before stdout has drained + // truncates output larger than the pipe buffer while still exiting 0. + // + // A reader that closes early (`... | head`) reports that twice: once to the + // write callback and once as a stream `error` event. The listener keeps the + // second one from surfacing as an uncaught exception, and the callback + // decides the outcome — a closed reader ends the pipeline normally, while + // any other write failure still has to reach the caller. + if (process.stdout.listenerCount('error') === 0) { + process.stdout.on('error', () => {}); + } + await new Promise((resolve, reject) => { + process.stdout.write(text, (err) => { + if (err && (err as NodeJS.ErrnoException).code !== 'EPIPE') reject(err); + else resolve(); + }); + }); +} + +async function run(fn: () => Promise): Promise { + try { + await fn(); + } catch (err) { + process.stderr.write( + `${oneLine(err instanceof Error ? err.message : String(err))}\n`, + ); + process.exitCode = 1; + } +} + +function finiteNumber(value: unknown, flag: string, minimum: number): number { + if (typeof value !== 'number' || !Number.isFinite(value) || value < minimum) { + throw new Error(`${flag} must be a finite number >= ${minimum}.`); + } + return value; +} + +async function snapshot(board: string, actor?: string): Promise { + const [tasks, asks] = await Promise.all([ + listBoardTasks(board), + listAsks(board), + ]); + return { + board, + tasks: actor ? tasks.filter((task) => task.owner === actor) : tasks, + asks: actor + ? asks.filter((ask) => ask.from === actor || ask.to === actor) + : asks, + }; +} + +export const boardCommand: CommandModule = { + command: 'board', + describe: 'Share work with other agents through a board', + builder: (yargs: Argv) => + yargs + .option('board', { + type: 'string', + describe: 'Board name', + }) + .option('as', { + type: 'string', + describe: 'Declared actor name', + }) + .option('json', { type: 'boolean', describe: 'Emit JSON' }) + + .command({ + command: 'show', + describe: 'Print the board once', + handler: (argv) => + run(async () => { + const a = argv as CommonArgs; + const board = requireBoardName(a.board); + const actor = + a.as === undefined ? undefined : requireActorName(a.as); + const state = await snapshot(board, actor); + await emit(a, state, renderBoard(state)); + }), + }) + + .command({ + command: 'task ', + describe: 'Create a task', + builder: (y: Argv) => + y + .positional('subject', { type: 'string', demandOption: true }) + .option('owner', { type: 'string' }), + handler: (argv) => + run(async () => { + const a = argv as CommonArgs & { subject: string; owner?: string }; + const task = await createBoardTask({ + board: requireBoardName(a.board), + createdBy: requireActorName(a.as), + subject: a.subject, + owner: a.owner, + }); + await emit(a, task, `${task.id} ${task.subject}`); + }), + }) + + .command({ + command: 'claim ', + describe: 'Take ownership of a task', + builder: (y: Argv) => + y.positional('id', { type: 'string', demandOption: true }), + handler: (argv) => + run(async () => { + const a = argv as CommonArgs & { id: string }; + const task = await claimBoardTask( + requireBoardName(a.board), + a.id, + requireActorName(a.as), + ); + await emit(a, task, `${task.id} claimed by ${task.owner}`); + }), + }) + + .command({ + command: 'done ', + describe: 'Complete a task you own', + builder: (y: Argv) => + y + .positional('id', { type: 'string', demandOption: true }) + .option('note', { type: 'string' }), + handler: (argv) => + run(async () => { + const a = argv as CommonArgs & { id: string; note?: string }; + const task = await completeBoardTask( + requireBoardName(a.board), + a.id, + requireActorName(a.as), + a.note, + ); + await emit(a, task, `${task.id} completed`); + }), + }) + + .command({ + command: 'ask ', + describe: 'Ask another actor a question', + builder: (y: Argv) => + y + .positional('to', { type: 'string', demandOption: true }) + .positional('question', { type: 'string', demandOption: true }) + .option('about', { type: 'string' }) + .option('wait', { type: 'boolean' }) + .option('timeout', { type: 'number', default: 30 }) + .option('ttl', { type: 'number', default: 900 }), + handler: (argv) => + run(async () => { + const a = argv as CommonArgs & { + to: string; + question: string; + about?: string; + wait?: boolean; + timeout: number; + ttl: number; + }; + const board = requireBoardName(a.board); + // Validate the local wait before creating the ask: a rejected + // `--timeout` must not leave an orphaned open ask on the board. + const waitMs = a.wait + ? finiteNumber(a.timeout * 1000, '--timeout', 0) + : 0; + const ask = await createAsk({ + board, + from: requireActorName(a.as), + to: a.to, + question: a.question, + aboutTask: a.about, + ttlMs: finiteNumber(a.ttl, '--ttl', 0.001) * 1000, + }); + if (!a.wait) { + await emit(a, ask, `${ask.id} -> ${ask.to}`); + return; + } + + const deadline = Date.now() + waitMs; + for (;;) { + const current = await getAsk(board, ask.id); + if (!current) throw new Error(`Ask "${ask.id}" not found.`); + if (current.state !== 'open') { + await emit( + a, + current, + current.state === 'answered' + ? (current.answer ?? '') + : current.state, + ); + if (current.state === 'declined') process.exitCode = 2; + if (current.state === 'timeout') process.exitCode = 3; + return; + } + if (Date.now() >= deadline) { + process.stderr.write(`${ask.id} is still open.\n`); + process.exitCode = 4; + return; + } + await new Promise((resolve) => setTimeout(resolve, 250)); + } + }), + }) + + .command({ + command: 'answer ', + describe: 'Answer an ask addressed to you', + builder: (y: Argv) => + y + .positional('id', { type: 'string', demandOption: true }) + .positional('answer', { type: 'string', demandOption: true }), + handler: (argv) => + run(async () => { + const a = argv as CommonArgs & { id: string; answer: string }; + const ask = await answerAsk( + requireBoardName(a.board), + a.id, + requireActorName(a.as), + a.answer, + ); + await emit(a, ask, `${ask.id} answered`); + }), + }) + + .command({ + command: 'decline ', + describe: 'Decline an ask addressed to you', + builder: (y: Argv) => + y + .positional('id', { type: 'string', demandOption: true }) + .positional('reason', { type: 'string', demandOption: true }), + handler: (argv) => + run(async () => { + const a = argv as CommonArgs & { id: string; reason: string }; + const ask = await declineAsk( + requireBoardName(a.board), + a.id, + requireActorName(a.as), + a.reason, + ); + await emit(a, ask, `${ask.id} declined`); + }), + }) + + .command({ + command: 'prune', + describe: 'Remove settled items older than a cutoff', + builder: (y: Argv) => + y.option('older-than', { type: 'number', default: 7 }), + handler: (argv) => + run(async () => { + const a = argv as CommonArgs & { olderThan: number }; + requireActorName(a.as); + const board = requireBoardName(a.board); + const cutoff = + finiteNumber(a.olderThan, '--older-than', 0) * 86_400_000; + const [asks, tasks] = await Promise.all([ + pruneAsks(board, cutoff), + pruneBoardTasks(board, cutoff), + ]); + const removed = { asks, tasks }; + const total = asks.length + tasks.length; + await emit(a, removed, `Removed ${total} settled items.`); + }), + }) + + .demandCommand(1, 'You need at least one board command.') + .version(false), + handler: () => {}, +}; diff --git a/packages/cli/src/commands/board/board-cli.test.ts b/packages/cli/src/commands/board/board-cli.test.ts new file mode 100644 index 00000000000..8d2447853f1 --- /dev/null +++ b/packages/cli/src/commands/board/board-cli.test.ts @@ -0,0 +1,303 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import yargs from 'yargs'; +import { + afterEach, + beforeEach, + describe, + expect, + it, + vi, + type MockInstance, +} from 'vitest'; + +const core = vi.hoisted(() => ({ + answerAsk: vi.fn(), + assertSafeName: vi.fn(), + claimBoardTask: vi.fn(), + completeBoardTask: vi.fn(), + createAsk: vi.fn(), + createBoardTask: vi.fn(), + declineAsk: vi.fn(), + getAsk: vi.fn(), + listAsks: vi.fn(), + listBoardTasks: vi.fn(), + pruneAsks: vi.fn(), + pruneBoardTasks: vi.fn(), +})); + +vi.mock('@qwen-code/qwen-code-core/board', () => core); + +import { boardCommand } from '../board.js'; +import { renderBoard } from './render.js'; + +const TASK_ID = 't-00000000-0000-4000-8000-000000000001'; +const ASK_ID = 'a-00000000-0000-4000-8000-000000000002'; + +async function parse(command: string): Promise { + await yargs(command.split(' ')) + .command(boardCommand) + .exitProcess(false) + .fail(false) + .parseAsync(); +} + +// Real stdout invokes the write callback once the chunk is flushed, and `emit` +// waits for it, so a faithful stub has to invoke it too. +function stubWrite(): typeof process.stdout.write { + return (( + _chunk: unknown, + encodingOrCb?: unknown, + cb?: (err?: Error | null) => void, + ) => { + const done = typeof encodingOrCb === 'function' ? encodingOrCb : cb; + (done as ((err?: Error | null) => void) | undefined)?.(null); + return true; + }) as unknown as typeof process.stdout.write; +} + +function written(spy: MockInstance): string { + return spy.mock.calls.map((call) => String(call[0])).join(''); +} + +describe('board CLI', () => { + let stdout: MockInstance; + let stderr: MockInstance; + + beforeEach(() => { + vi.clearAllMocks(); + process.exitCode = undefined; + stdout = vi.spyOn(process.stdout, 'write').mockImplementation(stubWrite()); + stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true); + core.listBoardTasks.mockResolvedValue([]); + core.listAsks.mockResolvedValue([]); + }); + + afterEach(() => { + stdout.mockRestore(); + stderr.mockRestore(); + process.exitCode = undefined; + }); + + it('requires an explicit board', async () => { + await parse('board show'); + expect(stderr).toHaveBeenCalledWith('Pass --board .\n'); + expect(process.exitCode).toBe(1); + }); + + it('requires and forwards the actor for mutations', async () => { + core.createBoardTask.mockResolvedValue({ id: TASK_ID, subject: 'check' }); + await parse('board task check --board demo --as author --json'); + expect(core.createBoardTask).toHaveBeenCalledWith({ + board: 'demo', + createdBy: 'author', + subject: 'check', + owner: undefined, + }); + expect(stdout).toHaveBeenCalledWith( + `${JSON.stringify({ id: TASK_ID, subject: 'check' })}\n`, + expect.any(Function), + ); + }); + + it('passes the declared actor when answering', async () => { + core.answerAsk.mockResolvedValue({ id: ASK_ID, state: 'answered' }); + await parse(`board answer ${ASK_ID} yes --board demo --as web --json`); + expect(core.answerAsk).toHaveBeenCalledWith('demo', ASK_ID, 'web', 'yes'); + }); + + it.each([ + ['answered', undefined], + ['declined', 2], + ['timeout', 3], + ] as const)('maps an ask %s outcome to exit code %s', async (state, code) => { + core.createAsk.mockResolvedValue({ id: ASK_ID, to: 'web' }); + core.getAsk.mockResolvedValue({ id: ASK_ID, state, answer: 'yes' }); + await parse( + 'board ask web question --board demo --as api --wait --timeout 1 --ttl 1 --json', + ); + expect(process.exitCode).toBe(code); + }); + + it('uses exit code 4 when the local wait ends first', async () => { + core.createAsk.mockResolvedValue({ id: ASK_ID, to: 'web' }); + core.getAsk.mockResolvedValue({ id: ASK_ID, state: 'open' }); + await parse( + 'board ask web question --board demo --as api --wait --timeout 0 --ttl 1', + ); + expect(process.exitCode).toBe(4); + }); + + it('reports an ask removed while waiting as missing', async () => { + core.createAsk.mockResolvedValue({ id: ASK_ID, to: 'web' }); + core.getAsk.mockResolvedValue(null); + await parse( + 'board ask web question --board demo --as api --wait --timeout 0 --ttl 1', + ); + expect(stderr).toHaveBeenCalledWith(`Ask "${ASK_ID}" not found.\n`); + expect(process.exitCode).toBe(1); + }); + + it('sanitizes human output and errors', async () => { + core.createBoardTask.mockResolvedValue({ + id: TASK_ID, + subject: 'check\x1b]52;c;pw\x07', + }); + await parse('board task check --board demo --as author'); + expect(written(stdout)).not.toContain('\x1b'); + expect(written(stdout)).not.toContain('\x07'); + + core.listBoardTasks.mockRejectedValue(new Error('bad\x1b]52;c;pw\x07')); + await parse('board show --board demo'); + expect(stderr.mock.calls.flat().join('')).not.toContain('\x1b'); + expect(stderr.mock.calls.flat().join('')).not.toContain('\x07'); + }); + + it('keeps the show panel multi-line while still sanitizing it', async () => { + core.listBoardTasks.mockResolvedValue([ + { + id: TASK_ID, + subject: 'first\nsecond', + status: 'in_progress', + owner: 'worker', + }, + { + id: 't-00000000-0000-4000-8000-000000000003', + subject: 'third\x1b]52;c;pw\x07', + status: 'pending', + owner: null, + }, + ]); + await parse('board show --board demo'); + const out = written(stdout); + expect(out.trim().split('\n')).toHaveLength(3); + expect(out).toContain('first second'); + expect(out).not.toContain('\x1b'); + expect(out).not.toContain('\x07'); + }); + + it('prints a multi-line ask answer without flattening it', async () => { + core.createAsk.mockResolvedValue({ id: ASK_ID, to: 'web' }); + core.getAsk.mockResolvedValue({ + id: ASK_ID, + state: 'answered', + answer: 'line one\nline two', + }); + await parse( + 'board ask web question --board demo --as api --wait --timeout 1 --ttl 1', + ); + expect(stdout).toHaveBeenCalledWith( + 'line one\nline two\n', + expect.any(Function), + ); + }); + + it('rejects a negative prune cutoff before deleting', async () => { + await parse('board prune --board demo --as human --older-than -1'); + expect(core.pruneAsks).not.toHaveBeenCalled(); + expect(process.exitCode).toBe(1); + }); + + it.each(['-5', 'abc'])( + 'rejects --timeout %s before creating the ask', + async (timeout) => { + await parse( + `board ask web question --board demo --as api --wait --timeout=${timeout}`, + ); + expect(core.createAsk).not.toHaveBeenCalled(); + expect(stderr).toHaveBeenCalledWith( + '--timeout must be a finite number >= 0.\n', + ); + expect(process.exitCode).toBe(1); + }, + ); + + it('waits for stdout to drain before the handler resolves', async () => { + let flush: (() => void) | undefined; + stdout.mockImplementation((( + _chunk: unknown, + cb?: (err?: Error | null) => void, + ) => { + flush = () => cb?.(null); + return false; + }) as unknown as typeof process.stdout.write); + + let settled = false; + const parsed = parse('board show --board demo --json').then(() => { + settled = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + // `parseArguments` calls process.exit once this resolves, so it must not + // resolve while a pipe write is still queued. + expect(settled).toBe(false); + expect(flush).toBeDefined(); + + flush?.(); + await parsed; + expect(settled).toBe(true); + }); + + it('stops quietly when the reader closes the pipe', async () => { + const epipe: NodeJS.ErrnoException = new Error('write EPIPE'); + epipe.code = 'EPIPE'; + stdout.mockImplementation((( + _chunk: unknown, + cb?: (err?: Error | null) => void, + ) => { + // Real stdout reports a closed reader to the callback and then re-emits + // it on the stream. + cb?.(epipe); + process.stdout.emit('error', epipe); + return false; + }) as unknown as typeof process.stdout.write); + + await parse('board show --board demo --json'); + expect(written(stderr)).toBe(''); + expect(process.exitCode).toBeUndefined(); + }); + + it('reports a write failure that is not a closed reader', async () => { + const enospc: NodeJS.ErrnoException = new Error('no space left on device'); + enospc.code = 'ENOSPC'; + stdout.mockImplementation((( + _chunk: unknown, + cb?: (err?: Error | null) => void, + ) => { + cb?.(enospc); + return false; + }) as unknown as typeof process.stdout.write); + + await parse('board show --board demo --json'); + expect(written(stderr)).toContain('no space left on device'); + expect(process.exitCode).toBe(1); + }); +}); + +describe('board rendering', () => { + it('keeps full actionable ids and one line per item', () => { + const output = renderBoard({ + board: 'demo', + tasks: [ + { + schemaVersion: 1, + id: TASK_ID, + subject: 'first\nsecond', + createdBy: 'author', + owner: 'worker', + status: 'in_progress', + createdAt: 1, + updatedAt: 1, + notes: [], + }, + ], + asks: [], + }); + expect(output).toContain(TASK_ID); + expect(output).toContain('first second'); + expect(output.split('\n')).toHaveLength(2); + }); +}); diff --git a/packages/cli/src/commands/board/context.ts b/packages/cli/src/commands/board/context.ts new file mode 100644 index 00000000000..30fe5afbeff --- /dev/null +++ b/packages/cli/src/commands/board/context.ts @@ -0,0 +1,23 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { assertSafeName } from '@qwen-code/qwen-code-core/board'; + +export function requireBoardName(value: unknown): string { + if (typeof value !== 'string' || !value) { + throw new Error('Pass --board .'); + } + assertSafeName('board name', value); + return value; +} + +export function requireActorName(value: unknown): string { + if (typeof value !== 'string' || !value) { + throw new Error('Pass --as .'); + } + assertSafeName('actor name', value); + return value; +} diff --git a/packages/cli/src/commands/board/render.ts b/packages/cli/src/commands/board/render.ts new file mode 100644 index 00000000000..1acd493989b --- /dev/null +++ b/packages/cli/src/commands/board/render.ts @@ -0,0 +1,38 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { + AskRecord, + BoardTaskRecord, +} from '@qwen-code/qwen-code-core/board'; +import { sanitizeTerminalText } from '../../ui/utils/textUtils.js'; + +export interface BoardSnapshot { + board: string; + tasks: BoardTaskRecord[]; + asks: AskRecord[]; +} + +export function oneLine(value: string): string { + return sanitizeTerminalText(value).replace(/[\r\n\t]+/g, ' '); +} + +export function renderBoard(snapshot: BoardSnapshot): string { + const lines = [`board: ${oneLine(snapshot.board)}`]; + for (const ask of snapshot.asks) { + lines.push( + `? ${ask.id} [${ask.state}] ${oneLine(ask.from)} -> ${oneLine(ask.to)}: ${oneLine(ask.question)}`, + ); + } + for (const task of snapshot.tasks) { + lines.push( + `- ${task.id} [${task.status}] ${oneLine(task.owner ?? 'unowned')}: ${oneLine(task.subject)}`, + ); + } + + if (lines.length === 1) lines.push('(empty)'); + return lines.join('\n'); +} diff --git a/packages/cli/src/config/config.ts b/packages/cli/src/config/config.ts index bf538d25955..b60ed38f2a8 100755 --- a/packages/cli/src/config/config.ts +++ b/packages/cli/src/config/config.ts @@ -73,6 +73,7 @@ import { authCommand } from '../commands/auth.js'; import { reviewCommand } from '../commands/review.js'; import { serveCommand } from '../commands/serve.js'; import { sessionsCommand } from '../commands/sessions.js'; +import { boardCommand } from '../commands/board.js'; import { updateCommand } from '../commands/update.js'; import { isValidSessionId } from './session-id.js'; @@ -1087,6 +1088,7 @@ export async function parseArguments(): Promise { .command(hooksCommand) // Register Channel subcommands .command(channelCommand) + .command(boardCommand) // Register /review skill helpers (presubmit checks, cleanup) .command(reviewCommand) // Register `qwen serve` (Stage 1 daemon) @@ -1120,6 +1122,7 @@ export async function parseArguments(): Promise { result._[0] === 'channel' || result._[0] === 'review' || result._[0] === 'sessions' || + result._[0] === 'board' || result._[0] === 'update') ) { // Note: `serve` is intentionally NOT in this list. Its handler blocks diff --git a/packages/cli/vitest.config.ts b/packages/cli/vitest.config.ts index 7f13fb4d8e8..c92995b3835 100644 --- a/packages/cli/vitest.config.ts +++ b/packages/cli/vitest.config.ts @@ -31,6 +31,10 @@ export default defineConfig({ __dirname, '../core/src/services/tool-write-origin.ts', ), + '@qwen-code/qwen-code-core/board': path.resolve( + __dirname, + '../core/src/board.ts', + ), '@qwen-code/qwen-code-core': path.resolve(__dirname, '../core/index.ts'), // cli's daemon-status-provider.test.ts imports `FakeAgent` / // `makeChannel` from acp-bridge's package-private diff --git a/packages/core/package.json b/packages/core/package.json index 07b1f7ac892..6795c2398a2 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -33,6 +33,10 @@ "types": "./dist/src/hooks/user-prompt-submit-context.d.ts", "import": "./dist/src/hooks/user-prompt-submit-context.js" }, + "./board": { + "types": "./dist/src/board.d.ts", + "import": "./dist/src/board.js" + }, "./package.json": "./package.json", "./dist/*": "./dist/*", "./src/*": "./src/*" diff --git a/packages/core/src/agents/team/asks.ts b/packages/core/src/agents/team/asks.ts new file mode 100644 index 00000000000..70f4bc14ef3 --- /dev/null +++ b/packages/core/src/agents/team/asks.ts @@ -0,0 +1,288 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import * as fs from 'node:fs/promises'; +import * as path from 'node:path'; +import { atomicWriteJSON } from '../../utils/atomicFileWrite.js'; +import { createDebugLogger } from '../../utils/debugLogger.js'; +import { isNodeError } from '../../utils/errors.js'; +import { + assertItemId, + assertSafeName, + createBoardRecord, + getCollectionDir, + pruneCollection, + withItemLock, +} from './board-lock.js'; + +const debug = createDebugLogger('BOARD_ASKS'); + +export const ASKS_COLLECTION = 'asks'; +export const DEFAULT_ASK_TTL_MS = 15 * 60 * 1000; +const MAX_TEXT_LENGTH = 65536; +const ASK_FILE = + /^a-[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\.json$/; + +export type AskState = 'open' | 'answered' | 'declined' | 'timeout'; + +export interface AskRecord { + schemaVersion: 1; + id: string; + from: string; + to: string; + question: string; + aboutTask?: string; + state: AskState; + createdAt: number; + expiresAt: number; + answer: string | null; + reason: string | null; + settledAt: number | null; +} + +function asksDir(board: string): string { + return getCollectionDir(board, ASKS_COLLECTION); +} + +function askPath(board: string, id: string): string { + return path.join(asksDir(board), `${id}.json`); +} + +function assertText(field: string, value: string): void { + if (!value.trim()) throw new Error(`${field} must not be empty.`); + if (value.length > MAX_TEXT_LENGTH) { + throw new Error(`${field} exceeds ${MAX_TEXT_LENGTH} characters.`); + } +} + +function parseAsk(value: unknown): AskRecord { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error('Ask record must be an object.'); + } + const ask = value as Partial; + if (ask.schemaVersion !== 1) throw new Error('Unsupported ask schema.'); + if (typeof ask.id !== 'string') throw new Error('Ask id must be text.'); + assertItemId('ask id', ask.id, 'a'); + if (typeof ask.from !== 'string' || typeof ask.to !== 'string') { + throw new Error('Ask actors must be names.'); + } + assertSafeName('actor name', ask.from); + assertSafeName('actor name', ask.to); + if (typeof ask.question !== 'string') { + throw new Error('Ask question must be text.'); + } + assertText('question', ask.question); + if (ask.aboutTask !== undefined) { + if (typeof ask.aboutTask !== 'string') { + throw new Error('Ask aboutTask must be a task id.'); + } + assertItemId('task id', ask.aboutTask, 't'); + } + if (!['open', 'answered', 'declined'].includes(ask.state ?? '')) { + throw new Error('Invalid ask state.'); + } + if (!Number.isFinite(ask.createdAt) || !Number.isFinite(ask.expiresAt)) { + throw new Error('Ask timestamps must be finite numbers.'); + } + if ((ask.expiresAt ?? 0) <= (ask.createdAt ?? 0)) { + throw new Error('Ask expiresAt must follow createdAt.'); + } + if (ask.answer !== null && typeof ask.answer !== 'string') { + throw new Error('Ask answer must be text or null.'); + } + if (ask.reason !== null && typeof ask.reason !== 'string') { + throw new Error('Ask reason must be text or null.'); + } + if (ask.settledAt !== null && !Number.isFinite(ask.settledAt)) { + throw new Error('Ask settledAt must be a finite number or null.'); + } + if (ask.settledAt !== null && (ask.settledAt ?? 0) < (ask.createdAt ?? 0)) { + throw new Error('Ask settledAt precedes createdAt.'); + } + if (ask.answer) assertText('answer', ask.answer); + if (ask.reason) assertText('reason', ask.reason); + if (ask.state === 'open') { + if (ask.answer !== null || ask.reason !== null || ask.settledAt !== null) { + throw new Error('An open ask cannot contain a result.'); + } + } else if (ask.state === 'answered') { + if (!ask.answer || ask.reason !== null || ask.settledAt === null) { + throw new Error('An answered ask requires answer and settledAt.'); + } + } else if (!ask.reason || ask.answer !== null || ask.settledAt === null) { + throw new Error('A declined ask requires reason and settledAt.'); + } + return ask as AskRecord; +} + +function settleAsk(ask: AskRecord, now = Date.now()): AskRecord { + if (ask.state !== 'open' || now < ask.expiresAt) return ask; + return { ...ask, state: 'timeout', settledAt: ask.expiresAt }; +} + +export async function createAsk(opts: { + board: string; + from: string; + to: string; + question: string; + aboutTask?: string; + ttlMs?: number; +}): Promise { + assertSafeName('actor name', opts.from); + assertSafeName('actor name', opts.to); + if (opts.from === opts.to) + throw new Error('An ask must target another actor.'); + assertText('question', opts.question); + if (opts.aboutTask) assertItemId('task id', opts.aboutTask, 't'); + const ttl = opts.ttlMs ?? DEFAULT_ASK_TTL_MS; + if (!Number.isFinite(ttl) || ttl <= 0) { + throw new Error('ttlMs must be a positive finite number.'); + } + const now = Date.now(); + if (!Number.isFinite(now + ttl)) throw new Error('ttlMs is too large.'); + return createBoardRecord(opts.board, ASKS_COLLECTION, 'a', (id) => ({ + schemaVersion: 1, + id, + from: opts.from, + to: opts.to, + question: opts.question, + ...(opts.aboutTask ? { aboutTask: opts.aboutTask } : {}), + state: 'open', + createdAt: now, + expiresAt: now + ttl, + answer: null, + reason: null, + settledAt: null, + })); +} + +export async function getAsk( + board: string, + id: string, +): Promise { + assertSafeName('board name', board); + assertItemId('ask id', id, 'a'); + let raw: string; + try { + raw = await fs.readFile(askPath(board, id), 'utf8'); + } catch (err) { + if (isNodeError(err) && err.code === 'ENOENT') return null; + throw err; + } + try { + const ask = parseAsk(JSON.parse(raw)); + if (ask.id !== id) throw new Error('Ask id does not match its filename.'); + return settleAsk(ask); + } catch (err) { + debug.warn(`skipping invalid ask ${id}:`, err); + return null; + } +} + +export async function listAsks(board: string): Promise { + assertSafeName('board name', board); + let files: string[]; + try { + files = await fs.readdir(asksDir(board)); + } catch (err) { + if (isNodeError(err) && err.code === 'ENOENT') return []; + throw err; + } + const asks = ( + await Promise.all( + files + .filter((file) => ASK_FILE.test(file)) + .map((file) => getAsk(board, file.slice(0, -5))), + ) + ).filter((ask): ask is AskRecord => ask !== null); + return asks.sort((a, b) => a.createdAt - b.createdAt); +} + +async function settleOnDisk( + board: string, + id: string, + by: string, + apply: (ask: AskRecord) => AskRecord, +): Promise { + assertSafeName('board name', board); + assertItemId('ask id', id, 'a'); + assertSafeName('actor name', by); + const target = askPath(board, id); + return withItemLock( + target, + async () => { + const current = settleAsk( + parseAsk(JSON.parse(await fs.readFile(target, 'utf8'))), + ); + if (current.id !== id) { + throw new Error('Ask id does not match its filename.'); + } + if (current.to !== by) { + throw new Error(`Ask "${id}" is addressed to "${current.to}".`); + } + if (current.state !== 'open') { + throw new Error(`Ask "${id}" is already ${current.state}.`); + } + const next = parseAsk(apply(current)); + await atomicWriteJSON(target, next, { mode: 0o600, forceMode: true }); + return next; + }, + () => { + throw new Error(`Ask "${id}" not found.`); + }, + ); +} + +export function answerAsk( + board: string, + id: string, + by: string, + answer: string, +): Promise { + assertText('answer', answer); + return settleOnDisk(board, id, by, (ask) => ({ + ...ask, + state: 'answered', + answer, + settledAt: Date.now(), + })); +} + +export function declineAsk( + board: string, + id: string, + by: string, + reason: string, +): Promise { + assertText('reason', reason); + return settleOnDisk(board, id, by, (ask) => ({ + ...ask, + state: 'declined', + reason, + settledAt: Date.now(), + })); +} + +export function pruneAsks( + board: string, + olderThanMs: number, + now: number = Date.now(), +): Promise { + return pruneCollection( + board, + ASKS_COLLECTION, + ASK_FILE, + (value) => { + const ask = parseAsk(value); + if (ask.state === 'open') { + return ask.expiresAt <= now ? ask.expiresAt : null; + } + return ask.settledAt; + }, + olderThanMs, + now, + ); +} diff --git a/packages/core/src/agents/team/board-items.test.ts b/packages/core/src/agents/team/board-items.test.ts new file mode 100644 index 00000000000..c1f385882c5 --- /dev/null +++ b/packages/core/src/agents/team/board-items.test.ts @@ -0,0 +1,176 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import * as fs from 'node:fs/promises'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + answerAsk, + createAsk, + declineAsk, + listAsks, + pruneAsks, +} from './asks.js'; +import { getCollectionDir, withItemLock } from './board-lock.js'; + +vi.mock('../../config/storage.js', async (importOriginal) => { + const original = + await importOriginal(); + let globalDir = ''; + return { + ...original, + Storage: { + ...original.Storage, + getGlobalQwenDir: () => globalDir, + __setMockGlobalDir: (dir: string) => { + globalDir = dir; + }, + }, + }; +}); + +import { Storage } from '../../config/storage.js'; + +function setGlobalDir(dir: string): void { + ( + Storage as unknown as { __setMockGlobalDir: (value: string) => void } + ).__setMockGlobalDir(dir); +} + +describe('board asks', () => { + let tmpDir: string; + + beforeEach(async () => { + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'board-items-')); + setGlobalDir(tmpDir); + }); + + afterEach(async () => { + await fs.rm(tmpDir, { recursive: true, force: true }); + }); + + it('lets only the addressed actor answer or decline', async () => { + const ask = await createAsk({ + board: 'demo', + from: 'api', + to: 'web', + question: 'is status a string?', + }); + await expect(answerAsk('demo', ask.id, 'api', 'yes')).rejects.toThrow( + 'addressed to "web"', + ); + await expect( + answerAsk('demo', ask.id, 'web', 'yes'), + ).resolves.toMatchObject({ state: 'answered', answer: 'yes' }); + + const second = await createAsk({ + board: 'demo', + from: 'api', + to: 'web', + question: 'can you check?', + }); + await expect( + declineAsk('demo', second.id, 'web', 'not now'), + ).resolves.toMatchObject({ state: 'declined', reason: 'not now' }); + }); + + it('skips malformed records in list operations', async () => { + await createAsk({ + board: 'demo', + from: 'api', + to: 'web', + question: 'healthy', + }); + const asksDir = getCollectionDir('demo', 'asks'); + await fs.writeFile( + path.join(asksDir, 'a-00000000-0000-4000-8000-000000000000.json'), + JSON.stringify({ schemaVersion: 99 }), + ); + // Valid UUIDv4 filenames so these records reach content validation + // (all-zero names are rejected by the filename filter before parsing). + await fs.writeFile( + path.join(asksDir, 'a-00000000-0000-4000-8000-000000000001.json'), + '{}', + ); + await expect(listAsks('demo')).resolves.toMatchObject([ + { question: 'healthy' }, + ]); + }); + + it('rejects records whose id does not match the filename', async () => { + const firstAsk = await createAsk({ + board: 'demo', + from: 'api', + to: 'web', + question: 'first', + }); + const secondAsk = await createAsk({ + board: 'demo', + from: 'api', + to: 'web', + question: 'second', + }); + const firstAskPath = path.join( + getCollectionDir('demo', 'asks'), + `${firstAsk.id}.json`, + ); + await fs.writeFile( + firstAskPath, + JSON.stringify({ ...firstAsk, id: secondAsk.id }), + ); + await expect(listAsks('demo')).resolves.toMatchObject([ + { question: 'second' }, + ]); + await expect(answerAsk('demo', firstAsk.id, 'web', 'yes')).rejects.toThrow( + 'does not match its filename', + ); + }); + + it('re-checks prune eligibility while holding the item lock', async () => { + const ask = await createAsk({ + board: 'demo', + from: 'api', + to: 'web', + question: 'still needed?', + ttlMs: 1000, + }); + const target = path.join( + getCollectionDir('demo', 'asks'), + `${ask.id}.json`, + ); + const pruneNow = ask.expiresAt + 1; + + let pruning: Promise | undefined; + await withItemLock(target, async () => { + pruning = pruneAsks('demo', 0, pruneNow); + await new Promise((resolve) => setImmediate(resolve)); + await fs.writeFile( + target, + JSON.stringify({ + ...ask, + expiresAt: pruneNow + 1000, + }), + ); + }); + + await expect(pruning).resolves.toEqual([]); + await expect(listAsks('demo')).resolves.toMatchObject([{ state: 'open' }]); + }); + + it('reports pruned items by id, not by filename', async () => { + const ask = await createAsk({ + board: 'demo', + from: 'api', + to: 'web', + question: 'settled', + ttlMs: 1000, + }); + await declineAsk('demo', ask.id, 'web', 'not my area'); + await expect(pruneAsks('demo', 0)).resolves.toEqual([ask.id]); + await expect(listAsks('demo')).resolves.toEqual([]); + }); +}); diff --git a/packages/core/src/agents/team/board-lock.ts b/packages/core/src/agents/team/board-lock.ts new file mode 100644 index 00000000000..7d29375a542 --- /dev/null +++ b/packages/core/src/agents/team/board-lock.ts @@ -0,0 +1,235 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Two-tier locking for board item files. + * + * An in-process `Mutex` per path serializes local writers so they don't + * stampede the OS lock (the cause of Windows `ELOCKED` flakiness), wrapping a + * `proper-lockfile` cross-process lock that guards writers in other agent + * processes, over `atomicWriteJSON`. + * + * The same discipline already exists in `tasks.ts` and `mailbox.ts`. This + * implementation stays local to Agent Board so this feature does not also + * rewrite those established paths. + */ + +import { randomUUID } from 'node:crypto'; +import * as fsp from 'node:fs/promises'; +import * as path from 'node:path'; +import lockfile from 'proper-lockfile'; +import { Mutex } from 'async-mutex'; +import { Storage } from '../../config/storage.js'; +import { isNodeError } from '../../utils/errors.js'; +import { createDebugLogger } from '../../utils/debugLogger.js'; + +const debug = createDebugLogger('BOARD_LOCK'); + +/** Root for all boards: `~/.qwen/boards/`. */ +export const BOARDS_DIR = 'boards'; + +export function getBoardsRootDir(): string { + return path.join(Storage.getGlobalQwenDir(), BOARDS_DIR); +} + +/** `~/.qwen/boards/{board}/` */ +export function getBoardDir(board: string): string { + return path.join(getBoardsRootDir(), board); +} + +/** `~/.qwen/boards/{board}/{collection}/` */ +export function getCollectionDir(board: string, collection: string): string { + return path.join(getBoardDir(board), collection); +} + +/** + * Board and collection names reach the filesystem directly, so reject anything + * that could escape the root. Deliberately stricter than "no slashes": a name + * that survives this is also safe to print, to type into a command, and to use + * as a JSON key. + */ +const SAFE_NAME = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/; +const WINDOWS_DEVICE_NAME = /^(con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\.|$)/i; + +export function assertSafeName(kind: string, name: string): void { + if ( + !SAFE_NAME.test(name) || + name === '.' || + name === '..' || + name.endsWith('.') || + WINDOWS_DEVICE_NAME.test(name) + ) { + throw new Error( + `Invalid ${kind} "${name}". Use 1-64 characters: letters, digits, ` + + `dot, dash or underscore, starting with a letter or digit.`, + ); + } +} + +const ITEM_ID = + /^[at]-[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/; + +export function assertItemId( + kind: string, + id: string, + prefix?: 'a' | 't', +): void { + if (!ITEM_ID.test(id) || (prefix && !id.startsWith(`${prefix}-`))) { + throw new Error(`Invalid ${kind} "${id}".`); + } +} + +async function ensurePrivateDir(dir: string): Promise { + await fsp.mkdir(dir, { recursive: true, mode: 0o700 }); + await fsp.chmod(dir, 0o700); +} + +export async function createBoardRecord( + board: string, + collection: string, + prefix: 'a' | 't', + build: (id: string) => T, +): Promise { + assertSafeName('board name', board); + assertSafeName('collection name', collection); + const root = getBoardsRootDir(); + const boardDir = getBoardDir(board); + const collectionDir = getCollectionDir(board, collection); + await ensurePrivateDir(root); + await ensurePrivateDir(boardDir); + await ensurePrivateDir(collectionDir); + + for (let attempt = 0; attempt < 3; attempt++) { + const id = `${prefix}-${randomUUID()}`; + const record = build(id); + try { + await fsp.writeFile( + path.join(collectionDir, `${id}.json`), + JSON.stringify(record, null, 2), + { flag: 'wx', mode: 0o600, flush: true }, + ); + return record; + } catch (err) { + if (isNodeError(err) && err.code === 'EEXIST') continue; + throw err; + } + } + throw new Error(`Could not allocate a ${collection} id.`); +} + +const lockOptions: lockfile.LockOptions = { + retries: { + retries: 30, + minTimeout: 5, + maxTimeout: 100, + factor: 2, + randomize: true, + }, + stale: 5000, + onCompromised: (err) => debug.warn('board item lock compromised:', err), +}; + +const fileLocks = new Map(); + +export function withItemLock( + filePath: string, + fn: () => Promise, + onMissing?: () => T, +): Promise { + let lock = fileLocks.get(filePath); + if (!lock) { + lock = new Mutex(); + fileLocks.set(filePath, lock); + } + return lock + .runExclusive(async () => { + let release: (() => Promise) | undefined; + try { + release = await lockfile.lock(filePath, lockOptions); + } catch (err) { + if (isNodeError(err) && err.code === 'ENOENT' && onMissing) { + return onMissing(); + } + throw err; + } + try { + return await fn(); + } catch (err) { + if (isNodeError(err) && err.code === 'ENOENT' && onMissing) { + return onMissing(); + } + throw err; + } finally { + try { + await release?.(); + } catch (err) { + debug.warn('failed to release lock:', err); + } + } + }) + .finally(() => { + // runExclusive releases before this callback; keep the mutex while a + // queued caller has already acquired it. + const held = fileLocks.get(filePath); + if (held && !held.isLocked()) fileLocks.delete(filePath); + }); +} + +export async function pruneCollection( + board: string, + collection: string, + filenamePattern: RegExp, + settledAt: (record: unknown) => number | null, + olderThanMs: number, + now: number = Date.now(), +): Promise { + assertSafeName('board name', board); + assertSafeName('collection name', collection); + if (!Number.isFinite(olderThanMs) || olderThanMs < 0) { + throw new Error('olderThanMs must be a non-negative finite number.'); + } + if (!Number.isFinite(now)) throw new Error('now must be a finite number.'); + const dir = getCollectionDir(board, collection); + let files: string[]; + try { + files = await fsp.readdir(dir); + } catch (err) { + if (isNodeError(err) && err.code === 'ENOENT') return []; + throw err; + } + + const removed: string[] = []; + for (const file of files) { + if (!filenamePattern.test(file)) continue; + const full = path.join(dir, file); + await withItemLock( + full, + async () => { + const raw = await fsp.readFile(full, 'utf8'); + let timestamp: number | null; + try { + timestamp = settledAt(JSON.parse(raw)); + } catch (err) { + debug.warn(`skipping invalid ${file}:`, err); + return; + } + if ( + timestamp === null || + !Number.isFinite(timestamp) || + now - timestamp < olderThanMs + ) { + return; + } + await fsp.unlink(full); + // Report the item id, so callers can reconcile against the ids the + // rest of the board surface reports. + removed.push(path.basename(file, '.json')); + }, + () => {}, + ); + } + return removed; +} diff --git a/packages/core/src/agents/team/board-tasks.test.ts b/packages/core/src/agents/team/board-tasks.test.ts new file mode 100644 index 00000000000..b2b24fbed9d --- /dev/null +++ b/packages/core/src/agents/team/board-tasks.test.ts @@ -0,0 +1,174 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import * as fs from 'node:fs/promises'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + claimBoardTask, + completeBoardTask, + createBoardTask, + listBoardTasks, +} from './board-tasks.js'; +import { getCollectionDir } from './board-lock.js'; + +vi.mock('../../config/storage.js', async (importOriginal) => { + const original = + await importOriginal(); + let globalDir = ''; + return { + ...original, + Storage: { + ...original.Storage, + getGlobalQwenDir: () => globalDir, + __setMockGlobalDir: (dir: string) => { + globalDir = dir; + }, + }, + }; +}); + +import { Storage } from '../../config/storage.js'; + +function setGlobalDir(dir: string): void { + ( + Storage as unknown as { __setMockGlobalDir: (value: string) => void } + ).__setMockGlobalDir(dir); +} + +describe('board tasks', () => { + let tmpDir: string; + + beforeEach(async () => { + tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), 'board-tasks-')); + setGlobalDir(tmpDir); + }); + + afterEach(async () => { + await fs.rm(tmpDir, { recursive: true, force: true }); + }); + + it('creates collision-resistant ids concurrently', async () => { + const tasks = await Promise.all( + Array.from({ length: 40 }, (_, index) => + createBoardTask({ + board: 'demo', + createdBy: 'author', + subject: `task ${index}`, + }), + ), + ); + expect(new Set(tasks.map((task) => task.id)).size).toBe(40); + expect(tasks.every((task) => /^t-[0-9a-f-]{36}$/.test(task.id))).toBe(true); + }); + + it('requires ownership before completion', async () => { + const task = await createBoardTask({ + board: 'demo', + createdBy: 'author', + subject: 'check api', + }); + await expect(completeBoardTask('demo', task.id, 'other')).rejects.toThrow( + 'not in progress', + ); + await claimBoardTask('demo', task.id, 'worker'); + const completed = await completeBoardTask( + 'demo', + task.id, + 'worker', + 'done', + ); + expect(completed.status).toBe('completed'); + expect(completed.notes).toEqual(['done']); + }); + + it('skips a malformed foreign record without hiding healthy work', async () => { + await createBoardTask({ + board: 'demo', + createdBy: 'author', + subject: 'healthy', + }); + const malformedId = 't-00000000-0000-4000-8000-000000000000'; + const malformedPath = path.join( + getCollectionDir('demo', 'tasks'), + `${malformedId}.json`, + ); + await fs.writeFile(malformedPath, '{broken'); + // A valid UUIDv4 filename so this record reaches content validation + // (all-zero names are rejected by the filename filter before parsing). + await fs.writeFile( + path.join( + getCollectionDir('demo', 'tasks'), + 't-00000000-0000-4000-8000-000000000001.json', + ), + '{}', + ); + await expect(listBoardTasks('demo')).resolves.toMatchObject([ + { subject: 'healthy' }, + ]); + await expect(claimBoardTask('demo', malformedId, 'worker')).rejects.toThrow( + 'JSON', + ); + await expect(fs.readFile(malformedPath, 'utf8')).resolves.toBe('{broken'); + }); + + it('rejects empty owners and records whose id does not match the filename', async () => { + await expect( + createBoardTask({ + board: 'demo', + createdBy: 'author', + subject: 'empty owner', + owner: '', + }), + ).rejects.toThrow('Invalid actor name'); + + const first = await createBoardTask({ + board: 'demo', + createdBy: 'author', + subject: 'first', + }); + const second = await createBoardTask({ + board: 'demo', + createdBy: 'author', + subject: 'second', + }); + const firstPath = path.join( + getCollectionDir('demo', 'tasks'), + `${first.id}.json`, + ); + await fs.writeFile(firstPath, JSON.stringify({ ...first, id: second.id })); + await expect(listBoardTasks('demo')).resolves.toMatchObject([ + { subject: 'second' }, + ]); + await expect(claimBoardTask('demo', first.id, 'worker')).rejects.toThrow( + 'does not match its filename', + ); + await expect(fs.readFile(firstPath, 'utf8')).resolves.toContain(second.id); + }); + + it('creates private directories and files', async () => { + const task = await createBoardTask({ + board: 'demo', + createdBy: 'author', + subject: 'private', + }); + if (process.platform === 'win32') return; + const dir = getCollectionDir('demo', 'tasks'); + expect((await fs.stat(dir)).mode & 0o777).toBe(0o700); + expect( + (await fs.stat(path.join(dir, `${task.id}.json`))).mode & 0o777, + ).toBe(0o600); + }); + + it('rejects unsafe board directory names', async () => { + for (const board of ['../escape', 'con', 'trailing.']) { + await expect( + createBoardTask({ board, createdBy: 'author', subject: 'unsafe' }), + ).rejects.toThrow('Invalid board name'); + } + }); +}); diff --git a/packages/core/src/agents/team/board-tasks.ts b/packages/core/src/agents/team/board-tasks.ts new file mode 100644 index 00000000000..8cebdf4fb45 --- /dev/null +++ b/packages/core/src/agents/team/board-tasks.ts @@ -0,0 +1,241 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import * as fs from 'node:fs/promises'; +import * as path from 'node:path'; +import { atomicWriteJSON } from '../../utils/atomicFileWrite.js'; +import { createDebugLogger } from '../../utils/debugLogger.js'; +import { isNodeError } from '../../utils/errors.js'; +import { + assertItemId, + assertSafeName, + createBoardRecord, + getCollectionDir, + pruneCollection, + withItemLock, +} from './board-lock.js'; + +const debug = createDebugLogger('BOARD_TASKS'); + +export const TASKS_COLLECTION = 'tasks'; +const MAX_TEXT_LENGTH = 65536; +const TASK_FILE = + /^t-[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\.json$/; + +export type BoardTaskStatus = 'pending' | 'in_progress' | 'completed'; + +export interface BoardTaskRecord { + schemaVersion: 1; + id: string; + subject: string; + createdBy: string; + owner: string | null; + status: BoardTaskStatus; + createdAt: number; + updatedAt: number; + notes: string[]; +} + +function tasksDir(board: string): string { + return getCollectionDir(board, TASKS_COLLECTION); +} + +function taskPath(board: string, id: string): string { + return path.join(tasksDir(board), `${id}.json`); +} + +function assertText(field: string, value: string): void { + if (!value.trim()) throw new Error(`${field} must not be empty.`); + if (value.length > MAX_TEXT_LENGTH) { + throw new Error(`${field} exceeds ${MAX_TEXT_LENGTH} characters.`); + } +} + +function parseTask(value: unknown): BoardTaskRecord { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error('Task record must be an object.'); + } + const task = value as Partial; + if (task.schemaVersion !== 1) throw new Error('Unsupported task schema.'); + if (typeof task.id !== 'string') throw new Error('Task id must be text.'); + assertItemId('task id', task.id, 't'); + if (typeof task.subject !== 'string') { + throw new Error('Task subject must be text.'); + } + assertText('subject', task.subject); + if (typeof task.createdBy !== 'string') { + throw new Error('Task createdBy must be a name.'); + } + assertSafeName('actor name', task.createdBy); + if (task.owner !== null && typeof task.owner !== 'string') { + throw new Error('Task owner must be a name or null.'); + } + if (task.owner !== null) assertSafeName('actor name', task.owner); + if (!['pending', 'in_progress', 'completed'].includes(task.status ?? '')) { + throw new Error('Invalid task status.'); + } + if (!Number.isFinite(task.createdAt) || !Number.isFinite(task.updatedAt)) { + throw new Error('Task timestamps must be finite numbers.'); + } + if ((task.updatedAt ?? 0) < (task.createdAt ?? 0)) { + throw new Error('Task updatedAt precedes createdAt.'); + } + if ( + !Array.isArray(task.notes) || + !task.notes.every((n) => typeof n === 'string') + ) { + throw new Error('Task notes must be strings.'); + } + for (const note of task.notes) assertText('note', note); + return task as BoardTaskRecord; +} + +export async function createBoardTask(opts: { + board: string; + createdBy: string; + subject: string; + owner?: string; +}): Promise { + assertSafeName('actor name', opts.createdBy); + assertText('subject', opts.subject); + if (opts.owner !== undefined) assertSafeName('actor name', opts.owner); + const now = Date.now(); + return createBoardRecord(opts.board, TASKS_COLLECTION, 't', (id) => ({ + schemaVersion: 1, + id, + subject: opts.subject, + createdBy: opts.createdBy, + owner: opts.owner ?? null, + status: 'pending', + createdAt: now, + updatedAt: now, + notes: [], + })); +} + +async function getBoardTask( + board: string, + id: string, +): Promise { + assertSafeName('board name', board); + assertItemId('task id', id, 't'); + let raw: string; + try { + raw = await fs.readFile(taskPath(board, id), 'utf8'); + } catch (err) { + if (isNodeError(err) && err.code === 'ENOENT') return null; + throw err; + } + try { + const task = parseTask(JSON.parse(raw)); + if (task.id !== id) throw new Error('Task id does not match its filename.'); + return task; + } catch (err) { + debug.warn(`skipping invalid task ${id}:`, err); + return null; + } +} + +export async function listBoardTasks( + board: string, +): Promise { + assertSafeName('board name', board); + let files: string[]; + try { + files = await fs.readdir(tasksDir(board)); + } catch (err) { + if (isNodeError(err) && err.code === 'ENOENT') return []; + throw err; + } + const tasks = ( + await Promise.all( + files + .filter((file) => TASK_FILE.test(file)) + .map((file) => getBoardTask(board, file.slice(0, -5))), + ) + ).filter((task): task is BoardTaskRecord => task !== null); + return tasks.sort((a, b) => a.createdAt - b.createdAt); +} + +async function mutate( + board: string, + id: string, + apply: (task: BoardTaskRecord) => BoardTaskRecord, +): Promise { + assertSafeName('board name', board); + assertItemId('task id', id, 't'); + const target = taskPath(board, id); + return withItemLock( + target, + async () => { + const current = parseTask(JSON.parse(await fs.readFile(target, 'utf8'))); + if (current.id !== id) { + throw new Error('Task id does not match its filename.'); + } + const next = parseTask({ ...apply(current), updatedAt: Date.now() }); + await atomicWriteJSON(target, next, { mode: 0o600, forceMode: true }); + return next; + }, + () => { + throw new Error(`Task "${id}" not found.`); + }, + ); +} + +export function claimBoardTask( + board: string, + id: string, + by: string, +): Promise { + assertSafeName('actor name', by); + return mutate(board, id, (task) => { + if (task.status === 'completed') { + throw new Error(`Task "${id}" is already completed.`); + } + if (task.status === 'in_progress' && task.owner !== by) { + throw new Error(`Task "${id}" is already claimed by "${task.owner}".`); + } + return { ...task, owner: by, status: 'in_progress' }; + }); +} + +export function completeBoardTask( + board: string, + id: string, + by: string, + note?: string, +): Promise { + assertSafeName('actor name', by); + if (note !== undefined) assertText('note', note); + return mutate(board, id, (task) => { + if (task.owner !== by || task.status !== 'in_progress') { + throw new Error(`Task "${id}" is not in progress for "${by}".`); + } + return { + ...task, + status: 'completed', + ...(note ? { notes: [...task.notes, note] } : {}), + }; + }); +} + +export function pruneBoardTasks( + board: string, + olderThanMs: number, + now?: number, +): Promise { + return pruneCollection( + board, + TASKS_COLLECTION, + TASK_FILE, + (value) => { + const task = parseTask(value); + return task.status === 'completed' ? task.updatedAt : null; + }, + olderThanMs, + now, + ); +} diff --git a/packages/core/src/board.ts b/packages/core/src/board.ts new file mode 100644 index 00000000000..74f1a3dbc58 --- /dev/null +++ b/packages/core/src/board.ts @@ -0,0 +1,27 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +// A dedicated subpath keeps the board dependency chain out of ACP startup. +export { assertSafeName } from './agents/team/board-lock.js'; +export { + claimBoardTask, + completeBoardTask, + createBoardTask, + listBoardTasks, + pruneBoardTasks, + type BoardTaskRecord, + type BoardTaskStatus, +} from './agents/team/board-tasks.js'; +export { + answerAsk, + createAsk, + declineAsk, + getAsk, + listAsks, + pruneAsks, + type AskRecord, + type AskState, +} from './agents/team/asks.js';