diff --git a/src/mcp/pushChannel.ts b/src/mcp/pushChannel.ts new file mode 100644 index 0000000..0349b95 --- /dev/null +++ b/src/mcp/pushChannel.ts @@ -0,0 +1,122 @@ +import type { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'; +import { parseSSEBuffer } from '../cli/commands/watch.js'; +import { remoteClient } from '../cli/remoteClient.js'; + +/** + * Echter Push an den Agenten über MCP — ohne blockierenden Tool-Call. + * + * WARUM DAS HIER STEHT (die Lehre aus einem ganzen Tag): + * Bisher konnte ein Agent nur empfangen, während er in `agenthub_work` + * blockierte. Kehrte der Aufruf leer zurück und das Modell startete ihn nicht + * neu, war der Agent unerreichbar — kein Reopen, keine Nachricht kam an, und + * ein Mensch musste ihn anstoßen. Wir haben um diese Tatsache herum gebaut + * (Watchdog, Check-in-Kanal, längere Timeouts) statt sie aufzulösen. + * + * MCP kann das von Haus aus: der Server darf dem Client JEDERZEIT + * `notifications/message` schicken. Unsere Bridge ist ohnehin ein langlebiger + * Prozess pro Agent — sie hängt sich dauerhaft an den SSE-Stream des Hubs und + * reicht relevante Ereignisse als MCP-Notification hoch. Kein Blockieren, + * kein Timeout, keine Loop-Disziplin nötig. + * + * ⚠️ VORAUSSETZUNG: Der Server MUSS die `logging`-Capability deklarieren — + * `sendLoggingMessage` prüft `_capabilities.logging` und verwirft die + * Notification sonst STILL. Genau daran wäre dieser Weg unbemerkt gescheitert. + * + * ⚠️ OFFEN (client-abhängig, nicht von uns entscheidbar): ob der jeweilige + * CLI-Host die Notification dem Modell zeigt. Das ist empirisch zu prüfen und + * darf nicht behauptet werden. Deshalb bleibt der Check-in-Kanal als Netz + * bestehen — dieser Push ist die Verbesserung, nicht der alleinige Verlass. + */ + +/** Reconnect-Backoff, wenn der SSE-Stream abreißt (Hub-Neustart o. Ä.). */ +const RECONNECT_DELAYS_MS = [1000, 3000, 8000]; + +interface PushState { + agent: string; + names: Set; + stop: () => void; +} + +let active: PushState | undefined; + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** Betrifft dieses Event den Agenten, für den wir pushen? */ +function relevantFor( + event: { type?: string; action?: string; assignedTo?: string; status?: string }, + names: Set, +): boolean { + const target = String(event.assignedTo ?? '').toLowerCase(); + if (target && names.has(target)) return true; + // Ohne Adressat ist ein Task-Event nur interessant, wenn es ein Reopen sein + // könnte — der Claim wird dabei abgeräumt, der Adressat bleibt aber stehen, + // also greift oben bereits `assignedTo`. Alles andere ignorieren wir bewusst: + // Rauschen entwertet den Kanal (Lehre aus dem Watchdog-Spam). + return false; +} + +function describe(event: { type?: string; action?: string; id?: string; status?: string }): string { + const id = event.id ?? '?'; + if (event.type === 'message') return `Neue Nachricht ${id} für dich — agenthub_inbox lesen.`; + if (event.type === 'ask') return `Deine Frage ${id} wurde beantwortet — agenthub_ask_list.`; + if (event.status === 'open') return `${id} wartet auf dich (neu oder zurückgegeben) — agenthub_work aufrufen.`; + if (event.status === 'cancelled') return `${id} wurde abgebrochen — STOPP, nicht weiterbauen.`; + return `${id}: ${event.type}/${event.action} (${event.status ?? '—'}).`; +} + +/** + * Bindet die Bridge an einen Agenten und beginnt zu pushen. Idempotent: + * derselbe Agent startet den Stream nicht doppelt, ein anderer löst den alten ab. + */ +export function bindPushChannel(server: McpServer, serverUrl: string, agent: string): void { + const name = agent.trim().toLowerCase(); + if (!name || active?.agent === name) return; + active?.stop(); + + let stopped = false; + const state: PushState = { agent: name, names: new Set([name]), stop: () => { stopped = true; } }; + active = state; + + void (async () => { + // Aliase mitnehmen, damit ein umbenannter Agent seine Events weiter bekommt. + try { + const identity = await remoteClient.getAgentIdentity(serverUrl, agent); + for (const alias of identity.names) state.names.add(alias.toLowerCase()); + } catch { /* Namensvergleich reicht */ } + + let attempt = 0; + while (!stopped) { + try { + const res = await fetch(new URL('/events', serverUrl).toString(), { + headers: { Accept: 'text/event-stream' }, + }); + if (!res.body) throw new Error('SSE ohne Body'); + attempt = 0; + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + while (!stopped) { + const { done, value } = await reader.read(); + if (done) break; + if (value) buffer += decoder.decode(value, { stream: true }); + const { events, remaining } = parseSSEBuffer(buffer); + buffer = remaining; + for (const event of events) { + if (!relevantFor(event, state.names)) continue; + try { + await server.server.sendLoggingMessage({ + level: 'info', + logger: 'agenthub', + data: { agenthub: describe(event), event }, + }); + } catch { /* Client mag Notifications nicht — kein Grund abzubrechen */ } + } + } + } catch { /* Hub weg oder Neustart — unten neu versuchen */ } + if (stopped) return; + await sleep(RECONNECT_DELAYS_MS[Math.min(attempt++, RECONNECT_DELAYS_MS.length - 1)]!); + } + })(); +} diff --git a/src/mcp/server.ts b/src/mcp/server.ts index 7f54841..c8d8fa8 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -19,6 +19,7 @@ import { import { createHandoff, getHandoff } from '../core/services/handoffService.js'; import { appendTaskLog } from '../core/services/taskLogService.js'; import { resolvePending, hasPending } from '../core/services/checkinService.js'; +import { bindPushChannel } from './pushChannel.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'; @@ -200,12 +201,24 @@ export function waitForTask( export async function startMcpServer(cwd: string): Promise { const { root, serverUrl } = await resolveMcpContext(cwd); const remote = !!serverUrl; - const server = new McpServer({ name: 'agenthub', version: VERSION }); + // `logging` MUSS deklariert werden: sendLoggingMessage prueft + // _capabilities.logging und verwirft die Notification sonst kommentarlos. + // Ohne diese Zeile gaebe es keinen Push — und man saehe nicht, warum. + const server = new McpServer( + { name: 'agenthub', version: VERSION }, + { capabilities: { logging: {} } }, + ); + + /** Bridge an den Agenten binden, sobald er sich zu erkennen gibt. */ + const bindPush = (agent?: string) => { + if (remote && agent) bindPushChannel(server, serverUrl!, agent); + }; 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'); + bindPush(agent); return asText(`AgentHub: ${agent} joined (${role ?? 'implementer'})`); }); @@ -213,6 +226,7 @@ export async function startMcpServer(cwd: string): Promise { 'Block until there is work for you, then return it. The default 50s timeout stays below known MCP client limits; SSE still wakes immediately, so this only causes more empty wake-ups and does not add delivery latency. 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 }) => { + bindPush(agent); const ctx: AgentContext = { serverUrl, projectCwd: root, agent, role: role ?? 'implementer' }; if (remote) await remoteClient.setLoop(serverUrl!, agent, true); const leave = async (reason: string) => {