diff --git a/package.json b/package.json index c28bc02..5aecdf7 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "agenthub", - "version": "0.4.0", + "version": "0.5.0", "description": "Local coordination layer for AI coding agents", "type": "module", "main": "./dist/index.js", diff --git a/src/cli/commands/watch.ts b/src/cli/commands/watch.ts index 5ed1433..9d8b164 100644 --- a/src/cli/commands/watch.ts +++ b/src/cli/commands/watch.ts @@ -108,17 +108,41 @@ export function formatEvent(event: AgentHubEvent): string { return detail ? `${head} ${detail}` : head; } +/** Fetch any tasks currently in `review` (an implementer is awaiting a verdict). */ +async function fetchReviewTasks(serverUrl: string): Promise { + try { + const res = await fetch(new URL('/tasks', serverUrl).toString()); + if (!res.ok) return []; + const tasks = (await res.json()) as Array<{ + id: string; + title?: string; + status?: string; + role?: string; + assignedTo?: string; + }>; + return tasks + .filter((t) => t.status === 'review') + .map((t) => ({ type: 'task', action: 'updated', id: t.id, title: t.title, status: 'review', role: t.role, assignedTo: t.assignedTo })); + } catch { + return []; + } +} + /** * Connect to the AgentHub server's SSE endpoint and stream events to stdout. * * Exits the process when: * - `--once` is set and the first event arrives (exit 0). + * - `--await-review` is set and a task enters `review` (an implementer + * submitted) — including any task already in review on connect. Lets the + * architect run it in the background and be notified the moment a + * submission needs a verdict. * - The server closes the stream (normal exit). * - A connection error occurs (exit 1). */ export async function watchEvents( serverUrl: string, - options: { once?: boolean; role?: string } = {}, + options: { once?: boolean; role?: string; awaitReview?: boolean } = {}, ): Promise { const url = new URL('/events', serverUrl); // Pass role to the server for an additional server-side filter (saves @@ -149,6 +173,17 @@ export async function watchEvents( const decoder = new TextDecoder(); let buffer = ''; + // --await-review: surface a submission that's ALREADY pending on connect, + // so the architect isn't blind to reviews submitted before this watcher. + if (options.awaitReview) { + const pending = await fetchReviewTasks(serverUrl); + if (pending.length > 0) { + for (const ev of pending) console.log(formatEvent(ev)); + await reader.cancel(); + return; + } + } + while (true) { let done: boolean; let value: Uint8Array | undefined; @@ -179,6 +214,13 @@ export async function watchEvents( await reader.cancel(); return; } + + // --await-review: exit the moment an implementer submits (task → review), + // so the architect (running this in the background) gets notified. + if (options.awaitReview && event.type === 'task' && event.status === 'review') { + await reader.cancel(); + return; + } } } } diff --git a/src/cli/index.ts b/src/cli/index.ts index 94c4b48..4dcba3f 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -114,7 +114,7 @@ async function runRemote(serverUrl: string, fn: () => Promise): Promise', 'AgentHub server URL (env: AGENTHUB_SERVER)'); program @@ -525,6 +525,7 @@ export function createProgram(cwd: string): Command { .description('Stream live AgentHub events from a running server (SSE)') .option('--once', 'Exit 0 after the first event (useful as a blocking wait for agents)') .option('--role ', 'Client-side role filter (only show events for this role)') + .option('--await-review', 'Exit when an implementer submits (task → review); architect review-queue notifier') .action(async (options) => { const { serverUrl } = await resolveContext(program, cwd); if (!serverUrl) { @@ -532,7 +533,11 @@ export function createProgram(cwd: string): Command { process.exit(1); return; } - await watchEvents(serverUrl, { once: options.once as boolean | undefined, role: options.role as string | undefined }); + await watchEvents(serverUrl, { + once: options.once as boolean | undefined, + role: options.role as string | undefined, + awaitReview: options.awaitReview as boolean | undefined, + }); }); // ─── server ────────────────────────────────────────────────────────────── diff --git a/tests/sse.test.ts b/tests/sse.test.ts index 5235978..35b2fd4 100644 --- a/tests/sse.test.ts +++ b/tests/sse.test.ts @@ -15,7 +15,7 @@ import { buildApp } from '../src/server/index.js'; import { startServer } from '../src/server/index.js'; import { eventBus } from '../src/server/events.js'; import type { AgentHubEvent } from '../src/server/events.js'; -import { parseSSEBuffer, formatEvent } from '../src/cli/commands/watch.js'; +import { parseSSEBuffer, formatEvent, watchEvents } from '../src/cli/commands/watch.js'; import { init } from '../src/cli/commands/init.js'; import { startEntityWatcher } from '../src/server/fsWatch.js'; import { createTask, claimTask } from '../src/core/services/taskService.js'; @@ -563,3 +563,59 @@ describe('status auto-refresh on mutation', () => { expect(status).toContain('TSK-0001'); // the just-created task shows as active }, 4000); }); + +// ─── 8. watch --await-review: architect review-queue notifier ───────────────── + +describe('watch --await-review', () => { + let cwd: string; + let server: Awaited>; + + beforeEach(async () => { + cwd = mkdtempSync(join(tmpdir(), 'ah-await-review-')); + init(cwd, { projectName: 'await-review', yes: true }); + server = await startServer(cwd, { host: '127.0.0.1', port: 0 }); + }); + + afterEach(async () => { + await server.app.close(); + rmSync(cwd, { recursive: true, force: true }); + }); + + async function submitForReview(): Promise { + await fetch(`${server.url}/tasks`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ title: 'kimi: t', role: 'implementer' }), + }); + await fetch(`${server.url}/tasks/TSK-0001`, { + method: 'PATCH', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ status: 'in_progress', assignedTo: 'kimi' }), + }); + } + + it('returns immediately when a task is already in review (initial check)', async () => { + await submitForReview(); + await fetch(`${server.url}/tasks/TSK-0001`, { + method: 'PATCH', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ status: 'review' }), + }); + + await expect(watchEvents(server.url, { awaitReview: true })).resolves.toBeUndefined(); + }, 4000); + + it('exits when a task transitions to review while watching', async () => { + await submitForReview(); + + const watching = watchEvents(server.url, { awaitReview: true }); + await new Promise((r) => setTimeout(r, 200)); // let it connect + pass the initial check + await fetch(`${server.url}/tasks/TSK-0001`, { + method: 'PATCH', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ status: 'review' }), + }); + + await expect(watching).resolves.toBeUndefined(); + }, 5000); +});