chahine/agenthub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(server): agent-health page + realtime auto-wake — SSE Last-Event-ID replay (durable SQLite event log), polling fallback wake on assign/message, /agent-health with reachability ampel + test-message delivery tracking + transport badge (TSK-0230)

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
chahinebrini 2026-07-24 01:20:20 +02:00
parent 7bf52a0be3
commit f5680c6982
12 changed files with 934 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/cli/commands/watch.ts
View File

@ -37,13 +37,17 @@ export function parseSSEBuffer(buffer: string): { events: AgentHubEvent[]; remai
for (const part of parts) { for (const part of parts) {
// A keepalive block looks like ":" — no data line. // A keepalive block looks like ":" — no data line.
const dataLines = part const lines = part.split(/\r?\n/);
.split(/\r?\n/) const eventName = lines.find((l) => l.startsWith('event: '))?.slice(7);
.filter((l) => l.startsWith('data: ')) if (eventName && eventName !== 'message') continue;
.map((l) => l.slice(6)); 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; if (dataLines.length === 0) continue;
try { 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 { } catch {
// Ignore malformed JSON — should never happen in practice. // 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); if (options.role) url.searchParams.set('role', options.role);
let response: Response; let response: Response;
let lastEventId: number | undefined;
try { try {
response = await fetch(url.toString(), { response = await fetch(url.toString(), {
headers: { Accept: 'text/event-stream' }, headers: { Accept: 'text/event-stream' },
@ -255,6 +260,7 @@ export async function watchEvents(
buffer = remaining; buffer = remaining;
for (const event of events) { for (const event of events) {
if (event.seq !== undefined) lastEventId = event.seq;
// Client-side role filter: skip tasks that don't match the requested // Client-side role filter: skip tasks that don't match the requested
// role. Non-task events (handoffs, decisions, memory) always print. // role. Non-task events (handoffs, decisions, memory) always print.
if (options.role && event.type === 'task' && event.role !== undefined && event.role !== options.role) { if (options.role && event.type === 'task' && event.role !== undefined && event.role !== options.role) {

12
src/cli/commands/work.ts
View File

@ -127,6 +127,7 @@ function waitAndClaim(ctx: WorkAgentContext): Promise<void> {
const deadline = ctx.timeoutSec ? Date.now() + ctx.timeoutSec * 1000 : undefined; const deadline = ctx.timeoutSec ? Date.now() + ctx.timeoutSec * 1000 : undefined;
const backoffs = ctx.reconnectBackoffMs ?? [2000, 5000, 10000]; const backoffs = ctx.reconnectBackoffMs ?? [2000, 5000, 10000];
let reconnectAttempt = 0; let reconnectAttempt = 0;
let lastEventId: number | undefined;
// Guards against overlapping checks (SSE-triggered vs. polling fallback). // Guards against overlapping checks (SSE-triggered vs. polling fallback).
let checking = false; let checking = false;
@ -187,7 +188,13 @@ function waitAndClaim(ctx: WorkAgentContext): Promise<void> {
controller = new AbortController(); controller = new AbortController();
try { try {
ctx.serverUrl = serverUrl; 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'); if (!res.body) throw new Error('SSE response has no body');
// Close the gap: a task or message may have appeared between the initial // Close the gap: a task or message may have appeared between the initial
@ -212,6 +219,9 @@ function waitAndClaim(ctx: WorkAgentContext): Promise<void> {
const { events, remaining } = parseSSEBuffer(buffer); const { events, remaining } = parseSSEBuffer(buffer);
buffer = remaining; 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. // Any task event may mean a task addressed to us just opened/reopened.
if (events.some((e) => e.type === 'task')) { if (events.some((e) => e.type === 'task')) {
if (await tryClaim()) return; if (await tryClaim()) return;

32
src/mcp/server.ts
View File

@ -24,7 +24,7 @@ import { createMessage, listInbox } from '../core/services/messageService.js';
import { addMemory, searchMemory } from '../core/services/memoryService.js'; import { addMemory, searchMemory } from '../core/services/memoryService.js';
import { createDecision } from '../core/services/decisionService.js'; import { createDecision } from '../core/services/decisionService.js';
import { getStatus } from '../core/services/statusService.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'; 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). * 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<typeof resolveReachableServerUrl>[1] = {},
): Promise<{ root: string; serverUrl?: string }> {
const root = findProjectRoot(cwd) ?? cwd; const root = findProjectRoot(cwd) ?? cwd;
let serverUrl = process.env.AGENTHUB_SERVER || undefined; let serverUrl = process.env.AGENTHUB_SERVER || undefined;
if (!serverUrl) { if (!serverUrl) {
@ -49,7 +52,9 @@ function resolveContext(cwd: string): { root: string; serverUrl?: string } {
// no project config — local/none // 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) { function asText(value: unknown) {
@ -70,8 +75,14 @@ async function resolveReconnectUrl(currentUrl: string): Promise<string> {
return discovered || currentUrl; return discovered || currentUrl;
} }
/** Block on the SSE stream until findClaim() returns a task, or timeout. */ /**
function waitForTask<T>( * 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<T>(
serverUrl: string, serverUrl: string,
findClaim: (serverUrl: string) => Promise<T | null>, findClaim: (serverUrl: string) => Promise<T | null>,
timeoutSec: number, timeoutSec: number,
@ -83,6 +94,7 @@ function waitForTask<T>(
const deadline = Date.now() + Math.max(1, timeoutSec) * 1000; const deadline = Date.now() + Math.max(1, timeoutSec) * 1000;
const backoffs = [2000, 5000, 10000]; const backoffs = [2000, 5000, 10000];
let reconnectAttempt = 0; let reconnectAttempt = 0;
let lastEventId: number | undefined;
let poll: NodeJS.Timeout | undefined; let poll: NodeJS.Timeout | undefined;
const finish = (v: T | null) => { const finish = (v: T | null) => {
if (settled) return; if (settled) return;
@ -119,7 +131,10 @@ function waitForTask<T>(
try { try {
const res = await fetch(new URL('/events', currentUrl).toString(), { const res = await fetch(new URL('/events', currentUrl).toString(), {
signal: controller.signal, 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'); 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. // Close the gap: a task may have arrived between the initial check and now.
@ -136,6 +151,9 @@ function waitForTask<T>(
if (value) buffer += decoder.decode(value, { stream: true }); if (value) buffer += decoder.decode(value, { stream: true });
const { events, remaining } = parseSSEBuffer(buffer); const { events, remaining } = parseSSEBuffer(buffer);
buffer = remaining; 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 // Wake on a new task, a message, or an ask (decision routed to the
// architect / answered back to the asker). // architect / answered back to the asker).
if (events.some((e) => e.type === 'task' || e.type === 'message' || e.type === 'ask')) { if (events.some((e) => e.type === 'task' || e.type === 'message' || e.type === 'ask')) {
@ -164,7 +182,7 @@ function waitForTask<T>(
} }
export async function startMcpServer(cwd: string): Promise<void> { export async function startMcpServer(cwd: string): Promise<void> {
const { root, serverUrl } = resolveContext(cwd); const { root, serverUrl } = await resolveMcpContext(cwd);
const remote = !!serverUrl; const remote = !!serverUrl;
const server = new McpServer({ name: 'agenthub', version: VERSION }); const server = new McpServer({ name: 'agenthub', version: VERSION });

339
src/server/agentHealth.ts Normal file
View File

@ -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 <script> tag (no </script breakout). */
function embedJson(value: unknown): string {
return JSON.stringify(value).replace(/</g, '\\u003c');
}
interface HealthPageData {
health: ReturnType<typeof computeHealth>;
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) => `<article class="agent-card" data-agent="${escapeHtml(a.name)}">
<header class="ac-head">
<span class="ac-name">${escapeHtml(a.name)}</span>
<span class="ac-role">${escapeHtml(a.role)}</span>
<span class="light light-${a.state}" data-light>${a.state}</span>
</header>
<div class="ac-meta" data-meta></div>
<div class="ac-test">
<input class="ac-input" type="text" value="health-check ping" maxlength="200" aria-label="Test message" />
<button class="ac-send" type="button">Send test</button>
</div>
<div class="ac-track" data-track></div>
<ul class="ac-msgs" data-msgs></ul>
</article>`,
)
.join('');
}
export function renderAgentHealthHtml(cwd: string, startedAtMs: number): string {
const config = loadConfig(cwd);
const data = pageData(cwd, startedAtMs);
return `<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<meta name="color-scheme" content="dark" />
<title>AgentHub Agent Health</title>
<style>
${designTokensCss()}
${appHeaderCss()}
body { padding: 96px 20px 32px; }
main { max-width: 1120px; margin: 0 auto; display: grid; gap: 12px; }
.summary { display: flex; flex-wrap: wrap; align-items: center; gap: 10px; background: var(--surface); border: 1px solid var(--border); border-radius: 8px; padding: 12px 14px; }
.summary h1 { font-size: 18px; margin: 0; }
.chip { font: 11px/1 var(--font-mono); border: 1px solid var(--border); border-radius: 999px; padding: 5px 10px; color: var(--muted); white-space: nowrap; }
.chip b { color: var(--text); font-weight: 700; }
.spacer { flex: 1; }
/* Transport badge: SSE live vs polling fallback — visibility only. */
.transport { display: inline-flex; align-items: center; gap: 7px; font: 700 11px/1 var(--font-mono); border-radius: 999px; padding: 6px 12px; border: 1px solid var(--border); }
.transport .tdot { width: 8px; height: 8px; border-radius: 50%; }
.transport.sse { color: #adf2c7; border-color: rgba(34,197,94,.4); }
.transport.sse .tdot { background: var(--green); box-shadow: 0 0 0 3px rgba(34,197,94,.16); animation: tp 1.5s ease-in-out infinite; }
.transport.poll { color: #f5d08a; border-color: rgba(210,153,34,.45); }
.transport.poll .tdot { background: var(--status-review); box-shadow: 0 0 0 3px rgba(210,153,34,.16); }
@keyframes tp { 0%,100% { opacity: 1; } 50% { opacity: .4; } }
.agents { display: grid; grid-template-columns: repeat(auto-fill, minmax(330px, 1fr)); gap: 12px; align-items: start; }
.agent-card { background: var(--surface); border: 1px solid var(--border); border-radius: 10px; padding: 12px 14px; display: grid; gap: 9px; min-width: 0; }
.ac-head { display: flex; align-items: baseline; gap: 8px; }
.ac-name { font-weight: 700; font-size: 14px; }
.ac-role { font: 10.5px/1 var(--font-mono); color: var(--muted); text-transform: uppercase; letter-spacing: .04em; }
.ac-head .light { margin-left: auto; }
/* The TSK-0226 traffic light, same four states, same colors. */
.light { font: 700 10px/1 var(--font-mono); text-transform: uppercase; letter-spacing: .05em; border-radius: 999px; padding: 4px 9px; border: 1px solid transparent; }
.light-active { color: #adf2c7; border-color: rgba(34,197,94,.4); background: rgba(34,197,94,.08); }
.light-busy { color: #a8d0ff; border-color: rgba(88,166,255,.4); background: rgba(88,166,255,.08); }
.light-idle { color: var(--muted); border-color: var(--border); }
.light-stale { color: #ffb4b4; border-color: rgba(239,68,68,.45); background: rgba(239,68,68,.08); }
.ac-meta { font: 11px/1.5 var(--font-mono); color: var(--muted); min-height: 16px; }
.ac-meta a { color: var(--green); text-decoration: none; }
.ac-meta a:hover { text-decoration: underline; }
.ac-test { display: flex; gap: 7px; }
.ac-input { flex: 1; min-width: 0; background: var(--bg); border: 1px solid var(--border); border-radius: 7px; color: var(--text); font: 12px/1 var(--font-mono); padding: 7px 9px; }
.ac-input:focus { outline: none; border-color: var(--accent); }
.ac-send { border: 1px solid var(--border); background: var(--raised); color: var(--text); font: 600 11.5px/1 var(--font-sans); border-radius: 7px; padding: 0 12px; cursor: pointer; white-space: nowrap; }
.ac-send:hover { border-color: var(--accent); }
.ac-send[disabled] { opacity: .55; cursor: default; }
.ac-track { font: 11px/1.4 var(--font-mono); min-height: 15px; }
.ac-track .t-sent { color: var(--muted); }
.ac-track .t-delivered { color: #a8d0ff; }
.ac-track .t-read { color: #adf2c7; }
.ac-track .t-timeout { color: #ffb4b4; }
.ac-msgs { list-style: none; margin: 0; padding: 0; display: grid; gap: 5px; border-top: 1px solid var(--border); padding-top: 8px; }
.ac-msgs:empty { display: none; }
.ac-msgs li { display: grid; grid-template-columns: auto minmax(0,1fr) auto; gap: 7px; align-items: baseline; font: 11px/1.45 var(--font-mono); color: var(--muted); }
.ac-msgs .dir { flex: 0 0 auto; }
.ac-msgs .dir.out { color: #a8d0ff; }
.ac-msgs .dir.in { color: var(--status-review); }
.ac-msgs .txt { min-width: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; color: var(--text); }
.ac-msgs .mst { font-size: 9.5px; text-transform: uppercase; letter-spacing: .04em; }
.ac-msgs .mst.unread { color: var(--muted); }
.ac-msgs .mst.delivered { color: #a8d0ff; }
.ac-msgs .mst.read, .ac-msgs .mst.acked { color: #adf2c7; }
.ac-msgs-head { font: 700 9.5px/1 var(--font-mono); text-transform: uppercase; letter-spacing: .06em; color: var(--muted); }
</style>
</head>
<body>
${appHeader(config.projectName, 'health')}
<main>
<section class="summary">
<h1>Agent Health</h1>
<span class="chip">hub <b>v${escapeHtml(data.health.version)}</b></span>
<span class="chip">uptime <b data-uptime>${data.health.uptimeSec}s</b></span>
<span class="chip" data-counts></span>
<span class="spacer"></span>
<span class="transport poll" id="transport" title="Page transport: SSE (durable replay via Last-Event-ID) or 3s polling fallback">
<span class="tdot"></span><span id="transportLabel">polling fallback</span>
</span>
<span class="chip" id="seqChip" title="Last durable event seq seen (TSK-0225)">seq <b>—</b></span>
</section>
<section class="agents" id="agents">
${renderAgentCardsSSR(data.health.agents)}
</section>
</main>
<script>
window.__HEALTH_DATA__ = ${embedJson(data)};
</script>
<script>
(function() {
var data = window.__HEALTH_DATA__ || { health: { agents: [], counts: {}, uptimeSec: 0 }, messages: [], architect: 'claude' };
var ARCHITECT = data.architect;
var MSG_COUNT = 5; // last N cross-messages per agent
var TRACK_TIMEOUT_MS = 90000; // test-message delivery timeout
var track = {}; // agent -> { id, sentAt, timedOut }
var lastSeq = null, eventCount = 0;
function esc(s) { return String(s == null ? '' : s).replace(/&/g,'&amp;').replace(/</g,'&lt;').replace(/>/g,'&gt;').replace(/"/g,'&quot;'); }
function agoSec(iso) { var t = Date.parse(iso); if (isNaN(t)) return null; return Math.max(0, Math.floor((Date.now() - t) / 1000)); }
function compact(s) { if (s == null) return '—'; if (s < 60) return s + 's'; var m = Math.floor(s / 60); if (m < 60) return m + 'm'; var h = Math.floor(m / 60); if (h < 48) return h + 'h'; return Math.floor(h / 24) + 'd'; }
function snippet(text, max) { var c = String(text || '').replace(/\\s+/g, ' ').trim(); return c.length <= max ? c : c.slice(0, max - 1) + '…'; }
function setTransport(mode) {
var el = document.getElementById('transport');
var label = document.getElementById('transportLabel');
if (!el || !label) return;
el.classList.toggle('sse', mode === 'sse');
el.classList.toggle('poll', mode !== 'sse');
label.textContent = mode === 'sse' ? 'SSE live' : 'polling fallback';
}
function setSeq() {
var chip = document.getElementById('seqChip');
if (chip) chip.innerHTML = 'seq <b>' + (lastSeq == null ? '—' : esc(String(lastSeq))) + '</b> · ' + eventCount + ' events';
}
function agentMessages(agent) {
return data.messages
.filter(function(m) { return m.from === agent || m.to === agent; })
.slice(-MSG_COUNT)
.reverse();
}
function trackStatus(agent) {
var t = track[agent];
if (!t) return '';
var msg = null;
for (var i = data.messages.length - 1; i >= 0; i--) {
if (data.messages[i].id === t.id) { msg = data.messages[i]; break; }
}
var st = msg ? (msg.status || 'unread') : 'unread';
if (st === 'unread' && Date.now() - t.sentAt > TRACK_TIMEOUT_MS) {
t.timedOut = true;
return '<span class="t-timeout">✕ timeout — no delivery within ' + Math.round(TRACK_TIMEOUT_MS / 1000) + 's (agent not reachable)</span>';
}
var elapsed = compact(Math.floor((Date.now() - t.sentAt) / 1000));
if (st === 'unread') return '<span class="t-sent">● sent ' + esc(t.id) + ' — waiting for delivery… (' + elapsed + ')</span>';
if (st === 'delivered') return '<span class="t-delivered">● delivered ' + esc(t.id) + ' after ' + elapsed + ' — agent auto-woke, waiting for ack…</span>';
return '<span class="t-read">● ' + esc(st) + ' ' + esc(t.id) + ' after ' + elapsed + ' — round trip complete</span>';
}
function render() {
var h = data.health;
var counts = h.counts || {};
var countsEl = document.querySelector('[data-counts]');
if (countsEl) countsEl.innerHTML = 'tasks <b>' + (counts.tasks || 0) + '</b> · open <b>' + (counts.open || 0) + '</b> · in&nbsp;progress <b>' + (counts.inProgress || 0) + '</b> · review <b>' + (counts.review || 0) + '</b> · unread <b>' + (counts.unreadMessages || 0) + '</b>';
var upEl = document.querySelector('[data-uptime]');
if (upEl) upEl.textContent = compact(h.uptimeSec);
var wrap = document.getElementById('agents');
if (!wrap) return;
// Ensure a card exists per agent (roster can grow while the page is open).
(h.agents || []).forEach(function(a) {
if (!wrap.querySelector('.agent-card[data-agent="' + esc(a.name) + '"]')) {
var el = document.createElement('article');
el.className = 'agent-card';
el.setAttribute('data-agent', a.name);
el.innerHTML = '<header class="ac-head"><span class="ac-name"></span><span class="ac-role"></span><span class="light" data-light></span></header>' +
'<div class="ac-meta" data-meta></div>' +
'<div class="ac-test"><input class="ac-input" type="text" value="health-check ping" maxlength="200" aria-label="Test message" />' +
'<button class="ac-send" type="button">Send test</button></div>' +
'<div class="ac-track" data-track></div><ul class="ac-msgs" data-msgs></ul>';
el.querySelector('.ac-name').textContent = a.name;
el.querySelector('.ac-role').textContent = a.role;
wrap.appendChild(el);
}
});
(h.agents || []).forEach(function(a) {
var card = wrap.querySelector('.agent-card[data-agent="' + esc(a.name) + '"]');
if (!card) return;
var light = card.querySelector('[data-light]');
light.className = 'light light-' + a.state;
light.textContent = a.state;
var seen = a.lastSeenAgoSec != null ? 'last seen ' + compact(a.lastSeenAgoSec) + ' ago' : 'never seen';
var task = a.taskId ? ' · <a href="/tasks/' + esc(a.taskId) + '">' + esc(a.taskId) + '</a>' : '';
card.querySelector('[data-meta]').innerHTML = esc(seen) + task;
var tr = card.querySelector('[data-track]');
var newTrack = trackStatus(a.name);
if (tr.innerHTML !== newTrack) tr.innerHTML = newTrack;
var msgs = agentMessages(a.name);
var ul = card.querySelector('[data-msgs]');
var html = msgs.length ? '<li class="ac-msgs-head"><span>cross-messaging (last ' + msgs.length + ')</span><span></span><span></span></li>' : '';
html += msgs.map(function(m) {
var out = m.from === a.name;
var other = out ? m.to : m.from;
var when = agoSec(m.createdAt || m.updatedAt);
return '<li><span class="dir ' + (out ? 'out' : 'in') + '">' + (out ? '→' : '←') + ' ' + esc(other) + '</span>' +
'<span class="txt" title="' + esc(m.text) + '">' + esc(snippet(m.text, 90)) + '</span>' +
'<span class="mst ' + esc(m.status || 'unread') + '">' + esc(m.status || 'unread') + ' · ' + compact(when) + '</span></li>';
}).join('');
if (ul.innerHTML !== html) ul.innerHTML = html;
});
}
async function refresh() {
try {
var hr = await fetch('/health', { headers: { accept: 'application/json' } });
if (hr.ok) data.health = await hr.json();
// Full list — side-effect-free. The ?agent= inbox variant flips
// unread→delivered as a read receipt and would fake delivery here.
var mr = await fetch('/messages', { headers: { accept: 'application/json' } });
if (mr.ok) data.messages = (await mr.json()).slice(-200);
render();
} catch (e) {
setTransport('poll');
}
}
// Test-message send → live delivery tracking (the auto-wake live proof).
document.addEventListener('click', async function(e) {
var btn = e.target.closest && e.target.closest('.ac-send');
if (!btn || btn.disabled) return;
var card = btn.closest('.agent-card');
var agent = card.getAttribute('data-agent');
var input = card.querySelector('.ac-input');
var text = (input.value || 'health-check ping').trim() || 'health-check ping';
btn.disabled = true;
try {
var res = await fetch('/messages', {
method: 'POST',
headers: { 'Content-Type': 'application/json', accept: 'application/json' },
body: JSON.stringify({ from: 'health-check', to: agent, text: text + ' · ' + new Date().toISOString() }),
});
if (!res.ok) throw new Error('POST /messages failed: ' + res.status);
var msg = await res.json();
track[agent] = { id: msg.id, sentAt: Date.now(), timedOut: false };
await refresh();
} catch (err) {
var tr = card.querySelector('[data-track]');
if (tr) tr.innerHTML = '<span class="t-timeout">✕ send failed: ' + esc(err.message || String(err)) + '</span>';
} finally {
btn.disabled = false;
}
});
// Transport: SSE primary (browser EventSource resends Last-Event-ID on
// reconnect → durable replay from TSK-0225), 3s polling as the fallback.
if ('EventSource' in window) {
var es = new EventSource('/events');
es.onopen = function() { setTransport('sse'); };
es.onmessage = function(ev) {
eventCount++;
if (ev.lastEventId) lastSeq = ev.lastEventId;
setSeq();
refresh();
};
es.onerror = function() { setTransport('poll'); };
window.addEventListener('pagehide', function() { es.close(); });
} else {
setTransport('poll');
}
setTransport('poll');
setSeq();
render();
setInterval(refresh, 3000);
})();
</script>
${taskModalHtml()}
${newTaskModalJs()}
</body>
</html>`;
}

59
src/server/eventLog.ts Normal file
View File

@ -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;
}

16
src/server/events.ts
View File

@ -1,9 +1,12 @@
import { EventEmitter } from 'node:events'; import { EventEmitter } from 'node:events';
import { appendHubEvent } from './eventLog.js';
export type AgentHubEventType = 'task' | 'handoff' | 'decision' | 'memory' | 'message' | 'ask' | 'agent'; export type AgentHubEventType = 'task' | 'handoff' | 'decision' | 'memory' | 'message' | 'ask' | 'agent';
export type AgentHubEventAction = 'created' | 'updated' | 'deleted' | 'joined' | 'left'; export type AgentHubEventAction = 'created' | 'updated' | 'deleted' | 'joined' | 'left';
export interface AgentHubEvent { export interface AgentHubEvent {
/** Monotonic durable sequence id for replayable hub-change events. */
seq?: number;
type: AgentHubEventType; type: AgentHubEventType;
action: AgentHubEventAction; action: AgentHubEventAction;
/** Entity id, or — for `agent` presence events — the agent name. */ /** 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. // signature; the other path sees it via `seenRecently()` and stays silent.
const DEDUP_TTL_MS = 15_000; const DEDUP_TTL_MS = 15_000;
const recentlyEmitted = new Map<string, number>(); const recentlyEmitted = new Map<string, number>();
let durableEventCwd: string | undefined;
export function configureDurableEvents(cwd: string): void {
durableEventCwd = cwd;
}
/** Stable key for a single logical mutation of one entity revision. */ /** Stable key for a single logical mutation of one entity revision. */
export function signatureOf(type: string, id: string, stamp: string | undefined): string { 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 { export function emitChange(event: AgentHubEvent, stamp: string | undefined): void {
markEmitted(signatureOf(event.type, event.id, stamp)); markEmitted(signatureOf(event.type, event.id, stamp));
if (durableEventCwd) {
try {
eventBus.publish(appendHubEvent(durableEventCwd, event));
} catch {
eventBus.publish(event);
}
return;
}
eventBus.publish(event); eventBus.publish(event);
} }

31
src/server/routes.ts
View File

@ -14,14 +14,16 @@ import { getRoster } from '../core/services/rosterService.js';
import { computeHealth, stampSeen } from '../core/services/presenceService.js'; import { computeHealth, stampSeen } from '../core/services/presenceService.js';
import { loadConfig, saveConfig } from '../core/config.js'; import { loadConfig, saveConfig } from '../core/config.js';
import { renderActivityHtml } from './activity.js'; import { renderActivityHtml } from './activity.js';
import { renderAgentHealthHtml } from './agentHealth.js';
import { renderBoardHtml } from './board/index.js'; import { renderBoardHtml } from './board/index.js';
import { renderTeamHtml } from './team.js'; import { renderTeamHtml } from './team.js';
import { renderArchiveHtml } from './archive.js'; import { renderArchiveHtml } from './archive.js';
import { renderDecisionsHtml } from './decisions.js'; import { renderDecisionsHtml } from './decisions.js';
import { renderMessagesHtml } from './messages.js'; import { renderMessagesHtml } from './messages.js';
import { renderTaskDetailHtml } from './taskDetail.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 type { AgentHubEvent } from './events.js';
import { listHubEventsAfter } from './eventLog.js';
import type { Task, Handoff, Decision, Memory, Message, Ask } from '../core/schema.js'; import type { Task, Handoff, Decision, Memory, Message, Ask } from '../core/schema.js';
import { readFile } from 'node:fs/promises'; import { readFile } from 'node:fs/promises';
import { fileURLToPath } from 'node:url'; 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<void> { export async function registerRoutes(app: FastifyInstance, cwd: string): Promise<void> {
configureDurableEvents(cwd);
// Uptime anchor for /health (buildApp ≈ server start). // Uptime anchor for /health (buildApp ≈ server start).
const startedAtMs = Date.now(); 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. // Hub health: status + version + uptime + compact counts + per-agent lights.
app.get('/health', async () => computeHealth(cwd, startedAtMs)); 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 // Static, self-contained Trello-like board. Polls /tasks, /handoffs and
// /decisions on the same origin; no build step, no deps. Cached once — the // /decisions on the same origin; no build step, no deps. Cached once — the
// markup is constant, only the data it fetches changes. // 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. // (events.ts) ensures each change is delivered exactly once.
app.get('/events', async (request, reply) => { app.get('/events', async (request, reply) => {
const { role } = request.query as { role?: string }; 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. // Take full control of the raw response so Fastify doesn't interfere.
reply.hijack(); 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. // Server-side role filter: skip tasks that belong to a different role.
// Handoffs, decisions and memory always pass through. // Handoffs, decisions and memory always pass through.
if (role && event.type === 'task' && event.role !== undefined && event.role !== role) { if (role && event.type === 'task' && event.role !== undefined && event.role !== role) {
return; 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); 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 // 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. // onmessage handler ignores them; only the task-detail live console listens.
const logListener = (payload: unknown) => { const logListener = (payload: unknown) => {

3
src/server/ui-shared.ts
View File

@ -291,7 +291,7 @@ export function pageHeader(projectName: string, current: 'board' | 'team' | 'act
// identically on the non-board pages. // 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). */ /** CSS for ONLY the shared header (board v2 reuses this inside boardV2Css). */
export function appHeaderOnlyCss(): string { export function appHeaderOnlyCss(): string {
@ -408,6 +408,7 @@ export function appHeader(projectName: string, current: HeaderPage): string {
<nav class="b2-nav" aria-label="Primary"> <nav class="b2-nav" aria-label="Primary">
${link('Board', '/board', 'board')} ${link('Board', '/board', 'board')}
${link('Team', '/team', 'team')} ${link('Team', '/team', 'team')}
${link('Health', '/agent-health', 'health')}
${link('Activity', '/activity', 'activity')} ${link('Activity', '/activity', 'activity')}
${link('Messages', '/messages', 'messages')} ${link('Messages', '/messages', 'messages')}
${link('Decisions', '/decisions', 'decisions')} ${link('Decisions', '/decisions', 'decisions')}

99
tests/agentHealth.test.ts Normal file
View File

@ -0,0 +1,99 @@
/**
* Regression tests for the agent health-check page (TSK-0230, HOF-0085).
* - GET /agent-health serves the page with the per-agent traffic light
* (same agentLight data as /health — reused, not duplicated).
* - The test-message wiring + cross-messaging data ship with the page.
* - The page surfaces the transport mode (SSE vs polling fallback).
*/
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import { mkdtempSync, rmSync } from 'fs';
import { tmpdir } from 'os';
import { join } from 'path';
import { buildApp } from '../src/server/index.js';
import { init } from '../src/cli/commands/init.js';
describe('GET /agent-health (TSK-0230)', () => {
let cwd: string;
let app: ReturnType<typeof buildApp>;
beforeEach(() => {
cwd = mkdtempSync(join(tmpdir(), 'ah-agent-health-'));
init(cwd, { projectName: 'health-test', yes: true });
app = buildApp(cwd);
});
afterEach(() => {
rmSync(cwd, { recursive: true, force: true });
});
it('serves the health-check page as HTML with nav + roster agents', async () => {
const res = await app.inject({ method: 'GET', url: '/agent-health' });
expect(res.statusCode).toBe(200);
expect(res.headers['content-type']).toContain('text/html');
const html = res.payload;
expect(html).toContain('<title>AgentHub Agent Health</title>');
expect(html).toContain('health-test');
// Nav link from the shared header.
expect(html).toContain('/agent-health');
// Roster agents get a card with a traffic light each.
expect(html).toContain('data-agent="claude"');
expect(html).toContain('data-agent="codex"');
expect(html).toContain('data-light');
});
it('marks an unclaimed assigned agent as stale (TSK-0226 ampel, reused)', async () => {
await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'waiting work', role: 'implementer' } });
await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { assignedTo: 'codex' } });
const res = await app.inject({ method: 'GET', url: '/agent-health' });
const html = res.payload;
// codex was never seen + has an open assignment ⇒ stale (red), per agentLight.
expect(html).toContain('light-stale');
});
it('ships the test-message send + delivery-tracking wiring', async () => {
const res = await app.inject({ method: 'GET', url: '/agent-health' });
const html = res.payload;
expect(html).toContain('ac-send');
expect(html).toContain('Send test');
expect(html).toContain('data-track');
// Delivery tracking observes the side-effect-free full list (never the
// ?agent= inbox variant, whose read receipt would fake delivery).
expect(html).toContain("fetch('/messages'");
expect(html).not.toContain("fetch('/messages?agent=");
});
it('embeds cross-messaging data (direction, status) for the per-agent view', async () => {
await app.inject({
method: 'POST',
url: '/messages',
payload: { from: 'claude', to: 'codex', text: 'cross-wire check' },
});
const res = await app.inject({ method: 'GET', url: '/agent-health' });
const html = res.payload;
expect(html).toContain('cross-wire check');
expect(html).toContain('__HEALTH_DATA__');
expect(html).toContain('cross-messaging');
});
it('surfaces the transport mode badge (SSE live vs polling fallback)', async () => {
const res = await app.inject({ method: 'GET', url: '/agent-health' });
const html = res.payload;
expect(html).toContain('id="transport"');
expect(html).toContain('polling fallback');
expect(html).toContain('SSE live');
// The page consumes the durable event seq (TSK-0225) via EventSource.
expect(html).toContain("new EventSource('/events')");
expect(html).toContain('lastEventId');
});
it('keeps the /health JSON endpoint intact alongside the page', async () => {
const res = await app.inject({ method: 'GET', url: '/health' });
expect(res.statusCode).toBe(200);
const json = res.json() as { status: string; agents: Array<{ name: string; state: string }> };
expect(json.status).toBe('ok');
expect(json.agents.map((a) => a.name)).toContain('claude');
});
});

230
tests/autowake.test.ts Normal file
View File

@ -0,0 +1,230 @@
/**
* Regression tests for TSK-0230 (HOF-0086): realtime auto-wake / auto-claim.
*
* An agent sitting in the work-wait — CLI `agenthub work` AND the MCP
* `agenthub_work` path (waitForTask) — must wake with NO manual poke when
* (a) the architect assigns an existing open task to it (task_assign), or
* (b) a message arrives for it,
* over the SSE stream AND over the polling fallback alone (SSE down).
* With default settings the auto-claim must land within ≤10s.
*
* Background: TSK-0230 was assigned + messaged to kimi-ah and the waiting
* loop did not claim it — the CEO had to poke the agent manually. These
* tests pin the wake path so that class of failure cannot regress.
*/
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import { mkdtempSync, rmSync } from 'fs';
import { tmpdir } from 'os';
import { join } from 'path';
import { init } from '../src/cli/commands/init.js';
import { workAgent } from '../src/cli/commands/work.js';
import { findAddressedOpenTask, type AgentContext } from '../src/cli/commands/start.js';
import { remoteClient } from '../src/cli/remoteClient.js';
import { waitForTask } from '../src/mcp/server.js';
import { startServer } from '../src/server/index.js';
import type { Task } from '../src/core/schema.js';
const AGENT = 'kimi';
function sleep(ms: number): Promise<void> {
return new Promise((r) => setTimeout(r, ms));
}
/** Create an open implementer task that is NOT addressed to AGENT. */
async function createUnaddressedTask(serverUrl: string): Promise<string> {
const res = await fetch(`${serverUrl}/tasks`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ title: 'someone-else: not for kimi', role: 'implementer' }),
});
const task = (await res.json()) as Task;
return task.id;
}
/** The architect's task_assign: address the open task to AGENT (no claim). */
async function assignTask(serverUrl: string, id: string): Promise<void> {
await fetch(`${serverUrl}/tasks/${id}`, {
method: 'PATCH',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ assignedTo: AGENT }),
});
}
async function postMessage(serverUrl: string, text: string): Promise<void> {
await fetch(`${serverUrl}/messages`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ from: 'claude', to: AGENT, text }),
});
}
async function getTaskViaApi(serverUrl: string, id: string): Promise<Task> {
// GET /tasks/:id returns { task, body }, not a flat Task.
const res = (await fetch(`${serverUrl}/tasks/${id}`).then((r) => r.json())) as { task: Task };
return res.task;
}
/**
* Simulate "SSE down, REST fine": requests to /events hang open without ever
* delivering a frame (the nasty case — a dropped event on an otherwise-open
* stream), everything else passes through to the real fetch. The wait may
* then only wake via the polling fallback (TSK-0224).
*/
function breakSseOnly(): void {
const realFetch = globalThis.fetch;
globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url;
if (new URL(url).pathname === '/events') {
return new Promise<Response>((_resolve, reject) => {
init?.signal?.addEventListener('abort', () =>
reject(new DOMException('The operation was aborted.', 'AbortError')),
);
});
}
return realFetch(input, init);
}) as typeof fetch;
}
describe('auto-wake on task_assign / message (TSK-0230)', () => {
let cwd: string;
let server: Awaited<ReturnType<typeof startServer>>;
const realFetch = globalThis.fetch;
beforeEach(async () => {
cwd = mkdtempSync(join(tmpdir(), 'ah-autowake-'));
init(cwd, { projectName: 'autowake-test', yes: true });
server = await startServer(cwd, { host: '127.0.0.1', port: 0 });
});
afterEach(async () => {
globalThis.fetch = realFetch;
await server.app.close().catch(() => undefined);
rmSync(cwd, { recursive: true, force: true });
});
// ── CLI work loop ─────────────────────────────────────────────────────────
it('CLI: auto-claims an assigned task via SSE (no poke)', async () => {
const taskId = await createUnaddressedTask(server.url);
const workDone = workAgent({ serverUrl: server.url, projectCwd: cwd, agent: AGENT, role: 'implementer', timeoutSec: 8 });
await sleep(200);
const t0 = Date.now();
await assignTask(server.url, taskId);
await workDone;
// SSE is live: the wake must be near-instant, far below the 10s bound.
expect(Date.now() - t0).toBeLessThan(2_000);
const task = await getTaskViaApi(server.url, taskId);
expect(task.status).toBe('in_progress');
expect(task.assignedTo).toBe(AGENT);
}, 12_000);
it('CLI: auto-claims an assigned task within ≤10s with SSE DOWN (polling fallback only, default interval)', async () => {
breakSseOnly();
const taskId = await createUnaddressedTask(server.url);
// Default pollIntervalMs (4000) on purpose: this is the ≤10s acceptance proof.
const workDone = workAgent({ serverUrl: server.url, projectCwd: cwd, agent: AGENT, role: 'implementer', timeoutSec: 10 });
await sleep(300);
const t0 = Date.now();
await assignTask(server.url, taskId);
await workDone;
expect(Date.now() - t0).toBeLessThan(10_000);
const task = await getTaskViaApi(server.url, taskId);
expect(task.status).toBe('in_progress');
expect(task.assignedTo).toBe(AGENT);
}, 15_000);
it('CLI: wakes on an incoming message with SSE DOWN (delivered, not read)', async () => {
breakSseOnly();
const workDone = workAgent({
serverUrl: server.url,
projectCwd: cwd,
agent: AGENT,
role: 'implementer',
timeoutSec: 8,
pollIntervalMs: 400,
});
await sleep(300);
await postMessage(server.url, 'wake without poke (sse down)');
await workDone; // resolves because the message woke the loop, not on timeout
const inbox = (await fetch(`${server.url}/messages?agent=${AGENT}`).then((r) => r.json())) as Array<{
text: string;
status: string;
}>;
const m = inbox.find((x) => x.text.includes('wake without poke'));
expect(m).toBeDefined();
// Surfaced (unread → delivered) but NOT auto-read — a timeout would have left it unread.
expect(m?.status).toBe('delivered');
}, 12_000);
// ── MCP agenthub_work path (waitForTask) ──────────────────────────────────
/** Mimics the agenthub_work finder: addressed-open-task lookup + claim. */
function mcpWorkFinder(ctx: AgentContext) {
return async (url: string) => {
ctx.serverUrl = url;
const found = await findAddressedOpenTask(ctx);
if (!found) return null;
await remoteClient.claimTask(url, found.task.id, ctx.agent);
return found.task;
};
}
it('MCP: waitForTask auto-claims an assigned task via SSE (no poke)', async () => {
const ctx: AgentContext = { serverUrl: server.url, projectCwd: cwd, agent: AGENT, role: 'implementer' };
const taskId = await createUnaddressedTask(server.url);
const waiting = waitForTask(server.url, mcpWorkFinder(ctx), 8);
await sleep(200);
const t0 = Date.now();
await assignTask(server.url, taskId);
const claimed = await waiting;
// SSE is live: the wake must be near-instant, far below the 10s bound.
expect(Date.now() - t0).toBeLessThan(2_000);
expect(claimed?.id).toBe(taskId);
const task = await getTaskViaApi(server.url, taskId);
expect(task.status).toBe('in_progress');
expect(task.assignedTo).toBe(AGENT);
}, 12_000);
it('MCP: waitForTask auto-claims an assigned task within ≤10s with SSE DOWN (polling fallback only)', async () => {
breakSseOnly();
const ctx: AgentContext = { serverUrl: server.url, projectCwd: cwd, agent: AGENT, role: 'implementer' };
const taskId = await createUnaddressedTask(server.url);
const waiting = waitForTask(server.url, mcpWorkFinder(ctx), 10);
await sleep(300);
const t0 = Date.now();
await assignTask(server.url, taskId);
const claimed = await waiting;
expect(Date.now() - t0).toBeLessThan(10_000);
expect(claimed?.id).toBe(taskId);
const task = await getTaskViaApi(server.url, taskId);
expect(task.status).toBe('in_progress');
}, 15_000);
it('MCP: waitForTask wakes on an incoming message with SSE DOWN', async () => {
breakSseOnly();
const finder = async (url: string) => {
const msgs = await remoteClient.getInbox(url, AGENT, true);
return msgs.length ? { messages: msgs } : null;
};
const waiting = waitForTask(server.url, finder, 8);
await sleep(300);
await postMessage(server.url, 'mcp message wake (sse down)');
const hit = await waiting;
expect(hit).not.toBeNull();
expect(hit?.messages[0]?.text).toContain('mcp message wake');
// listInbox read receipt: surfaced as delivered, never auto-read.
expect(hit?.messages[0]?.status).toBe('delivered');
}, 12_000);
});

37
tests/mcp.test.ts Normal file
View File

@ -0,0 +1,37 @@
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { init } from '../src/cli/commands/init.js';
import { loadConfig, saveConfig } from '../src/core/config.js';
import { resolveMcpContext } from '../src/mcp/server.js';
describe('MCP context resolution', () => {
let cwd: string;
const originalEnv = process.env.AGENTHUB_SERVER;
beforeEach(() => {
cwd = mkdtempSync(join(tmpdir(), 'ah-mcp-'));
init(cwd, { projectName: 'mcp-test', yes: true });
delete process.env.AGENTHUB_SERVER;
});
afterEach(() => {
if (originalEnv === undefined) delete process.env.AGENTHUB_SERVER;
else process.env.AGENTHUB_SERVER = originalEnv;
rmSync(cwd, { recursive: true, force: true });
});
it('self-heals a stale configured server URL before MCP tools call remote APIs', async () => {
const config = loadConfig(cwd);
config.serverUrl = 'http://agenthub.local:3377';
saveConfig(cwd, config);
const context = await resolveMcpContext(cwd, {
probe: async () => false,
discover: async () => 'http://127.0.0.1:3377',
});
expect(context).toEqual({ root: cwd, serverUrl: 'http://127.0.0.1:3377' });
});
});

77
tests/sse.test.ts
View File

@ -19,6 +19,7 @@ import { parseSSEBuffer, formatEvent, watchEvents } from '../src/cli/commands/wa
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';
import { listHubEventsAfter } from '../src/server/eventLog.js';
// ─── 1. Unit: SSE buffer parser ────────────────────────────────────────────── // ─── 1. Unit: SSE buffer parser ──────────────────────────────────────────────
@ -74,6 +75,19 @@ describe('parseSSEBuffer', () => {
const { events } = parseSSEBuffer(buf); const { events } = parseSSEBuffer(buf);
expect(events).toHaveLength(0); expect(events).toHaveLength(0);
}); });
it('attaches the SSE id field as event.seq', () => {
const buf = 'id: 42\ndata: {"type":"task","action":"created","id":"TSK-0042"}\n\n';
const { events } = parseSSEBuffer(buf);
expect(events).toHaveLength(1);
expect(events[0]).toMatchObject({ id: 'TSK-0042', seq: 42 });
});
it('ignores named task-log events in the generic change parser', () => {
const buf = 'event: task-log\ndata: {"taskId":"TSK-0001","text":"progress"}\n\n';
const { events } = parseSSEBuffer(buf);
expect(events).toHaveLength(0);
});
}); });
// ─── 2. Unit: formatEvent ──────────────────────────────────────────────────── // ─── 2. Unit: formatEvent ────────────────────────────────────────────────────
@ -168,6 +182,7 @@ describe('eventBus mutations', () => {
await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'Bus task', role: 'implementer' } }); await app.inject({ method: 'POST', url: '/tasks', payload: { title: 'Bus task', role: 'implementer' } });
expect(collected).toHaveLength(1); expect(collected).toHaveLength(1);
expect(collected[0]).toMatchObject({ type: 'task', action: 'created', id: 'TSK-0001', title: 'Bus task', role: 'implementer' }); expect(collected[0]).toMatchObject({ type: 'task', action: 'created', id: 'TSK-0001', title: 'Bus task', role: 'implementer' });
expect(collected[0].seq).toBe(1);
}); });
it('emits task/updated when PATCH /tasks/:id changes status', async () => { it('emits task/updated when PATCH /tasks/:id changes status', async () => {
@ -177,6 +192,7 @@ describe('eventBus mutations', () => {
await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'done' } }); await app.inject({ method: 'PATCH', url: '/tasks/TSK-0001', payload: { status: 'done' } });
expect(collected).toHaveLength(1); expect(collected).toHaveLength(1);
expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', id: 'TSK-0001', status: 'done' }); expect(collected[0]).toMatchObject({ type: 'task', action: 'updated', id: 'TSK-0001', status: 'done' });
expect(collected[0].seq).toBe(2);
}); });
it('emits task/updated with assignedTo when a task is claimed', async () => { it('emits task/updated with assignedTo when a task is claimed', async () => {
@ -441,6 +457,67 @@ describe('SSE stream e2e', () => {
}, 5000); }, 5000);
}); });
describe('SSE durable replay', () => {
let cwd: string;
let server: Awaited<ReturnType<typeof startServer>>;
beforeEach(async () => {
cwd = mkdtempSync(join(tmpdir(), 'ah-sse-replay-'));
init(cwd, { projectName: 'sse-replay-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('persists monotonic ids and replays missed events after Last-Event-ID', async () => {
await fetch(`${server.url}/tasks`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ title: 'first', role: 'implementer' }),
});
await fetch(`${server.url}/tasks`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ title: 'second', role: 'implementer' }),
});
expect(listHubEventsAfter(cwd, 0).map((e) => e.seq)).toEqual([1, 2]);
const controller = new AbortController();
const res = await fetch(`${server.url}/events`, {
signal: controller.signal,
headers: { Accept: 'text/event-stream', 'Last-Event-ID': '1' },
});
expect(res.ok).toBe(true);
expect(res.body).toBeTruthy();
const reader = res.body!.getReader();
const decoder = new TextDecoder();
let buffer = '';
const deadline = Date.now() + 1500;
while (Date.now() < deadline) {
const { done, value } = await reader.read();
if (done) break;
if (value) buffer += decoder.decode(value, { stream: true });
const parsed = parseSSEBuffer(buffer);
buffer = parsed.remaining;
if (parsed.events.length) {
await reader.cancel();
controller.abort();
expect(parsed.events).toHaveLength(1);
expect(parsed.events[0]).toMatchObject({ seq: 2, id: 'TSK-0002', title: 'second' });
return;
}
}
await reader.cancel().catch(() => undefined);
controller.abort();
throw new Error('Timed out waiting for replayed event');
}, 5000);
});
// ─── 5. fsWatch: CLI / direct file writes also emit (TSK-0006 part 2) ───────── // ─── 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 whole point of the watcher: a `agenthub task create` (no --server) writes
// the entity file directly, bypassing the REST emit path — yet a connected // the entity file directly, bypassing the REST emit path — yet a connected