diff --git a/src/cli/commands/watch.ts b/src/cli/commands/watch.ts index c3347e7..df78098 100644 --- a/src/cli/commands/watch.ts +++ b/src/cli/commands/watch.ts @@ -37,13 +37,17 @@ export function parseSSEBuffer(buffer: string): { events: AgentHubEvent[]; remai for (const part of parts) { // A keepalive block looks like ":" — no data line. - const dataLines = part - .split(/\r?\n/) - .filter((l) => l.startsWith('data: ')) - .map((l) => l.slice(6)); + const lines = part.split(/\r?\n/); + const eventName = lines.find((l) => l.startsWith('event: '))?.slice(7); + if (eventName && eventName !== 'message') continue; + const idLine = lines.find((l) => l.startsWith('id: ')); + const seq = idLine ? Number.parseInt(idLine.slice(4), 10) : undefined; + const dataLines = lines.filter((l) => l.startsWith('data: ')).map((l) => l.slice(6)); if (dataLines.length === 0) continue; try { - events.push(JSON.parse(dataLines.join('\n')) as AgentHubEvent); + const event = JSON.parse(dataLines.join('\n')) as AgentHubEvent; + if (seq !== undefined && Number.isFinite(seq)) event.seq = seq; + events.push(event); } catch { // Ignore malformed JSON — should never happen in practice. } @@ -192,6 +196,7 @@ export async function watchEvents( if (options.role) url.searchParams.set('role', options.role); let response: Response; + let lastEventId: number | undefined; try { response = await fetch(url.toString(), { headers: { Accept: 'text/event-stream' }, @@ -255,6 +260,7 @@ export async function watchEvents( buffer = remaining; for (const event of events) { + if (event.seq !== undefined) lastEventId = event.seq; // Client-side role filter: skip tasks that don't match the requested // role. Non-task events (handoffs, decisions, memory) always print. if (options.role && event.type === 'task' && event.role !== undefined && event.role !== options.role) { diff --git a/src/cli/commands/work.ts b/src/cli/commands/work.ts index 9b72038..8098170 100644 --- a/src/cli/commands/work.ts +++ b/src/cli/commands/work.ts @@ -127,6 +127,7 @@ function waitAndClaim(ctx: WorkAgentContext): Promise { const deadline = ctx.timeoutSec ? Date.now() + ctx.timeoutSec * 1000 : undefined; const backoffs = ctx.reconnectBackoffMs ?? [2000, 5000, 10000]; let reconnectAttempt = 0; + let lastEventId: number | undefined; // Guards against overlapping checks (SSE-triggered vs. polling fallback). let checking = false; @@ -187,7 +188,13 @@ function waitAndClaim(ctx: WorkAgentContext): Promise { controller = new AbortController(); try { ctx.serverUrl = serverUrl; - const res = await fetch(`${serverUrl}/events`, { signal: controller.signal, headers: { Accept: 'text/event-stream' } }); + const res = await fetch(`${serverUrl}/events`, { + signal: controller.signal, + headers: { + Accept: 'text/event-stream', + ...(lastEventId !== undefined ? { 'Last-Event-ID': String(lastEventId) } : {}), + }, + }); if (!res.body) throw new Error('SSE response has no body'); // Close the gap: a task or message may have appeared between the initial @@ -212,6 +219,9 @@ function waitAndClaim(ctx: WorkAgentContext): Promise { const { events, remaining } = parseSSEBuffer(buffer); buffer = remaining; + for (const event of events) { + if (event.seq !== undefined) lastEventId = event.seq; + } // Any task event may mean a task addressed to us just opened/reopened. if (events.some((e) => e.type === 'task')) { if (await tryClaim()) return; diff --git a/src/mcp/server.ts b/src/mcp/server.ts index 45d369b..fff4af9 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -24,7 +24,7 @@ import { createMessage, listInbox } from '../core/services/messageService.js'; import { addMemory, searchMemory } from '../core/services/memoryService.js'; import { createDecision } from '../core/services/decisionService.js'; import { getStatus } from '../core/services/statusService.js'; -import { discoverServer as discoverHubServer } from '../discovery.js'; +import { discoverServer as discoverHubServer, resolveReachableServerUrl } from '../discovery.js'; import { VERSION } from '../version.js'; /** @@ -39,7 +39,10 @@ import { VERSION } from '../version.js'; * * Transport: stdio (each agent's CLI spawns `agenthub mcp` as a subprocess). */ -function resolveContext(cwd: string): { root: string; serverUrl?: string } { +export async function resolveMcpContext( + cwd: string, + options: Parameters[1] = {}, +): Promise<{ root: string; serverUrl?: string }> { const root = findProjectRoot(cwd) ?? cwd; let serverUrl = process.env.AGENTHUB_SERVER || undefined; if (!serverUrl) { @@ -49,7 +52,9 @@ function resolveContext(cwd: string): { root: string; serverUrl?: string } { // no project config — local/none } } - return { root, serverUrl }; + if (!serverUrl) return { root }; + const reachable = await resolveReachableServerUrl(serverUrl, options); + return reachable ? { root, serverUrl: reachable } : { root }; } function asText(value: unknown) { @@ -70,8 +75,14 @@ async function resolveReconnectUrl(currentUrl: string): Promise { return discovered || currentUrl; } -/** Block on the SSE stream until findClaim() returns a task, or timeout. */ -function waitForTask( +/** + * Block on the SSE stream until findClaim() returns a task, or timeout. + * Exported for the auto-wake regression tests (TSK-0230): the tests drive it + * with a findClaim that mimics the agenthub_work finder, proving the wait + * wakes on task_assign / incoming message via SSE AND via the polling + * fallback when SSE is down. + */ +export function waitForTask( serverUrl: string, findClaim: (serverUrl: string) => Promise, timeoutSec: number, @@ -83,6 +94,7 @@ function waitForTask( const deadline = Date.now() + Math.max(1, timeoutSec) * 1000; const backoffs = [2000, 5000, 10000]; let reconnectAttempt = 0; + let lastEventId: number | undefined; let poll: NodeJS.Timeout | undefined; const finish = (v: T | null) => { if (settled) return; @@ -119,7 +131,10 @@ function waitForTask( try { const res = await fetch(new URL('/events', currentUrl).toString(), { signal: controller.signal, - headers: { Accept: 'text/event-stream' }, + headers: { + Accept: 'text/event-stream', + ...(lastEventId !== undefined ? { 'Last-Event-ID': String(lastEventId) } : {}), + }, }); if (!res.body) throw new Error('SSE response has no body'); // Close the gap: a task may have arrived between the initial check and now. @@ -136,6 +151,9 @@ function waitForTask( if (value) buffer += decoder.decode(value, { stream: true }); const { events, remaining } = parseSSEBuffer(buffer); buffer = remaining; + for (const event of events) { + if (event.seq !== undefined) lastEventId = event.seq; + } // Wake on a new task, a message, or an ask (decision routed to the // architect / answered back to the asker). if (events.some((e) => e.type === 'task' || e.type === 'message' || e.type === 'ask')) { @@ -164,7 +182,7 @@ function waitForTask( } export async function startMcpServer(cwd: string): Promise { - const { root, serverUrl } = resolveContext(cwd); + const { root, serverUrl } = await resolveMcpContext(cwd); const remote = !!serverUrl; const server = new McpServer({ name: 'agenthub', version: VERSION }); diff --git a/src/server/agentHealth.ts b/src/server/agentHealth.ts new file mode 100644 index 0000000..91c1c89 --- /dev/null +++ b/src/server/agentHealth.ts @@ -0,0 +1,339 @@ +/** + * Agent health-check page, served at `GET /agent-health` (TSK-0230). + * + * One glance answers "which agents are reachable?" — and proves it: per agent + * a test message can be sent whose delivery status (sent → delivered → read) + * flips LIVE when the agent's work loop auto-wakes (HOF-0086 #8 — the button + * is the live proof that auto-wake works, no manual poke). + * + * Data sources (all reused, nothing reimplemented): + * - GET /health → the TSK-0226 traffic light (active/busy/idle/stale), + * lastSeen, busy-on/waiting task. + * - GET /messages → side-effect-free full list (listMessages) for the + * cross-messaging view + delivery tracking. NEVER the + * ?agent= inbox variant — listInbox flips unread→delivered + * as a read receipt and would fake delivery. + * - GET /events → SSE. The browser's EventSource sends Last-Event-ID on + * reconnect natively, so the durable event log (TSK-0225) + * replays what it missed. When SSE drops, the page keeps + * polling every 3s and SAYS so (transport badge). + */ + +import { loadConfig } from '../core/config.js'; +import { computeHealth, type AgentHealth } from '../core/services/presenceService.js'; +import { listMessages, type InboxMessage } from '../core/services/messageService.js'; +import { designTokensCss, escapeHtml, appHeader, appHeaderCss, taskModalHtml, newTaskModalJs } from './ui-shared.js'; + +/** JSON for safe embedding in a ; + messages: InboxMessage[]; + architect: string; +} + +function pageData(cwd: string, startedAtMs: number): HealthPageData { + const config = loadConfig(cwd); + return { + health: computeHealth(cwd, startedAtMs), + messages: listMessages(cwd).slice(-200), + architect: config.roles.architect?.preferredAgent ?? 'claude', + }; +} + +/** Initial SSR rows — the JS re-renders live afterwards (same card shape). */ +function renderAgentCardsSSR(agents: AgentHealth[]): string { + return agents + .map( + (a) => `
+
+ ${escapeHtml(a.name)} + ${escapeHtml(a.role)} + ${a.state} +
+
+
+ + +
+
+
    +
    `, + ) + .join(''); +} + +export function renderAgentHealthHtml(cwd: string, startedAtMs: number): string { + const config = loadConfig(cwd); + const data = pageData(cwd, startedAtMs); + + return ` + + + + + + AgentHub Agent Health + + + + ${appHeader(config.projectName, 'health')} +
    +
    +

    Agent Health

    + hub v${escapeHtml(data.health.version)} + uptime ${data.health.uptimeSec}s + + + + polling fallback + + seq — +
    +
    + ${renderAgentCardsSSR(data.health.agents)} +
    +
    + + + ${taskModalHtml()} + ${newTaskModalJs()} + +`; +} diff --git a/src/server/eventLog.ts b/src/server/eventLog.ts new file mode 100644 index 0000000..0bd1720 --- /dev/null +++ b/src/server/eventLog.ts @@ -0,0 +1,59 @@ +import Database from 'better-sqlite3'; +import { mkdirSync } from 'node:fs'; +import { dirname } from 'node:path'; +import { getIndexPath } from '../core/paths.js'; +import type { AgentHubEvent } from './events.js'; + +export interface LoggedAgentHubEvent extends AgentHubEvent { + seq: number; +} + +let cachedPath: string | undefined; +let cachedDb: Database.Database | undefined; + +function dbFor(cwd: string): Database.Database { + const path = getIndexPath(cwd); + if (cachedDb && cachedPath === path) return cachedDb; + cachedDb?.close(); + mkdirSync(dirname(path), { recursive: true }); + const db = new Database(path); + db.exec(` + CREATE TABLE IF NOT EXISTS hub_events ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + createdAt TEXT NOT NULL, + eventJson TEXT NOT NULL + ); + `); + cachedPath = path; + cachedDb = db; + return db; +} + +export function appendHubEvent(cwd: string, event: AgentHubEvent): LoggedAgentHubEvent { + const eventJson = JSON.stringify({ ...event, seq: undefined }); + const info = dbFor(cwd) + .prepare('INSERT INTO hub_events (createdAt, eventJson) VALUES (@createdAt, @eventJson)') + .run({ createdAt: new Date().toISOString(), eventJson }); + return { ...event, seq: Number(info.lastInsertRowid) }; +} + +export function listHubEventsAfter(cwd: string, afterSeq: number, limit = 1000): LoggedAgentHubEvent[] { + const rows = dbFor(cwd) + .prepare('SELECT seq, eventJson FROM hub_events WHERE seq > @afterSeq ORDER BY seq ASC LIMIT @limit') + .all({ afterSeq, limit }) as Array<{ seq: number; eventJson: string }>; + const events: LoggedAgentHubEvent[] = []; + for (const row of rows) { + try { + events.push({ ...(JSON.parse(row.eventJson) as AgentHubEvent), seq: row.seq }); + } catch { + // Ignore corrupt historical rows; new appends always write valid JSON. + } + } + return events; +} + +export function closeHubEventLogForTests(): void { + cachedDb?.close(); + cachedDb = undefined; + cachedPath = undefined; +} diff --git a/src/server/events.ts b/src/server/events.ts index 2a8b48c..50c55bf 100644 --- a/src/server/events.ts +++ b/src/server/events.ts @@ -1,9 +1,12 @@ import { EventEmitter } from 'node:events'; +import { appendHubEvent } from './eventLog.js'; export type AgentHubEventType = 'task' | 'handoff' | 'decision' | 'memory' | 'message' | 'ask' | 'agent'; export type AgentHubEventAction = 'created' | 'updated' | 'deleted' | 'joined' | 'left'; export interface AgentHubEvent { + /** Monotonic durable sequence id for replayable hub-change events. */ + seq?: number; type: AgentHubEventType; action: AgentHubEventAction; /** Entity id, or — for `agent` presence events — the agent name. */ @@ -60,6 +63,11 @@ eventBus.setMaxListeners(0); // signature; the other path sees it via `seenRecently()` and stays silent. const DEDUP_TTL_MS = 15_000; const recentlyEmitted = new Map(); +let durableEventCwd: string | undefined; + +export function configureDurableEvents(cwd: string): void { + durableEventCwd = cwd; +} /** Stable key for a single logical mutation of one entity revision. */ export function signatureOf(type: string, id: string, stamp: string | undefined): string { @@ -96,5 +104,13 @@ export function seenRecently(signature: string): boolean { */ export function emitChange(event: AgentHubEvent, stamp: string | undefined): void { markEmitted(signatureOf(event.type, event.id, stamp)); + if (durableEventCwd) { + try { + eventBus.publish(appendHubEvent(durableEventCwd, event)); + } catch { + eventBus.publish(event); + } + return; + } eventBus.publish(event); } diff --git a/src/server/routes.ts b/src/server/routes.ts index 51393ed..812b281 100644 --- a/src/server/routes.ts +++ b/src/server/routes.ts @@ -14,14 +14,16 @@ import { getRoster } from '../core/services/rosterService.js'; import { computeHealth, stampSeen } from '../core/services/presenceService.js'; import { loadConfig, saveConfig } from '../core/config.js'; import { renderActivityHtml } from './activity.js'; +import { renderAgentHealthHtml } from './agentHealth.js'; import { renderBoardHtml } from './board/index.js'; import { renderTeamHtml } from './team.js'; import { renderArchiveHtml } from './archive.js'; import { renderDecisionsHtml } from './decisions.js'; import { renderMessagesHtml } from './messages.js'; import { renderTaskDetailHtml } from './taskDetail.js'; -import { eventBus, emitChange } from './events.js'; +import { configureDurableEvents, eventBus, emitChange } from './events.js'; import type { AgentHubEvent } from './events.js'; +import { listHubEventsAfter } from './eventLog.js'; import type { Task, Handoff, Decision, Memory, Message, Ask } from '../core/schema.js'; import { readFile } from 'node:fs/promises'; import { fileURLToPath } from 'node:url'; @@ -47,6 +49,7 @@ function wantsHtml(request: { headers: { accept?: string } }): boolean { } export async function registerRoutes(app: FastifyInstance, cwd: string): Promise { + configureDurableEvents(cwd); // Uptime anchor for /health (buildApp ≈ server start). const startedAtMs = Date.now(); @@ -60,6 +63,13 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise // Hub health: status + version + uptime + compact counts + per-agent lights. app.get('/health', async () => computeHealth(cwd, startedAtMs)); + // Agent health-check page (TSK-0230): reachability traffic light per agent + // (same /health data), test-message send with live delivery tracking, + // cross-messaging view, and the page's own transport mode (SSE vs polling). + app.get('/agent-health', async (_request, reply) => + reply.type('text/html; charset=utf-8').send(renderAgentHealthHtml(cwd, startedAtMs)), + ); + // Static, self-contained Trello-like board. Polls /tasks, /handoffs and // /decisions on the same origin; no build step, no deps. Cached once — the // markup is constant, only the data it fetches changes. @@ -128,6 +138,12 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise // (events.ts) ensures each change is delivered exactly once. app.get('/events', async (request, reply) => { const { role } = request.query as { role?: string }; + const lastEventIdHeader = request.headers['last-event-id']; + const lastEventId = + typeof lastEventIdHeader === 'string' + ? Number.parseInt(lastEventIdHeader, 10) + : Number.parseInt((request.query as { lastEventId?: string }).lastEventId ?? '', 10); + const replayAfterSeq = Number.isFinite(lastEventId) && lastEventId >= 0 ? lastEventId : undefined; // Take full control of the raw response so Fastify doesn't interfere. reply.hijack(); @@ -159,17 +175,26 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise } }; - const listener = (event: AgentHubEvent) => { + const writeEvent = (event: AgentHubEvent) => { // Server-side role filter: skip tasks that belong to a different role. // Handoffs, decisions and memory always pass through. if (role && event.type === 'task' && event.role !== undefined && event.role !== role) { return; } - writeSse(`data: ${JSON.stringify(event)}\n\n`); + const idLine = event.seq !== undefined ? `id: ${event.seq}\n` : ''; + writeSse(`${idLine}data: ${JSON.stringify(event)}\n\n`); }; + const listener = (event: AgentHubEvent) => writeEvent(event); + eventBus.on('change', listener); + if (replayAfterSeq !== undefined) { + for (const event of listHubEventsAfter(cwd, replayAfterSeq)) { + writeEvent(event); + } + } + // Task-log lines ride a NAMED `task-log` SSE event so the board's generic // onmessage handler ignores them; only the task-detail live console listens. const logListener = (payload: unknown) => { diff --git a/src/server/ui-shared.ts b/src/server/ui-shared.ts index bcbaf2f..76df9d8 100644 --- a/src/server/ui-shared.ts +++ b/src/server/ui-shared.ts @@ -291,7 +291,7 @@ export function pageHeader(projectName: string, current: 'board' | 'team' | 'act // identically on the non-board pages. // ───────────────────────────────────────────────────────────────────────────── -export type HeaderPage = 'board' | 'team' | 'activity' | 'messages' | 'decisions' | 'archive' | 'task'; +export type HeaderPage = 'board' | 'team' | 'activity' | 'messages' | 'decisions' | 'archive' | 'task' | 'health'; /** CSS for ONLY the shared header (board v2 reuses this inside boardV2Css). */ export function appHeaderOnlyCss(): string { @@ -408,6 +408,7 @@ export function appHeader(projectName: string, current: HeaderPage): string {