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

fix(realtime): discovery self-heal, polling fallback, message ack semantics, CRLF SSE parser

Browse Source 
- probe configured server URL (~1s) before remote commands; on failure
  re-discover via LAN discovery, persist and use the resolved URL
- poll tryClaim/trySurfaceMessages every ~4s during the SSE wait in CLI
  work and MCP waitForTask — a lost SSE frame costs seconds, never sleep
- work/MCP no longer auto-mark messages read: surfacing sets delivered,
  explicit ack/read required; loops still wake only on unread (no spin)
- parseSSEBuffer accepts CRLF frame separators and joins multi data: lines
- bump version to 0.10.1 (package.json, CLI --version, MCP server)
This commit is contained in:
chahinebrini 2026-07-23 16:53:33 +02:00
parent 52e5d0bf28
commit 9551c90597
9 changed files with 383 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
package.json
View File

@ -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",

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

@ -24,21 +24,26 @@ export type { AgentHubEvent } from '../../server/events.js';
* Parse all complete SSE events from a text buffer.
*
* SSE wire format: "data: <json>\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.
}

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

@ -19,6 +19,8 @@ interface WorkAgentContext extends AgentContext {
timeoutSec?: number;
discoverServer?: (timeoutMs?: number) => Promise<string | undefined>;
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<number> {
if (!ctx.serverUrl) return 0;
@ -46,11 +54,6 @@ async function drainAgentMessages(ctx: WorkAgentContext): Promise<number> {
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<void> {
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<void> {
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<boolean> => {
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<boolean> => {
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();

24
src/cli/index.ts
View File

@ -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<ResolvedCo
} catch {
return { projectCwd: root };
}
if (config.serverUrl) return { serverUrl: config.serverUrl, projectCwd: root };
if (config.serverUrl) {
// Self-heal a stale saved URL (e.g. agenthub.local after a network
// change): probe it with a short timeout; if it's dead, re-discover the
// live server, persist it and use it for this call — no hard failure.
const reachable = await resolveReachableServerUrl(config.serverUrl);
if (reachable === config.serverUrl) return { serverUrl: config.serverUrl, projectCwd: root };
if (reachable) {
try {
saveConfig(root, { ...config, serverUrl: reachable });
console.error(`AgentHub: ${config.serverUrl} unreachable — switched to discovered server ${reachable} (config updated).`);
} catch {
console.error(`AgentHub: ${config.serverUrl} unreachable — using discovered server ${reachable} for this call.`);
}
return { serverUrl: reachable, projectCwd: root };
}
// Nothing discovered: keep the configured URL so runRemote reports its
// usual friendly "not reachable" error (and re-attempts discovery).
return { serverUrl: config.serverUrl, projectCwd: root };
}
// Initialized project but no server configured yet: auto-find one on the
// LAN and remember it, so the agent connects with zero manual setup.
@ -120,7 +138,7 @@ async function runRemote(serverUrl: string, fn: () => Promise<void>): Promise<vo
export function createProgram(cwd: string): Command {
const program = new Command('agenthub')
.description('Local coordination layer for AI coding agents')
.version('0.9.1')
.version('0.10.1')
.option('--server <url>', 'AgentHub server URL (env: AGENTHUB_SERVER)');
program

9
src/core/services/messageService.ts
View File

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

39
src/discovery.ts
View File

@ -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<boolean> {
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<boolean>;
discover?: (timeoutMs?: number) => Promise<string | undefined>;
discoverTimeoutMs?: number;
} = {},
): Promise<string | undefined> {
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<string | undefined> {
return new Promise((resolve) => {
const socket = dgram.createSocket('udp4');

39
src/mcp/server.ts
View File

@ -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<T>(
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<T>(
export async function startMcpServer(cwd: string): Promise<void> {
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<void> {
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);

210
tests/realtime-regression.test.ts Normal file
View File

@ -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<ReturnType<typeof startServer>>;
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<Uint8Array>({
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<ReturnType<typeof startServer>>;
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);
});
});

8
tests/work.test.ts
View File

@ -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 () => {