diff --git a/package.json b/package.json index 65c177e..212491f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "agenthub", - "version": "0.10.0", + "version": "0.10.1", "description": "Local coordination layer for AI coding agents", "type": "module", "main": "./dist/index.js", diff --git a/src/cli/commands/watch.ts b/src/cli/commands/watch.ts index 3ed5ea0..c3347e7 100644 --- a/src/cli/commands/watch.ts +++ b/src/cli/commands/watch.ts @@ -24,21 +24,26 @@ export type { AgentHubEvent } from '../../server/events.js'; * Parse all complete SSE events from a text buffer. * * SSE wire format: "data: \n\n" per event, ": \n\n" for keepalives. - * This function splits on double-newline boundaries, extracts the `data:` - * line from each complete block, and returns anything that didn't end with - * "\n\n" as `remaining` (to be prepended to the next chunk). + * Frame separators are CRLF-tolerant: both "\n\n" and "\r\n\r\n" (and mixed + * line endings inside a frame) are accepted. An event with multiple `data:` + * lines is joined with "\n" before parsing, per the SSE spec. + * Anything that didn't end with a blank line is returned as `remaining` + * (to be prepended to the next chunk). */ export function parseSSEBuffer(buffer: string): { events: AgentHubEvent[]; remaining: string } { - const parts = buffer.split('\n\n'); + const parts = buffer.split(/\r?\n\r?\n/); const remaining = parts.pop() ?? ''; // last segment may be incomplete const events: AgentHubEvent[] = []; for (const part of parts) { // A keepalive block looks like ":" — no data line. - const dataLine = part.split('\n').find((l) => l.startsWith('data: ')); - if (!dataLine) continue; + const dataLines = part + .split(/\r?\n/) + .filter((l) => l.startsWith('data: ')) + .map((l) => l.slice(6)); + if (dataLines.length === 0) continue; try { - events.push(JSON.parse(dataLine.slice(6)) as AgentHubEvent); + events.push(JSON.parse(dataLines.join('\n')) as AgentHubEvent); } catch { // Ignore malformed JSON — should never happen in practice. } diff --git a/src/cli/commands/work.ts b/src/cli/commands/work.ts index b1d3c21..9b72038 100644 --- a/src/cli/commands/work.ts +++ b/src/cli/commands/work.ts @@ -19,6 +19,8 @@ interface WorkAgentContext extends AgentContext { timeoutSec?: number; discoverServer?: (timeoutMs?: number) => Promise; reconnectBackoffMs?: number[]; + /** Polling-fallback interval while the SSE wait is open (default 4000 ms). */ + pollIntervalMs?: number; /** * Unattended mode (TSK-0118): the agent runs without a human at the keyboard. * It must never pause for human input — when it needs a decision it routes an @@ -28,11 +30,17 @@ interface WorkAgentContext extends AgentContext { } /** - * Fetch + print + mark-read the agent's unread messages. Returns how many were - * surfaced. This is what lets the work loop wake on an architect follow-up / - * question (TSK-0119): after a `task review` submit the implementer re-arms - * `work` and stays reachable — a reopen or a new assignment wakes it via a task - * event, and a plain message wakes it here instead of leaving it dormant. + * Fetch + print the agent's unread messages. Returns how many were surfaced. + * This is what lets the work loop wake on an architect follow-up / question + * (TSK-0119): after a `task review` submit the implementer re-arms `work` and + * stays reachable — a reopen or a new assignment wakes it via a task event, + * and a plain message wakes it here instead of leaving it dormant. + * + * Surfacing flips each message `unread` → `delivered` (the listInbox read + * receipt), but deliberately does NOT mark it `read`: the message stays + * visible in the inbox until the agent explicitly acks/reads it, so a + * surfaced-but-missed message is never lost. The loop wakes only on `unread`, + * so a delivered message never re-wakes it (no spin). */ async function drainAgentMessages(ctx: WorkAgentContext): Promise { if (!ctx.serverUrl) return 0; @@ -46,11 +54,6 @@ async function drainAgentMessages(ctx: WorkAgentContext): Promise { console.log(`AgentHub: ${msgs.length} message${msgs.length === 1 ? '' : 's'} for ${ctx.agent}:`); for (const m of msgs) { console.log(` ${m.id} ${m.from} → ${m.to}${m.taskId ? ` [${m.taskId}]` : ''}: ${m.text}`); - try { - await remoteClient.markMessageRead(ctx.serverUrl, m.id); - } catch { - /* best-effort */ - } } return msgs.length; } @@ -102,9 +105,11 @@ function waitAndClaim(ctx: WorkAgentContext): Promise { return new Promise((resolve) => { let settled = false; let controller: AbortController | undefined; + let poll: NodeJS.Timeout | undefined; const finish = () => { if (settled) return; settled = true; + if (poll) clearInterval(poll); try { controller?.abort(); } catch { @@ -122,30 +127,61 @@ function waitAndClaim(ctx: WorkAgentContext): Promise { const deadline = ctx.timeoutSec ? Date.now() + ctx.timeoutSec * 1000 : undefined; const backoffs = ctx.reconnectBackoffMs ?? [2000, 5000, 10000]; let reconnectAttempt = 0; + // Guards against overlapping checks (SSE-triggered vs. polling fallback). + let checking = false; // Re-query then claim if a task addressed to us is now open. Returns true // if a task was claimed (so the caller can stop). const tryClaim = async (): Promise => { - ctx.serverUrl = serverUrl; - const f = await findAddressedOpenTask(ctx); - if (!f) return false; - if (timer) clearTimeout(timer); - await claimAndPrintTask(ctx, f.task, f.handoffs); - finish(); - return true; + if (checking) return false; + checking = true; + try { + ctx.serverUrl = serverUrl; + const f = await findAddressedOpenTask(ctx); + if (!f) return false; + if (timer) clearTimeout(timer); + await claimAndPrintTask(ctx, f.task, f.handoffs); + finish(); + return true; + } finally { + checking = false; + } }; // Wake on an architect follow-up message (not just tasks): surface it and // stop, so the implementer never sits dormant on a pending question. const trySurfaceMessages = async (): Promise => { - ctx.serverUrl = serverUrl; - const n = await drainAgentMessages(ctx); - if (n === 0) return false; - if (timer) clearTimeout(timer); - finish(); - return true; + if (checking) return false; + checking = true; + try { + ctx.serverUrl = serverUrl; + const n = await drainAgentMessages(ctx); + if (n === 0) return false; + if (timer) clearTimeout(timer); + finish(); + return true; + } finally { + checking = false; + } }; + // Polling fallback: SSE is instant when it works, but a dropped frame on an + // otherwise-open stream must never mean an infinite sleep — re-check tasks + // and messages every few seconds, so a lost event costs at most one + // poll interval (~4s by default). + poll = setInterval(() => { + if (settled) return; + void (async () => { + try { + if (await tryClaim()) return; + await trySurfaceMessages(); + } catch { + /* best-effort: the SSE path and the next tick remain */ + } + })(); + }, ctx.pollIntervalMs ?? 4000); + poll.unref?.(); + const waitLoop = async () => { while (!settled && remainingMs(deadline) > 0) { controller = new AbortController(); diff --git a/src/cli/index.ts b/src/cli/index.ts index efcee16..4104a7a 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -19,7 +19,7 @@ import { startMcpServer } from '../mcp/server.js'; import { installMcp } from '../mcp/install.js'; import { loadConfig, saveConfig } from '../core/config.js'; import { findProjectRoot } from '../core/paths.js'; -import { discoverServer } from '../discovery.js'; +import { discoverServer, resolveReachableServerUrl } from '../discovery.js'; import { remoteClient, RemoteError } from './remoteClient.js'; interface ResolvedContext { @@ -61,7 +61,25 @@ async function resolveContext(program: Command, cwd: string): Promise Promise): Promise', 'AgentHub server URL (env: AGENTHUB_SERVER)'); program diff --git a/src/core/services/messageService.ts b/src/core/services/messageService.ts index 49bf5df..0c361eb 100644 --- a/src/core/services/messageService.ts +++ b/src/core/services/messageService.ts @@ -78,9 +78,12 @@ export interface InboxMessage { * transitioned to `delivered` (a message the recipient has now been shown) AFTER * filtering and BEFORE returning, so the returned rows reflect the new status. * This is agent-scoped only — the architect-wide `listMessages` never mutates. - * (The `agenthub work` drainInbox path filters `unreadOnly` BEFORE this mutation - * and then marks read, so the delivered intermediate is invisible there — no - * regress and no spin.) + * + * Nothing auto-marks a message `read`: a delivered-but-unacked message stays + * visible in the default inbox view until the recipient explicitly acks/reads + * it (`message ack` / `message read`), so a message surfaced by a background + * work loop is never silently lost. Callers that wake on new mail filter + * `unreadOnly`, so a delivered message never re-wakes them (no spin). */ export function listInbox(cwd: string, agent: string, opts: { unreadOnly?: boolean } = {}): InboxMessage[] { const index = new Index(cwd); diff --git a/src/discovery.ts b/src/discovery.ts index 1e8500c..2661792 100644 --- a/src/discovery.ts +++ b/src/discovery.ts @@ -59,6 +59,45 @@ export function startDiscoveryBroadcaster(getServerUrl: string | (() => string), }; } +/** + * Cheap reachability probe for a configured server URL: GET /status with a + * short timeout. Used to detect a stale saved address (e.g. agenthub.local + * after a network change) before a remote command hard-fails. + */ +export async function probeServer(url: string, timeoutMs = 1000): Promise { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + try { + const res = await fetch(new URL('/status', url).toString(), { signal: controller.signal }); + return res.ok; + } catch { + return false; + } finally { + clearTimeout(timer); + } +} + +/** + * Self-heal a stale configured server URL: probe it, and if it is unreachable + * fall back to LAN discovery. Returns the URL to use for the call — the + * configured one when it answers, otherwise the discovered one, otherwise + * `undefined` (nothing reachable). Probe/discovery are injectable for tests. + */ +export async function resolveReachableServerUrl( + configuredUrl: string, + options: { + probe?: (url: string) => Promise; + discover?: (timeoutMs?: number) => Promise; + discoverTimeoutMs?: number; + } = {}, +): Promise { + const probe = options.probe ?? probeServer; + if (await probe(configuredUrl)) return configuredUrl; + const discover = options.discover ?? discoverServer; + const discovered = await discover(options.discoverTimeoutMs ?? 2000); + return discovered && discovered !== configuredUrl ? discovered : undefined; +} + export function discoverServer(timeoutMs = 3000, port = DISCOVERY_PORT): Promise { return new Promise((resolve) => { const socket = dgram.createSocket('udp4'); diff --git a/src/mcp/server.ts b/src/mcp/server.ts index a4ff193..06e38a5 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -20,7 +20,7 @@ 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, markMessageRead } from '../core/services/messageService.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'; @@ -82,14 +82,36 @@ function waitForTask( 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(); @@ -143,7 +165,7 @@ function waitForTask( export async function startMcpServer(cwd: string): Promise { const { root, serverUrl } = resolveContext(cwd); const remote = !!serverUrl; - const server = new McpServer({ name: 'agenthub', version: '0.8.0' }); + const server = new McpServer({ name: 'agenthub', version: '0.10.1' }); server.tool('agenthub_hello', 'Announce yourself to the hub (presence). Run once when you start.', { agent: z.string(), role: z.string().optional() }, @@ -162,15 +184,14 @@ export async function startMcpServer(cwd: string): Promise { if (nextServerUrl) ctx.serverUrl = nextServerUrl; return ctx.serverUrl!; }; - // Fetch + mark-read the agent's unread messages, so the work loop surfaces - // them once and doesn't spin on the same message. + // 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; - const msgs = remote ? await remoteClient.getInbox(activeUrl!, agent, true) : listInbox(root, agent, { unreadOnly: true }); - for (const m of msgs) { - try { if (remote) await remoteClient.markMessageRead(activeUrl!, m.id); else markMessageRead(root, m.id); } catch { /* best-effort */ } - } - return msgs; + return remote ? await remoteClient.getInbox(activeUrl!, agent, true) : listInbox(root, agent, { unreadOnly: true }); }; const findWork = async (nextServerUrl?: string) => { if (nextServerUrl) useServerUrl(nextServerUrl); diff --git a/tests/realtime-regression.test.ts b/tests/realtime-regression.test.ts new file mode 100644 index 0000000..1b966e7 --- /dev/null +++ b/tests/realtime-regression.test.ts @@ -0,0 +1,210 @@ +/** + * Regression tests for the realtime reliability patch (TSK-0224): + * (a) polling fallback claims a task even when its SSE event is lost; + * (b) parseSSEBuffer accepts CRLF frames and joins multiple data: lines; + * (c) delivered-but-unacked messages stay visible in the inbox; + * (d) discovery self-heal falls back when the configured URL is dead. + */ +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 { parseSSEBuffer } from '../src/cli/commands/watch.js'; +import { startServer } from '../src/server/index.js'; +import { createMessage, listInbox, getMessage } from '../src/core/services/messageService.js'; +import { probeServer, resolveReachableServerUrl } from '../src/discovery.js'; +import type { Task } from '../src/core/schema.js'; + +// ─── (a) polling fallback: claim despite a lost SSE event ──────────────────── + +describe('work — polling fallback (lost SSE event)', () => { + let cwd: string; + let server: Awaited>; + const realFetch = globalThis.fetch; + + beforeEach(async () => { + cwd = mkdtempSync(join(tmpdir(), 'ah-poll-')); + init(cwd, { projectName: 'poll-test', yes: true }); + server = await startServer(cwd, { host: '127.0.0.1', port: 0 }); + + // Suppress the SSE channel: /events returns an open stream that never + // delivers a frame. Every other request passes through to the real server. + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = typeof input === 'string' ? input : input instanceof URL ? input.toString() : input.url; + if (url.endsWith('/events')) { + const stream = new ReadableStream({ + start(controller) { + init?.signal?.addEventListener('abort', () => { + try { + controller.close(); + } catch { + /* already closed */ + } + }); + // Never enqueue: the SSE frame is "lost". + }, + }); + return new Response(stream, { status: 200, headers: { 'Content-Type': 'text/event-stream' } }); + } + return realFetch(input, init); + }) as typeof fetch; + }); + + afterEach(async () => { + globalThis.fetch = realFetch; + await server.app.close().catch(() => undefined); + rmSync(cwd, { recursive: true, force: true }); + }); + + it('claims a task via the polling fallback when its SSE event never arrives', async () => { + const workDone = workAgent({ + serverUrl: server.url, + projectCwd: cwd, + agent: 'kimi', + role: 'implementer', + timeoutSec: 8, + pollIntervalMs: 100, + }); + + // Delegate a task after the wait started. The SSE event is swallowed by the + // stub above, so only the polling fallback can notice it. + await new Promise((r) => setTimeout(r, 300)); + await realFetch(`${server.url}/tasks`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ title: 'kimi: claimed via polling', role: 'implementer' }), + }); + + await workDone; // must resolve via the poll, long before the 8s timeout + + const tasks = (await realFetch(`${server.url}/tasks`).then((r) => r.json())) as Task[]; + const mine = tasks.find((t) => (t.title ?? '').startsWith('kimi:')); + expect(mine).toBeDefined(); + expect(mine?.status).toBe('in_progress'); + expect(mine?.assignedTo).toBe('kimi'); + }, 10000); +}); + +// ─── (b) CRLF-tolerant SSE parser ──────────────────────────────────────────── + +describe('parseSSEBuffer — CRLF + multi-line data', () => { + it('parses CRLF-framed events', () => { + const buf = 'data: {"type":"task","action":"created","id":"TSK-0001"}\r\n\r\n'; + const { events, remaining } = parseSSEBuffer(buf); + expect(events).toHaveLength(1); + expect(events[0].id).toBe('TSK-0001'); + expect(remaining).toBe(''); + }); + + it('parses mixed LF/CRLF frames and CRLF keepalives', () => { + const buf = + ':\r\n\r\n' + + 'data: {"type":"memory","action":"created","id":"MEM-001"}\n\n' + + 'data: {"type":"decision","action":"created","id":"DEC-001"}\r\n\r\n'; + const { events, remaining } = parseSSEBuffer(buf); + expect(events).toHaveLength(2); + expect(events[0].id).toBe('MEM-001'); + expect(events[1].id).toBe('DEC-001'); + expect(remaining).toBe(''); + }); + + it('joins multiple data: lines of one event with a newline', () => { + const buf = 'data: {"type":"task",\r\ndata: "id":"TSK-0002"}\r\n\r\n'; + const { events } = parseSSEBuffer(buf); + expect(events).toHaveLength(1); + expect(events[0].type).toBe('task'); + expect(events[0].id).toBe('TSK-0002'); + }); + + it('keeps an incomplete CRLF tail in `remaining`', () => { + const buf = 'data: {"type":"task","id":"TSK-0001"}\r\n\r\ndata: {"type":"deci'; + const { events, remaining } = parseSSEBuffer(buf); + expect(events).toHaveLength(1); + expect(remaining).toBe('data: {"type":"deci'); + }); +}); + +// ─── (c) delivered-unacked messages stay visible ───────────────────────────── + +describe('message semantics — delivered stays visible until explicit ack/read', () => { + let cwd: string; + + beforeEach(() => { + cwd = mkdtempSync(join(tmpdir(), 'ah-msg-vis-')); + init(cwd, { projectName: 'msg-vis', yes: true }); + }); + + afterEach(() => { + rmSync(cwd, { recursive: true, force: true }); + }); + + it('a surfaced (delivered) message stays in the default inbox and never re-wakes', () => { + const msg = createMessage(cwd, { from: 'claude', to: 'kimi', text: 'look at this' }); + + // The work-loop drain path: fetch unread. First call surfaces it and flips + // it to delivered; nothing marks it read. + const first = listInbox(cwd, 'kimi', { unreadOnly: true }); + expect(first).toHaveLength(1); + expect(first[0].status).toBe('delivered'); + expect(getMessage(cwd, msg.id).message.status).toBe('delivered'); + + // No spin: the unread-filtered wake check is empty on the next pass… + expect(listInbox(cwd, 'kimi', { unreadOnly: true })).toHaveLength(0); + + // …but the info stays visible in the default inbox until explicit ack/read. + const inbox = listInbox(cwd, 'kimi'); + expect(inbox.map((m) => m.id)).toContain(msg.id); + expect(inbox.find((m) => m.id === msg.id)?.status).toBe('delivered'); + }); +}); + +// ─── (d) discovery self-heal ───────────────────────────────────────────────── + +describe('discovery self-heal — stale configured URL', () => { + let cwd: string; + let server: Awaited>; + + beforeEach(async () => { + cwd = mkdtempSync(join(tmpdir(), 'ah-heal-')); + init(cwd, { projectName: 'heal-test', yes: true }); + server = await startServer(cwd, { host: '127.0.0.1', port: 0 }); + }); + + afterEach(async () => { + await server.app.close().catch(() => undefined); + rmSync(cwd, { recursive: true, force: true }); + }); + + it('falls back to the discovered URL when the configured one is dead', async () => { + const discovered = await resolveReachableServerUrl('http://127.0.0.1:1', { + discover: async () => server.url, + }); + expect(discovered).toBe(server.url); + }); + + it('keeps the configured URL when it answers (discovery not consulted)', async () => { + let discoverCalled = false; + const resolved = await resolveReachableServerUrl(server.url, { + discover: async () => { + discoverCalled = true; + return 'http://example.invalid:9'; + }, + }); + expect(resolved).toBe(server.url); + expect(discoverCalled).toBe(false); + }); + + it('returns undefined when the configured URL is dead and nothing is discovered', async () => { + const resolved = await resolveReachableServerUrl('http://127.0.0.1:1', { + discover: async () => undefined, + }); + expect(resolved).toBeUndefined(); + }); + + it('probeServer distinguishes a live server from a dead one', async () => { + expect(await probeServer(server.url)).toBe(true); + expect(await probeServer('http://127.0.0.1:1', 300)).toBe(false); + }); +}); diff --git a/tests/work.test.ts b/tests/work.test.ts index 75ab423..efb7a64 100644 --- a/tests/work.test.ts +++ b/tests/work.test.ts @@ -131,12 +131,14 @@ describe('agenthub work — wait then auto-claim (server SSE)', () => { await workDone; // resolves because the message woke the loop (not on timeout) - // The loop drained + marked the message read — proving it woke on the message, - // not that it merely timed out (a timeout would leave it delivered/unread). + // The loop drained + surfaced the message (unread → delivered) — proving it + // woke on the message, not that it merely timed out (a timeout would leave + // it unread). Since TSK-0224 the loop no longer auto-marks read: the message + // stays visible in the inbox until an explicit ack/read. const inbox = (await fetch(`${server.url}/messages?agent=kimi`).then((r) => r.json())) as Array<{ text: string; status: string }>; const m = inbox.find((x) => x.text.includes('quick question')); expect(m).toBeDefined(); - expect(m?.status).toBe('read'); + expect(m?.status).toBe('delivered'); }, 7000); it('wakes and re-claims when a submitted task is reopened', async () => {