640 lines
25 KiB
TypeScript
640 lines
25 KiB
TypeScript
/**
|
|
* Tests for the SSE realtime channel (TSK-0006 part 1).
|
|
*
|
|
* Three concerns:
|
|
* 1. Unit: parseSSEBuffer + formatEvent (pure functions, no I/O).
|
|
* 2. Integration: eventBus emits on REST mutations (via app.inject).
|
|
* 3. E2E: a real server delivers a task/created event over the SSE stream.
|
|
*/
|
|
|
|
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
|
|
import { mkdtempSync, rmSync, readFileSync } from 'fs';
|
|
import { tmpdir } from 'os';
|
|
import { join } from 'path';
|
|
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, 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';
|
|
|
|
// ─── 1. Unit: SSE buffer parser ──────────────────────────────────────────────
|
|
|
|
describe('parseSSEBuffer', () => {
|
|
it('parses a single complete event', () => {
|
|
const buf = 'data: {"type":"task","action":"created","id":"TSK-0001"}\n\n';
|
|
const { events, remaining } = parseSSEBuffer(buf);
|
|
expect(events).toHaveLength(1);
|
|
expect(events[0].type).toBe('task');
|
|
expect(events[0].action).toBe('created');
|
|
expect(events[0].id).toBe('TSK-0001');
|
|
expect(remaining).toBe('');
|
|
});
|
|
|
|
it('parses multiple complete events in one buffer', () => {
|
|
const buf =
|
|
'data: {"type":"task","action":"created","id":"TSK-0001"}\n\n' +
|
|
'data: {"type":"memory","action":"created","id":"MEM-001"}\n\n';
|
|
const { events, remaining } = parseSSEBuffer(buf);
|
|
expect(events).toHaveLength(2);
|
|
expect(events[0].type).toBe('task');
|
|
expect(events[1].type).toBe('memory');
|
|
expect(remaining).toBe('');
|
|
});
|
|
|
|
it('ignores SSE keepalive comment blocks', () => {
|
|
const buf =
|
|
':\n\n' +
|
|
'data: {"type":"handoff","action":"created","id":"HOF-0001"}\n\n';
|
|
const { events } = parseSSEBuffer(buf);
|
|
expect(events).toHaveLength(1);
|
|
expect(events[0].id).toBe('HOF-0001');
|
|
});
|
|
|
|
it('keeps an incomplete tail in `remaining`', () => {
|
|
const buf =
|
|
'data: {"type":"task","action":"created","id":"TSK-0001"}\n\n' +
|
|
'data: {"type":"deci'; // truncated chunk
|
|
const { events, remaining } = parseSSEBuffer(buf);
|
|
expect(events).toHaveLength(1);
|
|
expect(remaining).toBe('data: {"type":"deci');
|
|
});
|
|
|
|
it('returns empty events and the full buffer when there is no complete event', () => {
|
|
const buf = 'data: {"type":"task"'; // no \n\n yet
|
|
const { events, remaining } = parseSSEBuffer(buf);
|
|
expect(events).toHaveLength(0);
|
|
expect(remaining).toBe('data: {"type":"task"');
|
|
});
|
|
|
|
it('silently skips malformed JSON without throwing', () => {
|
|
const buf = 'data: NOT_JSON\n\n';
|
|
const { events } = parseSSEBuffer(buf);
|
|
expect(events).toHaveLength(0);
|
|
});
|
|
});
|
|
|
|
// ─── 2. Unit: formatEvent ────────────────────────────────────────────────────
|
|
|
|
describe('formatEvent (AgentHub-branded)', () => {
|
|
it('every line starts with the AgentHub: prefix', () => {
|
|
const ev: AgentHubEvent = { type: 'decision', action: 'created', id: 'DEC-0001' };
|
|
expect(formatEvent(ev).startsWith('AgentHub: ')).toBe(true);
|
|
});
|
|
|
|
it('formats a created task as "Task received <id> <title> [status]"', () => {
|
|
const ev: AgentHubEvent = { type: 'task', action: 'created', id: 'TSK-0019', title: 'Magic-App Audit', status: 'open' };
|
|
expect(formatEvent(ev)).toMatch(/^AgentHub: Task received\s+TSK-0019\s+Magic-App Audit\s+\[open\]$/);
|
|
});
|
|
|
|
it('formats a created task with no title as "Task received <id> [status]"', () => {
|
|
const ev: AgentHubEvent = { type: 'task', action: 'created', id: 'TSK-0001', status: 'open' };
|
|
expect(formatEvent(ev)).toMatch(/^AgentHub: Task received\s+TSK-0001\s+\[open\]$/);
|
|
});
|
|
|
|
it('formats a completed task as "Task done <id> by <agent>"', () => {
|
|
const ev: AgentHubEvent = {
|
|
type: 'task',
|
|
action: 'updated',
|
|
id: 'TSK-0003',
|
|
title: 'Implement auth',
|
|
status: 'done',
|
|
role: 'implementer',
|
|
assignedTo: 'windows-claude',
|
|
};
|
|
expect(formatEvent(ev)).toMatch(/^AgentHub: Task done\s+TSK-0003\s+by windows-claude$/);
|
|
});
|
|
|
|
it('formats a claimed task as "Task claimed <id> by <agent>"', () => {
|
|
const ev: AgentHubEvent = { type: 'task', action: 'updated', id: 'TSK-0014', status: 'in_progress', assignedTo: 'codex' };
|
|
expect(formatEvent(ev)).toMatch(/^AgentHub: Task claimed\s+TSK-0014\s+by codex$/);
|
|
});
|
|
|
|
it('formats a handoff as "Handoff <id> → <role>"', () => {
|
|
const ev: AgentHubEvent = { type: 'handoff', action: 'created', id: 'HOF-0012', title: 'For codex', role: 'implementer' };
|
|
expect(formatEvent(ev)).toMatch(/^AgentHub: Handoff\s+HOF-0012\s+.*→ implementer$/);
|
|
});
|
|
|
|
it('formats a decision with no detail as just the prefix + id', () => {
|
|
const ev: AgentHubEvent = { type: 'decision', action: 'created', id: 'DEC-0001' };
|
|
expect(formatEvent(ev)).toMatch(/^AgentHub: Decision\s+DEC-0001$/);
|
|
});
|
|
|
|
it('formats an agent presence event as "<agent> joined (<role>)"', () => {
|
|
const ev: AgentHubEvent = { type: 'agent', action: 'joined', id: 'codex', role: 'implementer' };
|
|
expect(formatEvent(ev)).toBe('AgentHub: codex joined (implementer)');
|
|
});
|
|
|
|
it('formats an agent join with no role', () => {
|
|
const ev: AgentHubEvent = { type: 'agent', action: 'joined', id: 'kimi' };
|
|
expect(formatEvent(ev)).toBe('AgentHub: kimi joined');
|
|
});
|
|
|
|
it('formats a review submission as "Task review <id> by <agent>"', () => {
|
|
const ev: AgentHubEvent = { type: 'task', action: 'updated', id: 'TSK-0019', status: 'review', assignedTo: 'codex' };
|
|
expect(formatEvent(ev)).toMatch(/^AgentHub: Task review\s+TSK-0019\s+by codex$/);
|
|
});
|
|
|
|
it('formats a reopened task as "Task reopened <id>"', () => {
|
|
const ev: AgentHubEvent = { type: 'task', action: 'updated', id: 'TSK-0019', status: 'open' };
|
|
expect(formatEvent(ev)).toMatch(/^AgentHub: Task reopened\s+TSK-0019$/);
|
|
});
|
|
});
|
|
|
|
// ─── 3. Integration: eventBus fires on REST mutations ────────────────────────
|
|
|
|
describe('eventBus mutations', () => {
|
|
let cwd: string;
|
|
let app: ReturnType<typeof buildApp>;
|
|
const collected: AgentHubEvent[] = [];
|
|
|
|
beforeEach(() => {
|
|
cwd = mkdtempSync(join(tmpdir(), 'ah-sse-bus-'));
|
|
init(cwd, { projectName: 'sse-bus-test', yes: true });
|
|
app = buildApp(cwd);
|
|
eventBus.on('change', (e: AgentHubEvent) => collected.push(e));
|
|
});
|
|
|
|
afterEach(async () => {
|
|
eventBus.removeAllListeners('change');
|
|
collected.length = 0;
|
|
await app.close();
|
|
rmSync(cwd, { recursive: true, force: true });
|
|
});
|
|
|
|
it('emits task/created when POST /tasks succeeds', async () => {
|
|
await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'Bus task', role: 'implementer' } });
|
|
expect(collected).toHaveLength(1);
|
|
expect(collected[0]).toMatchObject({ type: 'task', action: 'created', id: 'TSK-0001', title: 'Bus task', role: 'implementer' });
|
|
});
|
|
|
|
it('emits task/updated when PATCH /tasks/:id changes status', async () => {
|
|
await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'T', role: 'implementer' } });
|
|
collected.length = 0; // clear the created event
|
|
|
|
await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'done' } });
|
|
expect(collected).toHaveLength(1);
|
|
expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', id: 'TSK-0001', status: 'done' });
|
|
});
|
|
|
|
it('emits task/updated with assignedTo when a task is claimed', async () => {
|
|
await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'T', role: 'implementer' } });
|
|
collected.length = 0;
|
|
|
|
await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'in_progress', assignedTo: 'mac-claude' } });
|
|
expect(collected).toHaveLength(1);
|
|
expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', status: 'in_progress', assignedTo: 'mac-claude' });
|
|
});
|
|
|
|
it('emits task/updated review when an implementer submits for review', async () => {
|
|
await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'T', role: 'implementer' } });
|
|
collected.length = 0;
|
|
await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'review' } });
|
|
expect(collected).toHaveLength(1);
|
|
expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', status: 'review' });
|
|
});
|
|
|
|
it('emits task/updated open when the architect reopens after review', async () => {
|
|
await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'T', role: 'implementer' } });
|
|
await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'review' } });
|
|
collected.length = 0;
|
|
await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'open' } });
|
|
expect(collected).toHaveLength(1);
|
|
expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', status: 'open' });
|
|
});
|
|
|
|
it('emits handoff/created when POST /handoffs succeeds', async () => {
|
|
await app.inject({
|
|
method: 'POST',
|
|
url: '/handoffs',
|
|
payload: { fromRole: 'architect', toRole: 'implementer', summary: 'Design complete', context: 'See decisions' },
|
|
});
|
|
expect(collected).toHaveLength(1);
|
|
expect(collected[0]).toMatchObject({ type: 'handoff', action: 'created' });
|
|
});
|
|
|
|
it('emits decision/created when POST /decisions succeeds', async () => {
|
|
await app.inject({
|
|
method: 'POST',
|
|
url: '/decisions',
|
|
payload: { title: 'Use SQLite', decision: 'SQLite for local storage', context: 'Low overhead' },
|
|
});
|
|
expect(collected).toHaveLength(1);
|
|
expect(collected[0]).toMatchObject({ type: 'decision', action: 'created', title: 'Use SQLite' });
|
|
});
|
|
|
|
it('emits memory/created when POST /memory succeeds', async () => {
|
|
await app.inject({
|
|
method: 'POST',
|
|
url: '/memory',
|
|
payload: { title: 'DNS cache fact', category: 'technical', content: 'TTL matters' },
|
|
});
|
|
expect(collected).toHaveLength(1);
|
|
expect(collected[0]).toMatchObject({ type: 'memory', action: 'created', title: 'DNS cache fact' });
|
|
});
|
|
|
|
it('does NOT emit when task creation fails (400)', async () => {
|
|
// title: '' fails TaskSchema min(1) — createTask throws, route returns 400.
|
|
const res = await app.inject({ method: 'POST', url: '/tasks', payload: { title: '' } });
|
|
expect(res.statusCode).toBe(400);
|
|
// No event should be emitted on failure.
|
|
expect(collected).toHaveLength(0);
|
|
});
|
|
});
|
|
|
|
// ─── 4. E2E: SSE stream delivers events over a real HTTP connection ───────────
|
|
|
|
describe('SSE stream e2e', () => {
|
|
let cwd: string;
|
|
let server: Awaited<ReturnType<typeof startServer>>;
|
|
|
|
beforeEach(async () => {
|
|
cwd = mkdtempSync(join(tmpdir(), 'ah-sse-e2e-'));
|
|
init(cwd, { projectName: 'sse-e2e-test', 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 });
|
|
});
|
|
|
|
it('sends an immediate keepalive comment when /events connects', async () => {
|
|
const controller = new AbortController();
|
|
const res = await fetch(`${server.url}/events`, { signal: controller.signal });
|
|
expect(res.status).toBe(200);
|
|
expect(res.headers.get('content-type')).toContain('text/event-stream');
|
|
expect(res.body).toBeTruthy();
|
|
|
|
const reader = res.body!.getReader();
|
|
const timeout = new Promise<never>((_, reject) => {
|
|
setTimeout(() => reject(new Error('SSE: no keepalive frame within 1 s')), 1000);
|
|
});
|
|
const chunk = await Promise.race([reader.read(), timeout]);
|
|
controller.abort();
|
|
|
|
expect(chunk.done).toBe(false);
|
|
expect(new TextDecoder().decode(chunk.value)).toContain(': connected\n\n');
|
|
}, 3000);
|
|
|
|
it('delivers task/created to a connected SSE subscriber', async () => {
|
|
const controller = new AbortController();
|
|
|
|
// Start listening BEFORE posting the task.
|
|
const firstEventPromise = new Promise<AgentHubEvent>((resolve, reject) => {
|
|
const timeout = setTimeout(
|
|
() => reject(new Error('SSE: no event received within 3 s')),
|
|
3000,
|
|
);
|
|
|
|
fetch(`${server.url}/events`, { signal: controller.signal })
|
|
.then(async (res) => {
|
|
if (!res.body) {
|
|
clearTimeout(timeout);
|
|
reject(new Error('SSE response has no body'));
|
|
return;
|
|
}
|
|
const reader = res.body.getReader();
|
|
const dec = new TextDecoder();
|
|
let buf = '';
|
|
|
|
while (true) {
|
|
let done: boolean;
|
|
let value: Uint8Array | undefined;
|
|
try {
|
|
({ done, value } = await reader.read());
|
|
} catch {
|
|
break; // AbortError when controller.abort() is called
|
|
}
|
|
if (done) break;
|
|
if (value) buf += dec.decode(value, { stream: true });
|
|
|
|
const { events, remaining } = parseSSEBuffer(buf);
|
|
buf = remaining;
|
|
|
|
for (const ev of events) {
|
|
clearTimeout(timeout);
|
|
resolve(ev);
|
|
return;
|
|
}
|
|
}
|
|
})
|
|
.catch((err: unknown) => {
|
|
// AbortError is expected when we abort after receiving the event.
|
|
if (err instanceof Error && err.name === 'AbortError') return;
|
|
reject(err);
|
|
});
|
|
});
|
|
|
|
// Give the SSE connection a moment to be established before posting.
|
|
await new Promise((r) => setTimeout(r, 80));
|
|
|
|
// Trigger a mutation through the REST API.
|
|
await fetch(`${server.url}/tasks`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({ title: 'SSE e2e task', role: 'implementer' }),
|
|
});
|
|
|
|
const event = await firstEventPromise;
|
|
controller.abort(); // stop the SSE listener
|
|
|
|
expect(event.type).toBe('task');
|
|
expect(event.action).toBe('created');
|
|
expect(event.id).toBe('TSK-0001');
|
|
expect(event.title).toBe('SSE e2e task');
|
|
expect(event.role).toBe('implementer');
|
|
}, 5000);
|
|
|
|
it('delivers an agent/joined presence event from POST /announce', async () => {
|
|
const controller = new AbortController();
|
|
|
|
const firstEventPromise = new Promise<AgentHubEvent>((resolve, reject) => {
|
|
const timeout = setTimeout(() => reject(new Error('SSE: no announce event within 3 s')), 3000);
|
|
fetch(`${server.url}/events`, { signal: controller.signal })
|
|
.then(async (res) => {
|
|
if (!res.body) { clearTimeout(timeout); reject(new Error('no body')); return; }
|
|
const reader = res.body.getReader();
|
|
const dec = new TextDecoder();
|
|
let buf = '';
|
|
while (true) {
|
|
let done: boolean; let value: Uint8Array | undefined;
|
|
try { ({ done, value } = await reader.read()); } catch { break; }
|
|
if (done) break;
|
|
if (value) buf += dec.decode(value, { stream: true });
|
|
const { events, remaining } = parseSSEBuffer(buf);
|
|
buf = remaining;
|
|
for (const ev of events) { clearTimeout(timeout); resolve(ev); return; }
|
|
}
|
|
})
|
|
.catch((err: unknown) => {
|
|
if (err instanceof Error && err.name === 'AbortError') return;
|
|
reject(err);
|
|
});
|
|
});
|
|
|
|
await new Promise((r) => setTimeout(r, 80));
|
|
await fetch(`${server.url}/announce`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({ agent: 'codex', role: 'implementer' }),
|
|
});
|
|
|
|
const event = await firstEventPromise;
|
|
controller.abort();
|
|
|
|
expect(event.type).toBe('agent');
|
|
expect(event.action).toBe('joined');
|
|
expect(event.id).toBe('codex');
|
|
expect(event.role).toBe('implementer');
|
|
}, 5000);
|
|
|
|
it('role filter on /events?role= suppresses non-matching tasks', async () => {
|
|
const controller = new AbortController();
|
|
const received: AgentHubEvent[] = [];
|
|
|
|
// Subscribe with role=architect — should NOT see the implementer task.
|
|
const listenDone = fetch(`${server.url}/events?role=architect`, { signal: controller.signal })
|
|
.then(async (res) => {
|
|
if (!res.body) return;
|
|
const reader = res.body.getReader();
|
|
const dec = new TextDecoder();
|
|
let buf = '';
|
|
while (true) {
|
|
let done: boolean;
|
|
let value: Uint8Array | undefined;
|
|
try {
|
|
({ done, value } = await reader.read());
|
|
} catch {
|
|
break;
|
|
}
|
|
if (done) break;
|
|
if (value) buf += dec.decode(value, { stream: true });
|
|
const { events, remaining } = parseSSEBuffer(buf);
|
|
buf = remaining;
|
|
received.push(...events);
|
|
}
|
|
})
|
|
.catch((err: unknown) => {
|
|
if (err instanceof Error && err.name === 'AbortError') return;
|
|
throw err;
|
|
});
|
|
|
|
await new Promise((r) => setTimeout(r, 80));
|
|
|
|
// Post a task for 'implementer' role — should be filtered out.
|
|
await fetch(`${server.url}/tasks`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({ title: 'Implementer task', role: 'implementer' }),
|
|
});
|
|
|
|
// Give time for any event to arrive (it shouldn't).
|
|
await new Promise((r) => setTimeout(r, 150));
|
|
|
|
controller.abort();
|
|
await listenDone;
|
|
|
|
expect(received).toHaveLength(0);
|
|
}, 5000);
|
|
});
|
|
|
|
// ─── 5. fsWatch: CLI / direct file writes also emit (TSK-0006 part 2) ─────────
|
|
// The whole point of the watcher: a `agenthub task create` (no --server) writes
|
|
// the entity file directly, bypassing the REST emit path — yet a connected
|
|
// watcher must still be triggered. Here we drive the services directly (exactly
|
|
// what the CLI does) and assert the eventBus fires.
|
|
|
|
async function waitFor(
|
|
collected: AgentHubEvent[],
|
|
pred: (e: AgentHubEvent) => boolean,
|
|
ms = 2000,
|
|
): Promise<AgentHubEvent | undefined> {
|
|
const deadline = Date.now() + ms;
|
|
while (Date.now() < deadline) {
|
|
const hit = collected.find(pred);
|
|
if (hit) return hit;
|
|
await new Promise((r) => setTimeout(r, 25));
|
|
}
|
|
return collected.find(pred);
|
|
}
|
|
|
|
describe('fsWatch emits for non-REST (CLI/file) writes', () => {
|
|
let cwd: string;
|
|
let stop: () => void;
|
|
const collected: AgentHubEvent[] = [];
|
|
const onChange = (e: AgentHubEvent) => collected.push(e);
|
|
|
|
beforeEach(async () => {
|
|
cwd = mkdtempSync(join(tmpdir(), 'ah-fswatch-'));
|
|
init(cwd, { projectName: 'fswatch-test', yes: true });
|
|
collected.length = 0;
|
|
eventBus.on('change', onChange);
|
|
stop = startEntityWatcher(cwd);
|
|
// fs.watch (FSEvents on macOS) needs a brief moment after watch() before it
|
|
// reliably delivers; a write fired in that gap is missed. The server starts
|
|
// its watcher long before any write, so this only matters for the test.
|
|
await new Promise((r) => setTimeout(r, 250));
|
|
});
|
|
|
|
afterEach(() => {
|
|
stop();
|
|
eventBus.off('change', onChange);
|
|
collected.length = 0;
|
|
rmSync(cwd, { recursive: true, force: true });
|
|
});
|
|
|
|
it('emits task/created when a task is written via the service (CLI path)', async () => {
|
|
const task = createTask(cwd, { title: 'CLI-created task', role: 'implementer' });
|
|
const ev = await waitFor(collected, (e) => e.type === 'task' && e.id === task.id);
|
|
expect(ev).toBeDefined();
|
|
expect(ev).toMatchObject({
|
|
type: 'task',
|
|
action: 'created',
|
|
id: task.id,
|
|
title: 'CLI-created task',
|
|
role: 'implementer',
|
|
});
|
|
}, 4000);
|
|
|
|
it('emits task/updated when a task file is mutated via the service (claim)', async () => {
|
|
const task = createTask(cwd, { title: 'T', role: 'implementer' });
|
|
await waitFor(collected, (e) => e.id === task.id); // drain the created event
|
|
collected.length = 0;
|
|
|
|
claimTask(cwd, task.id, 'windows-claude');
|
|
const ev = await waitFor(collected, (e) => e.id === task.id && e.action === 'updated');
|
|
expect(ev).toBeDefined();
|
|
expect(ev).toMatchObject({ type: 'task', action: 'updated', status: 'in_progress', assignedTo: 'windows-claude' });
|
|
}, 4000);
|
|
});
|
|
|
|
// ─── 6. fsWatch + REST dedup: each change is delivered exactly once ────────────
|
|
|
|
describe('fsWatch dedup with the REST emit path', () => {
|
|
let cwd: string;
|
|
let server: Awaited<ReturnType<typeof startServer>>;
|
|
const collected: AgentHubEvent[] = [];
|
|
const onChange = (e: AgentHubEvent) => collected.push(e);
|
|
|
|
beforeEach(async () => {
|
|
cwd = mkdtempSync(join(tmpdir(), 'ah-fswatch-dedup-'));
|
|
init(cwd, { projectName: 'fswatch-dedup', yes: true });
|
|
collected.length = 0;
|
|
eventBus.on('change', onChange);
|
|
server = await startServer(cwd, { host: '127.0.0.1', port: 0 });
|
|
});
|
|
|
|
afterEach(async () => {
|
|
eventBus.off('change', onChange);
|
|
collected.length = 0;
|
|
await server.app.close();
|
|
rmSync(cwd, { recursive: true, force: true });
|
|
});
|
|
|
|
it('REST POST /tasks emits exactly once (watcher echo suppressed)', async () => {
|
|
await fetch(`${server.url}/tasks`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({ title: 'dedup task', role: 'implementer' }),
|
|
});
|
|
|
|
// Wait well past the watcher debounce so any duplicate would have landed.
|
|
await new Promise((r) => setTimeout(r, 300));
|
|
|
|
const forTask = collected.filter((e) => e.type === 'task' && e.id === 'TSK-0001');
|
|
expect(forTask).toHaveLength(1);
|
|
expect(forTask[0].action).toBe('created');
|
|
}, 4000);
|
|
});
|
|
|
|
// ─── 7. status auto-refresh: snapshot stays fresh after mutations ─────────────
|
|
|
|
describe('status auto-refresh on mutation', () => {
|
|
let cwd: string;
|
|
let server: Awaited<ReturnType<typeof startServer>>;
|
|
|
|
beforeEach(async () => {
|
|
cwd = mkdtempSync(join(tmpdir(), 'ah-status-refresh-'));
|
|
init(cwd, { projectName: 'status-refresh', 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 });
|
|
});
|
|
|
|
it('regenerates status/latest.md with the new task after POST /tasks', async () => {
|
|
await fetch(`${server.url}/tasks`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({ title: 'Fresh status task', role: 'implementer' }),
|
|
});
|
|
|
|
// Wait past the status-refresh debounce (300 ms) + write.
|
|
await new Promise((r) => setTimeout(r, 600));
|
|
|
|
const status = readFileSync(join(cwd, '.agenthub', 'status', 'latest.md'), 'utf-8');
|
|
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<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);
|
|
});
|