diff --git a/src/mcp-server.mjs b/src/mcp-server.mjs index 208680d..8ee1b45 100644 --- a/src/mcp-server.mjs +++ b/src/mcp-server.mjs @@ -20,7 +20,7 @@ // Transport: stdio. Compatible with Claude Desktop / Code MCP client config. import { execSync } from 'node:child_process'; -import { readFileSync } from 'node:fs'; +import { readFileSync, writeFileSync, renameSync, statSync } from 'node:fs'; import { fileURLToPath } from 'node:url'; import { dirname, join } from 'node:path'; import { Server } from '@modelcontextprotocol/sdk/server/index.js'; @@ -214,6 +214,95 @@ export function captureTmuxPane(session, lines = 50) { } } +// --- notification ack (consumed-only) --------------------------------------- +// +// room_ack used to blank the whole notification file. Anything the poller +// appended between the agent's last room_list_new read and the ack was wiped +// unread (real incident 2026-07-08: an owner instruction sat 5 hours unseen). +// The fix: room_list_new remembers the exact bytes it returned; room_ack +// removes ONLY those bytes and preserves anything appended since. + +function countNotificationLines(raw) { + if (!raw) return 0; + return raw.split('\n').filter((l) => l.trim().length > 0).length; +} + +export function removeConsumedNotifications(currentRaw, consumedRaw) { + // Compute what should remain in the notification file after acking exactly + // the content a prior room_list_new returned. + // + // Fast path: the pollers only ever append, so the consumed content is + // normally still a byte-prefix of the current file — drop that prefix. + // Fallback: if the file was rewritten in between (manual edit, rotation), + // drop consumed lines by exact match (multiset semantics) and keep the rest. + const current = currentRaw || ''; + const consumed = consumedRaw || ''; + if (!consumed.trim()) { + // Last read saw an empty file — nothing was consumed, nothing to remove. + return { remainder: current, consumedLines: 0, mode: 'noop' }; + } + if (current.startsWith(consumed)) { + return { + remainder: current.slice(consumed.length), + consumedLines: countNotificationLines(consumed), + mode: 'prefix', + }; + } + const pending = new Map(); + for (const line of consumed.split('\n')) { + if (!line.trim()) continue; + pending.set(line, (pending.get(line) || 0) + 1); + } + const kept = []; + let removed = 0; + for (const line of current.split('\n')) { + if (line.trim() && (pending.get(line) || 0) > 0) { + pending.set(line, pending.get(line) - 1); + removed += 1; + continue; + } + kept.push(line); + } + return { remainder: kept.join('\n'), consumedLines: removed, mode: 'lines' }; +} + +function atomicWriteNotify(notifyFile, content) { + // Temp-file + rename in the same directory so readers never observe a + // half-written file. The pollers re-open the path on every appendFileSync, + // so appends after the rename land in the new file; the read→rename window + // inside a single ack is microseconds (vs the old read→think→truncate + // window of minutes). tail -F watchers see the rename as a rotation and + // re-open; the IAK Monitor/hook patterns re-read the whole file anyway. + let mode; + try { + mode = statSync(notifyFile).mode & 0o777; + } catch { + // File missing — default mode is fine. + } + const tmp = `${notifyFile}.ack-${process.pid}-${Date.now()}.tmp`; + writeFileSync(tmp, content, mode != null ? { mode } : undefined); + renameSync(tmp, notifyFile); +} + +export function ackNotificationFile(notifyFile, consumedRaw) { + // consumedRaw === null → no room_list_new recorded this session: legacy + // clear-everything behavior (kept for cold callers, inherently racy). + // Otherwise remove only the consumed content; late arrivals survive. + let current = ''; + try { + current = readFileSync(notifyFile, 'utf8'); + } catch { + current = ''; // missing file == already empty + } + if (consumedRaw == null) { + if (current !== '') atomicWriteNotify(notifyFile, ''); + return { mode: 'all', consumedLines: countNotificationLines(current), preservedLines: 0 }; + } + const { remainder, consumedLines, mode } = removeConsumedNotifications(current, consumedRaw); + if (remainder !== current) atomicWriteNotify(notifyFile, remainder); + return { mode, consumedLines, preservedLines: countNotificationLines(remainder) }; +} + function ok(text) { return { content: [{ type: 'text', text }] }; } @@ -326,6 +415,10 @@ export async function runMcpServer({ configPath } = {}) { ); } + // Per-session memory of what the last room_list_new returned, keyed by + // notification file path. room_ack uses it to clear only consumed lines. + const lastRoomListNew = new Map(); + const tools = [ { name: 'room_list_new', @@ -334,7 +427,9 @@ export async function runMcpServer({ configPath } = {}) { }, { name: 'room_ack', - description: 'Clear the notification file, acknowledging that messages have been read.', + description: + 'Acknowledge the messages returned by the last room_list_new. Only those lines are ' + + 'removed from the notification file; anything appended since that read is preserved.', inputSchema: { type: 'object', properties: {} }, }, { @@ -520,20 +615,39 @@ export async function runMcpServer({ configPath } = {}) { switch (name) { case 'room_list_new': { const notifyFile = config?.poller?.notification_file || '/tmp/iak-new-messages.txt'; + let raw = ''; try { - const content = readFileSync(notifyFile, 'utf8').trim(); - if (!content) return ok('No new messages.'); - return ok(content); + raw = readFileSync(notifyFile, 'utf8'); } catch { - return ok('No new messages.'); + raw = ''; } + // Remember exactly what this read returned so room_ack clears only + // these bytes — lines the poller appends after this point survive. + lastRoomListNew.set(notifyFile, raw); + const content = raw.trim(); + if (!content) return ok('No new messages.'); + return ok(content); } case 'room_ack': { const notifyFile = config?.poller?.notification_file || '/tmp/iak-new-messages.txt'; try { - const { writeFileSync } = await import('node:fs'); - writeFileSync(notifyFile, ''); - return ok('Acknowledged new messages.'); + const consumedRaw = lastRoomListNew.has(notifyFile) ? lastRoomListNew.get(notifyFile) : null; + const result = ackNotificationFile(notifyFile, consumedRaw); + lastRoomListNew.delete(notifyFile); + if (result.mode === 'all') { + return ok( + 'Acknowledged new messages (no room_list_new recorded this session — cleared the whole file; ' + + 'prefer room_list_new → room_ack so late arrivals are preserved).' + ); + } + if (result.preservedLines > 0) { + return ok( + `Acknowledged ${result.consumedLines} read message line(s); ` + + `${result.preservedLines} line(s) arrived after the last room_list_new and were preserved — ` + + 'call room_list_new again to see them.' + ); + } + return ok(`Acknowledged ${result.consumedLines} read message line(s).`); } catch (e) { return err('Failed to ack: ' + e.message); } diff --git a/test/mcp-server.test.mjs b/test/mcp-server.test.mjs index e948290..c3fc3fa 100644 --- a/test/mcp-server.test.mjs +++ b/test/mcp-server.test.mjs @@ -17,6 +17,8 @@ import { decideTmuxRunMode, configuredAgentSessions, roomApiConfigured, + removeConsumedNotifications, + ackNotificationFile, } from '../src/mcp-server.mjs'; const __dirname = dirname(fileURLToPath(import.meta.url)); @@ -148,7 +150,7 @@ test('roomApiConfigured: requires both API key and room', () => { // --- end-to-end stdio smoke ------------------------------------------------ -import { writeFileSync, mkdtempSync, rmSync } from 'node:fs'; +import { writeFileSync, appendFileSync, readFileSync, mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; function rpc(line) { return JSON.stringify(line) + '\n'; } @@ -226,3 +228,164 @@ test('iak-mcp.mjs with room API config exposes low-latency room tools', async () rmSync(dir, { recursive: true, force: true }); } }); + +// --- removeConsumedNotifications (pure) -------------------------------------- + +test('removeConsumedNotifications: consumed prefix is dropped, later append survives', () => { + const consumed = '[room] alice: first\n[room] bob: second\n'; + const current = consumed + '[room] petrus: arrived after the read\n'; + const r = removeConsumedNotifications(current, consumed); + assert.equal(r.remainder, '[room] petrus: arrived after the read\n'); + assert.equal(r.consumedLines, 2); + assert.equal(r.mode, 'prefix'); +}); + +test('removeConsumedNotifications: nothing appended → file drains to empty', () => { + const consumed = '[room] alice: first\n'; + const r = removeConsumedNotifications(consumed, consumed); + assert.equal(r.remainder, ''); + assert.equal(r.consumedLines, 1); +}); + +test('removeConsumedNotifications: empty read consumes nothing (late lines survive)', () => { + const r = removeConsumedNotifications('[room] petrus: late\n', ''); + assert.equal(r.remainder, '[room] petrus: late\n'); + assert.equal(r.consumedLines, 0); + assert.equal(r.mode, 'noop'); +}); + +test('removeConsumedNotifications: rewritten file falls back to per-line removal', () => { + // File no longer starts with what was read (e.g. rotated/edited in between): + // remove read lines by exact match, keep everything else. + const consumed = '[room] alice: first\n[room] bob: second\n'; + const current = '[room] petrus: new head\n[room] bob: second\n[room] alice: first\n[room] petrus: tail\n'; + const r = removeConsumedNotifications(current, consumed); + assert.equal(r.remainder, '[room] petrus: new head\n[room] petrus: tail\n'); + assert.equal(r.consumedLines, 2); + assert.equal(r.mode, 'lines'); +}); + +test('removeConsumedNotifications: duplicate lines removed with multiset semantics', () => { + const consumed = '[room] bot: ping\n'; + const current = '[room] bot: other\n[room] bot: ping\n[room] bot: ping\n'; + const r = removeConsumedNotifications(current, consumed); + // Only ONE copy of the read line is removed; the second (unread) survives. + assert.equal(r.remainder, '[room] bot: other\n[room] bot: ping\n'); + assert.equal(r.consumedLines, 1); +}); + +// --- ackNotificationFile (file-level) ---------------------------------------- + +test('ackNotificationFile: REGRESSION — poller append between read and ack survives', () => { + const dir = mkdtempSync(join(tmpdir(), 'iak-ack-test-')); + const notifyFile = join(dir, 'new-messages.txt'); + try { + // Agent reads (room_list_new captures this content)... + const consumed = '[room] alice: task A\n[room] bob: task B\n'; + writeFileSync(notifyFile, consumed); + // ...poller races in and appends BEFORE the agent acks (the 2026-07-08 + // incident shape: an owner instruction landed in this window)... + appendFileSync(notifyFile, '[room] petrus: send all 3\n'); + // ...agent acks what it read. + const r = ackNotificationFile(notifyFile, consumed); + assert.equal(r.consumedLines, 2); + assert.equal(r.preservedLines, 1); + // The late line MUST still be in the file, unread but not lost. + assert.equal(readFileSync(notifyFile, 'utf8'), '[room] petrus: send all 3\n'); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('ackNotificationFile: no prior read (null) keeps legacy clear-all contract', () => { + const dir = mkdtempSync(join(tmpdir(), 'iak-ack-test-')); + const notifyFile = join(dir, 'new-messages.txt'); + try { + writeFileSync(notifyFile, '[room] a: x\n[room] b: y\n'); + const r = ackNotificationFile(notifyFile, null); + assert.equal(r.mode, 'all'); + assert.equal(readFileSync(notifyFile, 'utf8'), ''); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('ackNotificationFile: missing notification file is a no-op ack', () => { + const dir = mkdtempSync(join(tmpdir(), 'iak-ack-test-')); + const notifyFile = join(dir, 'does-not-exist.txt'); + try { + const r = ackNotificationFile(notifyFile, '[room] a: x\n'); + assert.equal(r.consumedLines, 0); + assert.equal(r.preservedLines, 0); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +// --- end-to-end stdio regression: room_list_new → append → room_ack --------- + +function bootMcp(configPath) { + const child = spawn('node', [BIN, '--config', configPath], { stdio: ['pipe', 'pipe', 'pipe'] }); + let buf = ''; + const waiters = new Map(); + child.stdout.on('data', (b) => { + buf += b.toString('utf8'); + let idx; + while ((idx = buf.indexOf('\n')) >= 0) { + const line = buf.slice(0, idx); + buf = buf.slice(idx + 1); + if (!line.trim()) continue; + let msg; + try { msg = JSON.parse(line); } catch { continue; } + const settle = waiters.get(msg.id); + if (settle) { waiters.delete(msg.id); settle(msg); } + } + }); + let nextId = 100; + const request = (method, params) => + new Promise((resolvePromise, rejectPromise) => { + const id = nextId++; + waiters.set(id, resolvePromise); + setTimeout(() => { + if (waiters.delete(id)) rejectPromise(new Error(`timeout waiting for ${method} (id ${id})`)); + }, 10000).unref(); + child.stdin.write(rpc({ jsonrpc: '2.0', id, method, params })); + }); + child.stdin.write(rpc({ jsonrpc: '2.0', id: 1, method: 'initialize', + params: { protocolVersion: '2025-03-26', capabilities: {}, clientInfo: { name: 'test', version: '0' } } })); + child.stdin.write(rpc({ jsonrpc: '2.0', method: 'notifications/initialized' })); + const close = async () => { + child.kill('SIGTERM'); + await new Promise((r) => child.on('exit', r)); + }; + return { request, close }; +} + +test('iak-mcp.mjs REGRESSION: room_ack clears only what room_list_new returned', async () => { + const dir = mkdtempSync(join(tmpdir(), 'iak-mcp-test-')); + const cfgPath = join(dir, 'config.json'); + const notifyFile = join(dir, 'new-messages.txt'); + writeFileSync(cfgPath, JSON.stringify({ + poller: { notification_file: notifyFile }, + tmux: { allow: [], default_session: 't' }, + })); + writeFileSync(notifyFile, '[room] alice: first\n[room] bob: second\n'); + const { request, close } = bootMcp(cfgPath); + try { + const listed = await request('tools/call', { name: 'room_list_new', arguments: {} }); + assert.match(listed.result.content[0].text, /alice: first/); + // Poller races in between the read and the ack. + appendFileSync(notifyFile, '[room] petrus: send all 3\n'); + const acked = await request('tools/call', { name: 'room_ack', arguments: {} }); + assert.match(acked.result.content[0].text, /preserved/); + assert.equal(readFileSync(notifyFile, 'utf8'), '[room] petrus: send all 3\n'); + // Second read+ack drains the file completely. + const listed2 = await request('tools/call', { name: 'room_list_new', arguments: {} }); + assert.match(listed2.result.content[0].text, /send all 3/); + await request('tools/call', { name: 'room_ack', arguments: {} }); + assert.equal(readFileSync(notifyFile, 'utf8'), ''); + } finally { + await close(); + rmSync(dir, { recursive: true, force: true }); + } +});