fix(mcp): room_ack clears only consumed lines, not the whole file

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

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 });
+  }
+});