From e8c2e43cb02532f657ab212798ba99eb811372ab Mon Sep 17 00:00:00 2001 From: chahinebrini Date: Wed, 29 Jul 2026 21:08:21 +0200 Subject: [PATCH] =?UTF-8?q?feat(mcp):=20echter=20Push=20an=20Agenten=20?= =?UTF-8?q?=C3=BCber=20MCP-Notifications=20=E2=80=94=20ohne=20blockierende?= =?UTF-8?q?n=20Tool-Call?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DIE WURZEL, endlich an der Wurzel gefasst. 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 — ein Mensch musste ihn anstoßen. Wir haben einen ganzen Tag um diese Tatsache herum gebaut (Watchdog, Check-in-Kanal, längere Timeouts, notListening) statt sie aufzulösen. Der CEO hat zu Recht darauf bestanden, das Protokoll selbst zu nutzen. MCP kann Server→Client jederzeit `notifications/message` schicken. Die Bridge ist ohnehin ein langlebiger Prozess pro Agent: sie hängt sich jetzt dauerhaft an den SSE-Stream des Hubs (`pushChannel.ts`) und reicht relevante Ereignisse als Notification hoch — neue/zurückgegebene Task, Nachricht, beantwortete Frage, Abbruch. Bindung erfolgt, sobald der Agent seinen Namen nennt (hello/work), inklusive Alias-Auflösung; Reconnect mit Backoff. ⚠️ Die entscheidende Falle: `sendLoggingMessage` prüft `_capabilities.logging` und verwirft die Notification sonst STILL. Der Server deklarierte gar keine Capabilities — daran wäre dieser Weg unbemerkt gescheitert. Jetzt deklariert. Live verifiziert mit einem echten MCP-Client: Nachricht am Hub angelegt → Notification kam beim Client an, ohne dass dieser in einem Tool-Call wartete. OFFEN und bewusst nicht behauptet: ob die CLI-Hosts (Codex/Kimi) die Notification dem Modell zeigen — das ist client-abhängig und empirisch zu prüfen. Deshalb bleiben Check-in-Kanal und Work-Loop als Netz bestehen. Co-Authored-By: Claude Opus 5 (1M context) --- src/mcp/pushChannel.ts | 122 +++++++++++++++++++++++++++++++++++++++++ src/mcp/server.ts | 16 +++++- 2 files changed, 137 insertions(+), 1 deletion(-) create mode 100644 src/mcp/pushChannel.ts 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) => {