feat(watch): --await-review — architect review-queue notifier

Closes the architect-side gap: the auto-claim daemon wakes implementers on
new tasks, but the architect had no signal when an implementer submitted.

`agenthub watch --await-review` exits (after printing) the moment a task
enters `review` — including any task already in review on connect (initial
check) — so the architect can run it in the background and be re-invoked by
the harness the instant a submission needs a verdict.

- watch.ts: awaitReview option, initial fetchReviewTasks() check + SSE exit
  on task→review.
- index.ts: --await-review flag.
- tests: returns immediately on already-pending review; exits on a review
  transition while watching. 121/121 green.

Bump 0.4.0 -> 0.5.0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
chahinebrini 2026-06-27 18:53:40 +02:00
parent d2b26cc64c
commit ec378131e4
4 changed files with 108 additions and 5 deletions

View File

@ -1,6 +1,6 @@
{ {
"name": "agenthub", "name": "agenthub",
"version": "0.4.0", "version": "0.5.0",
"description": "Local coordination layer for AI coding agents", "description": "Local coordination layer for AI coding agents",
"type": "module", "type": "module",
"main": "./dist/index.js", "main": "./dist/index.js",

View File

@ -108,17 +108,41 @@ export function formatEvent(event: AgentHubEvent): string {
return detail ? `${head} ${detail}` : head; return detail ? `${head} ${detail}` : head;
} }
/** Fetch any tasks currently in `review` (an implementer is awaiting a verdict). */
async function fetchReviewTasks(serverUrl: string): Promise<AgentHubEvent[]> {
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. * Connect to the AgentHub server's SSE endpoint and stream events to stdout.
* *
* Exits the process when: * Exits the process when:
* - `--once` is set and the first event arrives (exit 0). * - `--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). * - The server closes the stream (normal exit).
* - A connection error occurs (exit 1). * - A connection error occurs (exit 1).
*/ */
export async function watchEvents( export async function watchEvents(
serverUrl: string, serverUrl: string,
options: { once?: boolean; role?: string } = {}, options: { once?: boolean; role?: string; awaitReview?: boolean } = {},
): Promise<void> { ): Promise<void> {
const url = new URL('/events', serverUrl); const url = new URL('/events', serverUrl);
// Pass role to the server for an additional server-side filter (saves // Pass role to the server for an additional server-side filter (saves
@ -149,6 +173,17 @@ export async function watchEvents(
const decoder = new TextDecoder(); const decoder = new TextDecoder();
let buffer = ''; 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) { while (true) {
let done: boolean; let done: boolean;
let value: Uint8Array | undefined; let value: Uint8Array | undefined;
@ -179,6 +214,13 @@ export async function watchEvents(
await reader.cancel(); await reader.cancel();
return; 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;
}
} }
} }
} }

View File

@ -114,7 +114,7 @@ async function runRemote(serverUrl: string, fn: () => Promise<void>): Promise<vo
export function createProgram(cwd: string): Command { export function createProgram(cwd: string): Command {
const program = new Command('agenthub') const program = new Command('agenthub')
.description('Local coordination layer for AI coding agents') .description('Local coordination layer for AI coding agents')
.version('0.4.0') .version('0.5.0')
.option('--server <url>', 'AgentHub server URL (env: AGENTHUB_SERVER)'); .option('--server <url>', 'AgentHub server URL (env: AGENTHUB_SERVER)');
program program
@ -525,6 +525,7 @@ export function createProgram(cwd: string): Command {
.description('Stream live AgentHub events from a running server (SSE)') .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('--once', 'Exit 0 after the first event (useful as a blocking wait for agents)')
.option('--role <role>', 'Client-side role filter (only show events for this role)') .option('--role <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) => { .action(async (options) => {
const { serverUrl } = await resolveContext(program, cwd); const { serverUrl } = await resolveContext(program, cwd);
if (!serverUrl) { if (!serverUrl) {
@ -532,7 +533,11 @@ export function createProgram(cwd: string): Command {
process.exit(1); process.exit(1);
return; 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 ────────────────────────────────────────────────────────────── // ─── server ──────────────────────────────────────────────────────────────

View File

@ -15,7 +15,7 @@ import { buildApp } from '../src/server/index.js';
import { startServer } from '../src/server/index.js'; import { startServer } from '../src/server/index.js';
import { eventBus } from '../src/server/events.js'; import { eventBus } from '../src/server/events.js';
import type { AgentHubEvent } 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 { init } from '../src/cli/commands/init.js';
import { startEntityWatcher } from '../src/server/fsWatch.js'; import { startEntityWatcher } from '../src/server/fsWatch.js';
import { createTask, claimTask } from '../src/core/services/taskService.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 expect(status).toContain('TSK-0001'); // the just-created task shows as active
}, 4000); }, 4000);
}); });
// ─── 8. watch --await-review: architect review-queue notifier ─────────────────
describe('watch --await-review', () => {
let cwd: string;
let server: Awaited<ReturnType<typeof startServer>>;
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<void> {
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);
});