7bf52a0be3
- Watchdog (configurable, ~60s): re-emits SSE + unread reminder + board warn log for unclaimed assigned tasks (3min high/critical, 10min else); alerts the architect on silent in_progress tasks (>15min, no auto-reassign); per-task cooldown against alert spam. - GET /health (status/version/uptime/counts + per-agent traffic light) with lastSeen stamped on announce/claim/review/log/message; board sidebar shows the agent ampel; new 'agenthub health' CLI command. - Budget: architect coordination actions (reviews, handoffs, messages, approvals) surface as estimated activity tokens + an 'actions' counter. - Version bump 0.10.2 (single source: src/version.ts).
377 lines
21 KiB
TypeScript
377 lines
21 KiB
TypeScript
import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
|
|
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
|
|
import { z } from 'zod';
|
|
import { findProjectRoot } from '../core/paths.js';
|
|
import { loadConfig } from '../core/config.js';
|
|
import { remoteClient } from '../cli/remoteClient.js';
|
|
import { parseSSEBuffer } from '../cli/commands/watch.js';
|
|
import { findAddressedOpenTask, listReviewTasks, isReviewerRole, type AgentContext } from '../cli/commands/start.js';
|
|
import {
|
|
listTasks,
|
|
getTask,
|
|
createTask,
|
|
claimTask,
|
|
assignTask,
|
|
reviewTask,
|
|
reopenTask,
|
|
doneTask,
|
|
} from '../core/services/taskService.js';
|
|
import { createHandoff, getHandoff } from '../core/services/handoffService.js';
|
|
import { appendTaskLog } from '../core/services/taskLogService.js';
|
|
import { createAsk, listAsks, answerAsk, escalateAsk } from '../core/services/askService.js';
|
|
import type { Ask } from '../core/schema.js';
|
|
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 { VERSION } from '../version.js';
|
|
|
|
/**
|
|
* AgentHub MCP server (TSK-0030).
|
|
*
|
|
* Exposes the hub operations as structured MCP tools instead of CLI strings —
|
|
* so an agent invokes a typed tool (it can't narrate it away, hallucinate a
|
|
* missing command, or get the syntax wrong). It's a thin layer over the SAME
|
|
* core: when a hub server is configured/reachable it proxies through the REST
|
|
* API (remoteClient) — so the board, SSE stream, status auto-refresh and CLI
|
|
* all keep working unchanged — otherwise it falls back to the local services.
|
|
*
|
|
* Transport: stdio (each agent's CLI spawns `agenthub mcp` as a subprocess).
|
|
*/
|
|
function resolveContext(cwd: string): { root: string; serverUrl?: string } {
|
|
const root = findProjectRoot(cwd) ?? cwd;
|
|
let serverUrl = process.env.AGENTHUB_SERVER || undefined;
|
|
if (!serverUrl) {
|
|
try {
|
|
serverUrl = loadConfig(root).serverUrl;
|
|
} catch {
|
|
// no project config — local/none
|
|
}
|
|
}
|
|
return { root, serverUrl };
|
|
}
|
|
|
|
function asText(value: unknown) {
|
|
return { content: [{ type: 'text' as const, text: typeof value === 'string' ? value : JSON.stringify(value, null, 2) }] };
|
|
}
|
|
|
|
function remainingMs(deadline: number | undefined): number {
|
|
return deadline === undefined ? Number.POSITIVE_INFINITY : Math.max(0, deadline - Date.now());
|
|
}
|
|
|
|
function sleep(ms: number): Promise<void> {
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
}
|
|
|
|
async function resolveReconnectUrl(currentUrl: string): Promise<string> {
|
|
if (process.env.AGENTHUB_SERVER) return process.env.AGENTHUB_SERVER;
|
|
const discovered = await discoverHubServer(2000);
|
|
return discovered || currentUrl;
|
|
}
|
|
|
|
/** Block on the SSE stream until findClaim() returns a task, or timeout. */
|
|
function waitForTask<T>(
|
|
serverUrl: string,
|
|
findClaim: (serverUrl: string) => Promise<T | null>,
|
|
timeoutSec: number,
|
|
): Promise<T | null> {
|
|
return new Promise((resolve) => {
|
|
let settled = false;
|
|
let controller: AbortController | undefined;
|
|
let currentUrl = serverUrl;
|
|
const deadline = Date.now() + Math.max(1, timeoutSec) * 1000;
|
|
const backoffs = [2000, 5000, 10000];
|
|
let reconnectAttempt = 0;
|
|
let poll: NodeJS.Timeout | undefined;
|
|
const finish = (v: T | null) => {
|
|
if (settled) return;
|
|
settled = true;
|
|
if (poll) clearInterval(poll);
|
|
try { controller?.abort(); } catch { /* already */ }
|
|
resolve(v);
|
|
};
|
|
const timer = setTimeout(() => finish(null), Math.max(1, timeoutSec) * 1000);
|
|
|
|
// Polling fallback: a dropped SSE frame on an otherwise-open stream must
|
|
// never mean an infinite sleep — re-run findClaim every few seconds, so a
|
|
// lost event costs at most one poll interval (~4s).
|
|
let polling = false;
|
|
poll = setInterval(() => {
|
|
if (settled || polling) return;
|
|
polling = true;
|
|
void (async () => {
|
|
try {
|
|
const hit = await findClaim(currentUrl);
|
|
if (hit) { clearTimeout(timer); finish(hit); }
|
|
} catch {
|
|
/* best-effort: the SSE path and the next tick remain */
|
|
} finally {
|
|
polling = false;
|
|
}
|
|
})();
|
|
}, 4000);
|
|
poll.unref?.();
|
|
|
|
const waitLoop = async () => {
|
|
while (!settled && remainingMs(deadline) > 0) {
|
|
controller = new AbortController();
|
|
try {
|
|
const res = await fetch(new URL('/events', currentUrl).toString(), {
|
|
signal: controller.signal,
|
|
headers: { Accept: 'text/event-stream' },
|
|
});
|
|
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.
|
|
const early = await findClaim(currentUrl);
|
|
if (early) { clearTimeout(timer); finish(early); return; }
|
|
const reader = res.body.getReader();
|
|
const decoder = new TextDecoder();
|
|
let buffer = '';
|
|
reconnectAttempt = 0;
|
|
while (!settled) {
|
|
let done: boolean; let value: Uint8Array | undefined;
|
|
try { ({ done, value } = await reader.read()); } catch { break; }
|
|
if (done) break;
|
|
if (value) buffer += decoder.decode(value, { stream: true });
|
|
const { events, remaining } = parseSSEBuffer(buffer);
|
|
buffer = remaining;
|
|
// 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')) {
|
|
const claimed = await findClaim(currentUrl);
|
|
if (claimed) { clearTimeout(timer); finish(claimed); return; }
|
|
}
|
|
}
|
|
} catch (err: unknown) {
|
|
if (settled || (err instanceof Error && err.name === 'AbortError')) return;
|
|
}
|
|
if (settled || remainingMs(deadline) <= 0) break;
|
|
currentUrl = await resolveReconnectUrl(currentUrl);
|
|
const backoff = backoffs[Math.min(reconnectAttempt, backoffs.length - 1)] ?? 10_000;
|
|
reconnectAttempt += 1;
|
|
await sleep(Math.min(backoff, remainingMs(deadline)));
|
|
}
|
|
clearTimeout(timer);
|
|
finish(null);
|
|
};
|
|
|
|
waitLoop().catch(() => {
|
|
clearTimeout(timer);
|
|
finish(null);
|
|
});
|
|
});
|
|
}
|
|
|
|
export async function startMcpServer(cwd: string): Promise<void> {
|
|
const { root, serverUrl } = resolveContext(cwd);
|
|
const remote = !!serverUrl;
|
|
const server = new McpServer({ name: 'agenthub', version: VERSION });
|
|
|
|
server.tool('agenthub_hello', 'Announce yourself to the hub (presence). Run once when you start.',
|
|
{ agent: z.string(), role: z.string().optional() },
|
|
async ({ agent, role }) => {
|
|
if (remote) await remoteClient.announce(serverUrl!, agent, role ?? 'implementer');
|
|
return asText(`AgentHub: ${agent} joined (${role ?? 'implementer'})`);
|
|
});
|
|
|
|
server.tool('agenthub_work',
|
|
'Block until there is work for you, then return it. IMPLEMENTER: waits for a task addressed to you (newly delegated OR reopened), claims it, returns it with its handoff (loop: work -> implement -> agenthub_task_review -> work). ARCHITECT/REVIEWER: waits for any task submitted to review and returns the pending set to approve (agenthub_task_done) or reject (agenthub_task_reopen) — so review submissions reach you in realtime.',
|
|
{ agent: z.string(), role: z.string().optional(), timeoutSec: z.number().optional(), unattended: z.boolean().optional() },
|
|
async ({ agent, role, timeoutSec, unattended }) => {
|
|
const ctx: AgentContext = { serverUrl, projectCwd: root, agent, role: role ?? 'implementer' };
|
|
const reviewer = isReviewerRole(ctx.role);
|
|
const useServerUrl = (nextServerUrl?: string) => {
|
|
if (nextServerUrl) ctx.serverUrl = nextServerUrl;
|
|
return ctx.serverUrl!;
|
|
};
|
|
// Fetch the agent's unread messages so the work loop surfaces them.
|
|
// Surfacing flips them `unread` → `delivered` (the listInbox read
|
|
// receipt), but nothing auto-marks them `read`: the message stays
|
|
// visible in the inbox until an explicit ack/read, and the loop wakes
|
|
// only on `unread`, so a delivered message never re-wakes it (no spin).
|
|
const drainInbox = async (nextServerUrl?: string) => {
|
|
const activeUrl = nextServerUrl ? useServerUrl(nextServerUrl) : ctx.serverUrl;
|
|
return remote ? await remoteClient.getInbox(activeUrl!, agent, true) : listInbox(root, agent, { unreadOnly: true });
|
|
};
|
|
const findWork = async (nextServerUrl?: string) => {
|
|
if (nextServerUrl) useServerUrl(nextServerUrl);
|
|
const found = await findAddressedOpenTask(ctx);
|
|
if (found) {
|
|
if (remote) await remoteClient.claimTask(ctx.serverUrl!, found.task.id, agent);
|
|
else claimTask(root, found.task.id, agent);
|
|
const detail = remote ? await remoteClient.getTask(ctx.serverUrl!, found.task.id) : getTask(root, found.task.id);
|
|
const hofEntry = found.handoffs.find((h) => h.taskId === found.task.id);
|
|
let handoff: unknown = null;
|
|
if (hofEntry) {
|
|
try { handoff = remote ? await remoteClient.getHandoff(ctx.serverUrl!, hofEntry.id) : getHandoff(root, hofEntry.id); } catch { /* optional */ }
|
|
}
|
|
return { claimed: found.task, body: (detail as { body?: string }).body, handoff, messages: await drainInbox(ctx.serverUrl) };
|
|
}
|
|
const messages = await drainInbox(ctx.serverUrl);
|
|
if (messages.length) return { claimed: null, messages, note: 'No task addressed to you, but you have messages — reply with agenthub_message.' };
|
|
return null;
|
|
};
|
|
// Architect/reviewer variant: wake on tasks submitted to review (not on
|
|
// tasks addressed to you). Returns the pending review set — never claims.
|
|
const listPendingAsks = async (): Promise<Ask[]> =>
|
|
remote ? await remoteClient.listAsks(ctx.serverUrl!, { status: 'pending' }) : listAsks(root, { status: 'pending' });
|
|
const findReview = async (nextServerUrl?: string) => {
|
|
if (nextServerUrl) useServerUrl(nextServerUrl);
|
|
const reviews = await listReviewTasks(ctx);
|
|
const messages = await drainInbox(ctx.serverUrl);
|
|
const asks = await listPendingAsks();
|
|
if (reviews.length || asks.length) {
|
|
return {
|
|
reviews: reviews.map((r) => ({ id: r.id, title: r.title, assignedTo: r.assignedTo })),
|
|
asks: asks.map((a) => ({ id: a.id, from: a.from, question: a.question, taskId: a.taskId })),
|
|
messages,
|
|
note: 'Tasks awaiting your review (approve: agenthub_task_done; reject: agenthub_task_reopen). Pending asks: answer with agenthub_ask_answer, or escalate to the CEO with agenthub_ask_escalate (release/publish/push, OSS, architecture pivots).',
|
|
};
|
|
}
|
|
if (messages.length) return { reviews: [], asks: [], messages, note: 'Nothing in review, but you have messages — reply with agenthub_message.' };
|
|
return null;
|
|
};
|
|
|
|
const finder: (nextServerUrl?: string) => Promise<Record<string, unknown> | null> = reviewer ? findReview : findWork;
|
|
|
|
// Self-perpetuating loop: every response reminds the agent to relaunch
|
|
// agenthub_work, so a finished task/message never leaves it dormant.
|
|
let LOOP = reviewer
|
|
? 'STAY REACHABLE: the instant you finish reviewing, call agenthub_work again — ' +
|
|
'run it in the background so the next review submission or ask reaches you in realtime. ' +
|
|
'Never end your turn without relaunching agenthub_work.'
|
|
: 'STAY REACHABLE: the instant you finish handling this, call agenthub_work again — ' +
|
|
'run it in the background so the next task or message reaches you in realtime. ' +
|
|
'Report meaningful progress with agenthub_task_log while you work. ' +
|
|
'ESPECIALLY after you submit with agenthub_task_review: do NOT end your turn — ' +
|
|
'immediately relaunch agenthub_work and keep it running, so the architect\'s approval, ' +
|
|
'reopen (with feedback) or follow-up message wakes you here instead of leaving you dormant. ' +
|
|
'Never end your turn without relaunching agenthub_work.';
|
|
if (unattended) {
|
|
LOOP += ' UNATTENDED MODE: never pause for human input — when you need a decision, ' +
|
|
'call agenthub_ask (it routes to the architect) and await the answer instead of stalling.';
|
|
}
|
|
const immediate = await finder();
|
|
if (immediate) return asText({ ...immediate, loop: LOOP });
|
|
const emptyMsg = reviewer
|
|
? `No task in review for ${agent}, and no hub server to wait on.`
|
|
: `No open task addressed to ${agent}, and no hub server to wait on.`;
|
|
if (!remote) return asText(emptyMsg);
|
|
const result = await waitForTask(serverUrl!, finder, timeoutSec ?? 300);
|
|
if (result) return asText({ ...result, loop: LOOP });
|
|
const kind = reviewer ? 'review submission' : 'task or message';
|
|
return asText(`No ${kind} for ${agent} within ${timeoutSec ?? 300}s. ${LOOP}`);
|
|
});
|
|
|
|
server.tool('agenthub_task_list', 'List tasks, optionally filtered by status and/or role.',
|
|
{ status: z.string().optional(), role: z.string().optional() },
|
|
async ({ status, role }) => asText(remote ? await remoteClient.listTasks(serverUrl!, { status, role }) : listTasks(root, { status, role })));
|
|
|
|
server.tool('agenthub_task_show', 'Show one task with its full body.',
|
|
{ id: z.string() },
|
|
async ({ id }) => asText(remote ? await remoteClient.getTask(serverUrl!, id) : (() => { const t = getTask(root, id); return { task: t.task, body: t.body }; })()));
|
|
|
|
server.tool('agenthub_task_create', 'Create a task.',
|
|
{
|
|
title: z.string(),
|
|
role: z.enum(['architect', 'implementer', 'reviewer', 'tester']).optional(),
|
|
priority: z.enum(['low', 'medium', 'high', 'critical']).optional(),
|
|
},
|
|
async (o) => asText(remote ? await remoteClient.createTask(serverUrl!, o) : createTask(root, o)));
|
|
|
|
server.tool('agenthub_task_assign', 'Address an open task to an agent without claiming it (architect). The agent\'s agenthub_work then auto-claims it.',
|
|
{ id: z.string(), agent: z.string() },
|
|
async ({ id, agent }) => asText(remote ? await remoteClient.assignTask(serverUrl!, id, agent) : assignTask(root, id, agent)));
|
|
|
|
server.tool('agenthub_task_claim', 'Claim a task (set in_progress + assignedTo).',
|
|
{ id: z.string(), agent: z.string() },
|
|
async ({ id, agent }) => asText(remote ? await remoteClient.claimTask(serverUrl!, id, agent) : claimTask(root, id, agent)));
|
|
|
|
server.tool('agenthub_task_review', 'Submit a finished task for architect review. Implementers use THIS, never agenthub_task_done.',
|
|
{ id: z.string() },
|
|
async ({ id }) => asText(remote ? await remoteClient.reviewTask(serverUrl!, id) : reviewTask(root, id)));
|
|
|
|
server.tool('agenthub_task_reopen', 'Re-trigger a task after review (architect: send back to the implementer).',
|
|
{ id: z.string() },
|
|
async ({ id }) => asText(remote ? await remoteClient.reopenTask(serverUrl!, id) : reopenTask(root, id)));
|
|
|
|
server.tool('agenthub_task_done', 'Approve and close a task. ARCHITECT ONLY — implementers must use agenthub_task_review.',
|
|
{ id: z.string() },
|
|
async ({ id }) => asText(remote ? await remoteClient.doneTask(serverUrl!, id, {}) : doneTask(root, id)));
|
|
|
|
server.tool('agenthub_task_log',
|
|
'Report meaningful progress on the task you are working on — one short line — so the architect can watch it live on the task console. Call it as you work (e.g. "wrote failing test", "green: 12 tests", "blocked on X").',
|
|
{ id: z.string(), text: z.string(), agent: z.string().optional(), level: z.string().optional() },
|
|
async ({ id, text, agent, level }) => asText(remote ? await remoteClient.appendTaskLog(serverUrl!, id, { text, agent, level }) : appendTaskLog(root, id, { text, agent, level })));
|
|
|
|
server.tool('agenthub_memory_add', 'Record a result / finding / blocker as a memory entry.',
|
|
{ title: z.string(), category: z.enum(['architecture', 'product', 'technical', 'implementation', 'lesson']).optional(), content: z.string() },
|
|
async (o) => asText(remote ? await remoteClient.addMemory(serverUrl!, o) : addMemory(root, o)));
|
|
|
|
server.tool('agenthub_memory_search', 'Search memory and tasks.',
|
|
{ q: z.string() },
|
|
async ({ q }) => asText(remote ? await remoteClient.searchMemory(serverUrl!, q) : searchMemory(root, q)));
|
|
|
|
server.tool('agenthub_handoff_read', 'Read a handoff (scope + context for a task).',
|
|
{ id: z.string() },
|
|
async ({ id }) => asText(remote ? await remoteClient.getHandoff(serverUrl!, id) : (() => { const h = getHandoff(root, id); return { handoff: h.handoff, body: h.body }; })()));
|
|
|
|
server.tool('agenthub_handoff_create', 'Create a handoff (architect delegates / gives feedback).',
|
|
{ fromRole: z.string(), toRole: z.string(), taskId: z.string().optional(), summary: z.string(), context: z.string().optional() },
|
|
async (o) => asText(remote ? await remoteClient.createHandoff(serverUrl!, o) : createHandoff(root, o)));
|
|
|
|
server.tool('agenthub_decision_create', 'Record an architecture/technical decision.',
|
|
{ title: z.string(), context: z.string().optional(), decision: z.string() },
|
|
async (o) => asText(remote ? await remoteClient.createDecision(serverUrl!, o) : createDecision(root, o)));
|
|
|
|
server.tool('agenthub_status', 'Show the current project status.',
|
|
{},
|
|
async () => asText(remote ? await remoteClient.getStatus(serverUrl!) : getStatus(root)));
|
|
|
|
server.tool('agenthub_message',
|
|
'Send a direct message to another agent (architect↔agent or agent↔agent) — for questions, clarifications and pings. Routed through the hub; the architect sees all messages.',
|
|
{ from: z.string(), to: z.string(), text: z.string(), taskId: z.string().optional() },
|
|
async (o) => asText(remote ? await remoteClient.sendMessage(serverUrl!, o) : createMessage(root, o)));
|
|
|
|
server.tool('agenthub_inbox',
|
|
'Read messages addressed to you. Pass unreadOnly:true for just the new ones. agenthub_work also surfaces unread messages automatically.',
|
|
{ agent: z.string(), unreadOnly: z.boolean().optional() },
|
|
async ({ agent, unreadOnly }) => asText(remote ? await remoteClient.getInbox(serverUrl!, agent, unreadOnly) : listInbox(root, agent, { unreadOnly })));
|
|
|
|
server.tool('agenthub_ask',
|
|
'Ask the architect a blocking question when you cannot decide alone (autonomous decision-routing). Routes to the architect — NEVER the CEO. With wait:true it blocks until the architect answers or escalates, then returns the resolution. Await exactly ONE ask at a time.',
|
|
{ from: z.string(), question: z.string(), taskId: z.string().optional(), wait: z.boolean().optional(), timeoutSec: z.number().optional() },
|
|
async ({ from, question, taskId, wait, timeoutSec }) => {
|
|
const ask = remote ? await remoteClient.createAsk(serverUrl!, { from, question, taskId }) : createAsk(root, { from, question, taskId });
|
|
if (!wait) return asText(ask);
|
|
if (!remote) return asText({ ...ask, note: 'Created; --wait needs a running hub server.' });
|
|
const settled = await waitForTask<Ask>(serverUrl!, async (url) => {
|
|
const { ask: cur } = await remoteClient.getAsk(url, ask.id);
|
|
return cur.status !== 'pending' ? cur : null;
|
|
}, timeoutSec ?? 300);
|
|
if (settled) return asText(settled);
|
|
return asText({ ...ask, note: `No answer within ${timeoutSec ?? 300}s — re-check with agenthub_ask_list.` });
|
|
});
|
|
|
|
server.tool('agenthub_ask_list',
|
|
'List asks (routed decisions). Architect: your pending decision queue. Filter by status (e.g. pending) and/or recipient.',
|
|
{ to: z.string().optional(), status: z.string().optional() },
|
|
async ({ to, status }) => asText(remote ? await remoteClient.listAsks(serverUrl!, { to, status }) : listAsks(root, { to, status })));
|
|
|
|
server.tool('agenthub_ask_answer',
|
|
'Answer an ask (architect resolves a routed decision within the approve/push gate).',
|
|
{ id: z.string(), text: z.string(), by: z.string().optional() },
|
|
async ({ id, text, by }) => asText(remote ? await remoteClient.answerAsk(serverUrl!, id, text, by) : answerAsk(root, id, text, by)));
|
|
|
|
server.tool('agenthub_ask_escalate',
|
|
'Escalate an ask to the CEO (architect: REQUIRED for release/publish/push, OSS decisions and architecture pivots). Closes the loop through the same ask — no second blocking channel.',
|
|
{ id: z.string(), note: z.string().optional() },
|
|
async ({ id, note }) => asText(remote ? await remoteClient.escalateAsk(serverUrl!, id, note) : escalateAsk(root, id, note)));
|
|
|
|
const transport = new StdioServerTransport();
|
|
await server.connect(transport);
|
|
// stdio servers must not write to stdout (it's the protocol channel); log to stderr.
|
|
process.stderr.write(`AgentHub MCP server ready (${remote ? `hub ${serverUrl}` : `local ${root}`}).\n`);
|
|
}
|