poller: seen cap 20000 + per-room watermark, so evicted old ids never resurface as new

src/common/room-history.mjs

@@ -24,6 +24,13 @@ import { replyIdOf } from './reply-context.mjs';
 
 const BODY_KEEP = 600;
 export const STATE_PREFIX = /^\s*(?:state|settled)\s*:\s*(.+)$/i;
+const MARKS_KEY = '_marks';
+// A message this much older than the room's watermark, and not in the seen
+// set, is treated as already handled: the seen-id cap (2000 ids across all
+// rooms) evicts old ids, and a fetch window wider than 10 then re-surfaces
+// months-old messages in quiet rooms as "new" (claudemm, 2 Sep 2026, after
+// #92 raised the window to 25). The tolerance covers out-of-order arrival.
+export const STALE_TOLERANCE_S = 120;
 
 function oneLine(s, n) {
   return String(s || '').replace(/\s+/g, ' ').trim().slice(0, n);
@@ -39,30 +46,65 @@ export class RoomHistory {
     this.path = path;
     this.maxPerRoom = maxPerRoom;
     this.rooms = {};
+    this.marks = {};   // room -> newest created_at ever processed or seeded
     this.load();
   }
 
   load() {
     try {
       const data = JSON.parse(readFileSync(this.path, 'utf8'));
-      if (data && typeof data === 'object') this.rooms = data;
+      if (data && typeof data === 'object') {
+        const { [MARKS_KEY]: marks, ...rooms } = data;
+        this.rooms = rooms;
+        this.marks = (marks && typeof marks === 'object') ? marks : {};
+      }
     } catch {
       this.rooms = {};
+      this.marks = {};
     }
   }
 
   save() {
     try {
       mkdirSync(dirname(this.path), { recursive: true });
       const tmp = this.path + '.tmp';
-      writeFileSync(tmp, JSON.stringify(this.rooms));
+      writeFileSync(tmp, JSON.stringify({ ...this.rooms, [MARKS_KEY]: this.marks }));
       renameSync(tmp, this.path);
       return true;
     } catch {
       return false;
     }
   }
 
+  /** Advance the room's watermark to `createdAt` if it is newer. */
+  markProcessed(room, createdAt) {
+    if (!room || !createdAt) return;
+    const t = Date.parse(createdAt);
+    if (!Number.isFinite(t)) return;
+    const cur = Date.parse(this.marks[room] || '');
+    if (!Number.isFinite(cur) || t > cur) this.marks[room] = new Date(t).toISOString();
+  }
+
+  watermark(room) {
+    return this.marks[room] || '';
+  }
+
+  /**
+   * True when a message is older than the room's watermark by more than the
+   * tolerance: it predates everything already handled here, so it can only
+   * be an evicted-from-seen resurfacing, never a genuinely new message.
+   */
+  isStale(room, createdAt, toleranceS = STALE_TOLERANCE_S, markIso = this.marks[room]) {
+    // `markIso` lets a caller classify a whole batch against the watermark as
+    // it stood BEFORE the batch: advancing it per message would let the first
+    // new message in a newest-first batch hide the backlog behind it (codex
+    // review of PR #95).
+    const mark = Date.parse(markIso || '');
+    const t = Date.parse(createdAt || '');
+    if (!Number.isFinite(mark) || !Number.isFinite(t)) return false;
+    return t < mark - toleranceS * 1000;
+  }
+
   /** Add or refresh a fetched batch. Idempotent; keeps newest-last order. */
   remember(room, msgs) {
     if (!room || !Array.isArray(msgs) || msgs.length === 0) return;

src/team-relay/room-poller.mjs

@@ -35,7 +35,12 @@ import { RoomHistory, threadSuffix, previousSuffix, stateOfPlayLine, ownLastPost
 // The marker is written after EACH handled message below, never once per
 // batch: a batch can hold a task that runs for an hour, and a restart inside
 // it replayed a whole day on the M5 (2026-09-01).
-const saveSeenIds = (path, ids) => saveSeenIdsShared(path, ids, 1000);
+// 20000, not 1000: the cap is global across rooms, and thinkoff-development
+// alone produces >1000 messages between restarts, so the quiet rooms' last
+// ids fell off the end of the file and EVERY restart replayed months-old
+// messages as new (2026-09-02: 80+ replayed lines from four rooms).
+const SEEN_CAP = 20000;
+const saveSeenIds = (path, ids) => saveSeenIdsShared(path, ids, SEEN_CAP);
 
 const DM_SEEN_FILE_DEFAULT = '/tmp/iak-dm-seen-ids.txt';
 
@@ -171,14 +176,36 @@ export function seedRoom({ seen, history, room, msgs }) {
   let added = 0;
   for (const m of msgs || []) {
     if (m && m.id && !seen.has(m.id)) { seen.add(m.id); added++; }
+    if (history && typeof history.markProcessed === 'function' && m && m.created_at) {
+      history.markProcessed(room, m.created_at);
+    }
   }
   if (history && typeof history.remember === 'function') history.remember(room, msgs || []);
   return added;
 }
 
+/**
+ * Is this fetched message one to handle, or an old one resurfacing? The
+ * seen set is capped, so a wide fetch window can show months-old messages
+ * whose ids were evicted; the room watermark catches those (they are older
+ * than everything already handled) and they are marked seen, never
+ * notified. Exported for the regression test.
+ */
+export function classifyFetched({ seen, history, room, m, mark }) {
+  if (!m || !m.id) return 'skip';
+  if (seen.has(m.id)) return 'seen';
+  const markIso = mark !== undefined ? mark : (history && typeof history.watermark === 'function' ? history.watermark(room) : undefined);
+  if (history && typeof history.isStale === 'function' && history.isStale(room, m.created_at, undefined, markIso)) {
+    seen.add(m.id);
+    return 'stale';
+  }
+  return 'new';
+}
+
 export async function startRoomPoller({ rooms, apiKey, handle, interval, config, sessionOpt }) {
   const seenFile = config?.poller?.seen_file || SEEN_FILE_DEFAULT;
   const heartbeatFile = config?.poller?.heartbeat_file || HEARTBEAT_FILE_DEFAULT;
+
   // Per-room message history: resolves reply targets older than the fetch
   // window, supplies the asker's previous message, the agent's own last post
   // and the room's "state:" facts (issue #90). One file per poller.
@@ -306,10 +333,19 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config,
       const roomLines = [];
       let roomPriority = false;
       let roomHeaderWritten = false;
+      let staleCount = 0;
+      // Classify the whole batch against the watermark as it stood before the
+      // batch, and advance it once afterwards: results arrive newest-first,
+      // so advancing per message would classify an outage backlog older than
+      // the first new message as stale and suppress it (codex, PR #95).
+      const markBefore = history.watermark(room);
+      let newestProcessed = '';
       for (const m of msgs) {
         const mid = m.id;
-        if (!mid || seen.has(mid)) continue;
+        const kind = classifyFetched({ seen, history, room, m, mark: markBefore });
+        if (kind !== 'new') { if (kind === 'stale') staleCount++; continue; }
         seen.add(mid);
+        if (m.created_at && (!newestProcessed || Date.parse(m.created_at) > Date.parse(newestProcessed))) newestProcessed = m.created_at;
 
         const sender = m.from || m.sender || '?';
         const normalizedSender = normalizeHandle(sender);
@@ -380,6 +416,11 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config,
       // Lines and the context header were written per message above;
       // newMessages only feeds the count/log below.
       newMessages.push(...roomLines);
+      if (newestProcessed) history.markProcessed(room, newestProcessed);
+      if (staleCount) {
+        console.log(`  ${room}: ${staleCount} old message(s) below the watermark ${history.watermark(room)} marked seen, not notified`);
+        saveSeenIds(seenFile, seen);
+      }
     }
 
     saveSeenIds(seenFile, seen);

test/room-history.test.mjs

@@ -7,7 +7,7 @@ import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { RoomHistory, threadSuffix, previousSuffix, stateOfPlayLine, ownLastPostLine } from '../src/common/room-history.mjs';
 import { resolveReplyTargets } from '../src/common/reply-context.mjs';
-import { seedRoom } from '../src/team-relay/room-poller.mjs';
+import { seedRoom, classifyFetched } from '../src/team-relay/room-poller.mjs';
 
 const R = 'thinkoff-development';
 function msg(id, from, body, created_at, reply_to) {
@@ -131,3 +131,35 @@ describe('first-run seed keeps the thread (codexmb, PR #92)', () => {
     assert.ok(threadSuffix(h, R, batch[0]).includes('Card v2 for your review'));
   });
 });
+
+
+describe('room watermark: evicted old ids must not resurface as new (claudemm, 2 Sep)', () => {
+  it('marks advance on seed and processing, persist, and classify stale vs new', () => {
+    const h = fresh();
+    const seen = new Set();
+    seedRoom({ seen, history: h, room: R, msgs: [
+      msg('s1', 'petrus', 'old one', '2026-06-01T10:00:00Z'),
+      msg('s2', 'petrus', 'old two', '2026-08-31T12:00:00Z')
+    ] });
+    assert.equal(h.watermark(R), '2026-08-31T12:00:00.000Z');
+    assert.ok(h.save());
+    const again = new RoomHistory(h.path);
+    assert.equal(again.watermark(R), '2026-08-31T12:00:00.000Z', 'watermark survives restart');
+    assert.equal(again.get(R, 's2').body, 'old two', 'rooms still load beside the marks');
+    // the seen cap evicted every old id: a wide window now shows a February message
+    const evicted = new Set();
+    const feb = msg('feb', 'petrus', 'from february', '2026-02-10T09:00:00Z');
+    assert.equal(classifyFetched({ seen: evicted, history: again, room: R, m: feb }), 'stale');
+    assert.ok(evicted.has('feb'), 'stale message is marked seen so it never comes back');
+    // a genuinely new message is new; one 60 s before the mark is within tolerance
+    assert.equal(classifyFetched({ seen: evicted, history: again, room: R, m: msg('n1', 'petrus', 'new', '2026-09-02T14:50:00Z') }), 'new');
+    assert.equal(classifyFetched({ seen: evicted, history: again, room: R, m: msg('n2', 'petrus', 'late arrival', '2026-08-31T11:59:00Z') }), 'new');
+    evicted.add('n2');   // the loop adds a NEW id after classifying it
+    assert.equal(classifyFetched({ seen: evicted, history: again, room: R, m: msg('n2', 'petrus', 'dup', '2026-08-31T11:59:00Z') }), 'seen');
+    again.markProcessed(R, '2026-09-02T14:50:00Z');
+    again.markProcessed(R, '2026-09-01T00:00:00Z');   // older: never moves the mark back
+    assert.equal(again.watermark(R), '2026-09-02T14:50:00.000Z');
+    // no watermark yet (fresh poller, first run): nothing is stale
+    assert.equal(classifyFetched({ seen: new Set(), history: fresh(), room: R, m: feb }), 'new');
+  });
+});

test/seen-state.test.mjs

@@ -33,8 +33,9 @@ test('the live poller remembers each message as it is handled, not once per batc
     const stub = path.join(stubDir, 'curl');
     // first call (seeding, limit=50) must return nothing so the ids are unseen;
     // later calls return the two messages
+    const t1 = new Date(Date.now() - 2000).toISOString(); const t2 = new Date(Date.now() - 1000).toISOString();
     writeFileSync(stub, `#!/bin/sh
-case "$*" in *limit=50*) echo "[]";; *) echo '[{"id":"m1","from":"petrus","body":"hello one","created_at":"2026-09-02T00:00:01Z"},{"id":"m2","from":"petrus","body":123,"created_at":"2026-09-02T00:00:02Z"}]';; esac
+case "$*" in *limit=50*) echo "[]";; *) echo '[{"id":"m1","from":"petrus","body":"hello one","created_at":"${t1}"},{"id":"m2","from":"petrus","body":123,"created_at":"${t2}"}]';; esac
 `);
     chmodSync(stub, 0o755);
     process.env.PATH = `${stubDir}:${savedPath}`;
@@ -66,8 +67,9 @@ test('with #92 thread context: the state-of-play header lands once, before the f
   try {
     const stubDir = path.join(dir, 'bin'); mkdirSync(stubDir);
     const stub = path.join(stubDir, 'curl');
+    const t1 = new Date(Date.now() - 2000).toISOString(); const t2 = new Date(Date.now() - 1000).toISOString();
     writeFileSync(stub, `#!/bin/sh
-case "$*" in *limit=50*) echo "[]";; *) echo '[{"id":"s1","from":"@claudeMB","body":"state: the card is the benchmark table v2, not hardware","created_at":"2026-09-02T00:00:01Z"},{"id":"o1","from":"petrus","body":"claudemm what is the card","created_at":"2026-09-02T00:00:02Z"}]';; esac
+case "$*" in *limit=50*) echo "[]";; *) echo '[{"id":"s1","from":"@claudeMB","body":"state: the card is the benchmark table v2, not hardware","created_at":"${t1}"},{"id":"o1","from":"petrus","body":"claudemm what is the card","created_at":"${t2}"}]';; esac
 `);
     chmodSync(stub, 0o755);
     process.env.PATH = `${stubDir}:${savedPath}`;

test/watermark-live.test.mjs

@@ -0,0 +1,88 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+// Regression (claudemm, 2 Sep 2026): after #92 widened the fetch window, a
+// restart replayed months-old messages whose ids the seen cap had evicted.
+import assert from 'node:assert/strict';
+import { mkdtempSync, rmSync, writeFileSync, readFileSync, chmodSync, mkdirSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import path from 'node:path';
+import test from 'node:test';
+import { loadSeenIds } from '../src/common/seen-ids.mjs';
+import { startRoomPoller } from '../src/team-relay/room-poller.mjs';
+
+test('old messages below the room watermark are marked seen and never notified, new ones still are', async () => {
+  const dir = mkdtempSync(path.join(tmpdir(), 'iak-wm-'));
+  const savedPath = process.env.PATH;
+  let timers;
+  try {
+    const stubDir = path.join(dir, 'bin'); mkdirSync(stubDir);
+    // seed (limit=50): two messages from 31 Aug set the watermark.
+    // first poll (limit=25): the wide window now shows three February/June
+    // messages with ids nobody has seen (evicted from the seen cap on a real
+    // box) plus one genuinely new message.
+    writeFileSync(path.join(stubDir, 'curl'), `#!/bin/sh
+case "$*" in
+  *limit=50*) echo '[{"id":"a1","from":"petrus","body":"old aug one","created_at":"2026-08-31T12:00:00Z"},{"id":"a2","from":"petrus","body":"old aug two","created_at":"2026-08-31T12:05:00Z"}]';;
+  *) echo '[{"id":"n1","from":"petrus","body":"genuinely new","created_at":"2026-09-02T14:50:00Z"},{"id":"a2","from":"petrus","body":"old aug two","created_at":"2026-08-31T12:05:00Z"},{"id":"f1","from":"petrus","body":"from february","created_at":"2026-02-10T09:00:00Z"},{"id":"f2","from":"@x","body":"from june","created_at":"2026-06-03T09:00:00Z"},{"id":"f3","from":"petrus","body":"from may","created_at":"2026-05-01T09:00:00Z"}]';;
+esac
+`);
+    chmodSync(path.join(stubDir, 'curl'), 0o755);
+    process.env.PATH = `${stubDir}:${savedPath}`;
+    const seenFile = path.join(dir, 'seen'); const notifyFile = path.join(dir, 'notify');
+    const origLog = console.log; console.log = () => {};
+    try {
+      timers = await startRoomPoller({
+        rooms: ['r'], apiKey: 'k', handle: '@t', interval: 3600,
+        config: { poller: { seen_file: seenFile, notification_file: notifyFile, heartbeat_file: path.join(dir, 'hb'), nudge_mode: 'none', owner_handle: 'petrus', history_file: path.join(dir, 'hist.json') }, queue: { path: path.join(dir, 'q.jsonl') } }
+      });
+    } finally { console.log = origLog; }
+    const lines = readFileSync(notifyFile, 'utf8').split('\n').filter(Boolean);
+    assert.equal(lines.filter((l) => /petrus: genuinely new/.test(l)).length, 1, `the new message is notified once: ${lines.join(' | ')}`);
+    assert.ok(!lines.some((l) => /from february|from june|from may/.test(l)), `old messages never reach the notification file: ${lines.join(' | ')}`);
+    const seen = loadSeenIds(seenFile);
+    for (const id of ['a1', 'a2', 'n1', 'f1', 'f2', 'f3']) assert.ok(seen.has(id), `${id} is in the seen file`);
+    const hist = JSON.parse(readFileSync(path.join(dir, 'hist.json'), 'utf8'));
+    assert.equal(hist._marks.r, '2026-09-02T14:50:00.000Z', 'watermark advanced to the new message');
+  } finally {
+    if (timers?.roomTimer) clearInterval(timers.roomTimer);
+    if (timers?.dmTimer) clearInterval(timers.dmTimer);
+    process.env.PATH = savedPath;
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test('an outage backlog behind the first new message in a newest-first batch is still delivered (codex, PR #95)', async () => {
+  const { mkdtempSync, rmSync, writeFileSync, chmodSync, mkdirSync, readFileSync } = await import('node:fs');
+  const { tmpdir } = await import('node:os');
+  const path = (await import('node:path')).default;
+  const { startRoomPoller } = await import('../src/team-relay/room-poller.mjs');
+  const dir = mkdtempSync(path.join(tmpdir(), 'iak-wm-backlog-'));
+  const savedPath = process.env.PATH; const origLog = console.log; console.log = () => {};
+  let timers;
+  try {
+    const stubDir = path.join(dir, 'bin'); mkdirSync(stubDir);
+    // seed sets the watermark at 12:00; the poller then "returns from an outage"
+    // and the fetch shows, newest first: a 14:50 message and a 13:00 backlog
+    // message. Both are newer than the watermark and both must be delivered.
+    writeFileSync(path.join(stubDir, 'curl'), `#!/bin/sh
+case "$*" in
+  *limit=50*) echo '[{"id":"a1","from":"petrus","body":"seed","created_at":"2026-08-31T12:00:00Z"}]';;
+  *) echo '[{"id":"n1","from":"petrus","body":"newest first","created_at":"2026-09-02T14:50:00Z"},{"id":"b1","from":"petrus","body":"backlog from the outage","created_at":"2026-09-02T13:00:00Z"},{"id":"a1","from":"petrus","body":"seed","created_at":"2026-08-31T12:00:00Z"}]';;
+esac
+`);
+    chmodSync(path.join(stubDir, 'curl'), 0o755);
+    process.env.PATH = `${stubDir}:${savedPath}`;
+    const notifyFile = path.join(dir, 'notify');
+    timers = await startRoomPoller({ rooms: ['r'], apiKey: 'k', handle: '@t', interval: 3600,
+      config: { poller: { seen_file: path.join(dir, 'seen'), notification_file: notifyFile, heartbeat_file: path.join(dir, 'hb'), nudge_mode: 'none', history_file: path.join(dir, 'hist.json') }, queue: { path: path.join(dir, 'q.jsonl') } } });
+    const lines = readFileSync(notifyFile, 'utf8').split('\n').filter(Boolean);
+    assert.equal(lines.filter((l) => /petrus: newest first/.test(l)).length, 1, 'new message delivered once');
+    assert.equal(lines.filter((l) => /petrus: backlog from the outage/.test(l)).length, 1, `backlog behind it delivered too: ${lines.join(' | ')}`);
+    assert.ok(!lines.some((l) => /petrus: seed/.test(l)), 'the seeded message is not re-delivered');
+    const hist = JSON.parse(readFileSync(path.join(dir, 'hist.json'), 'utf8'));
+    assert.equal(hist._marks.r, '2026-09-02T14:50:00.000Z', 'watermark advanced once, to the newest processed');
+  } finally {
+    if (timers?.roomTimer) clearInterval(timers.roomTimer);
+    if (timers?.dmTimer) clearInterval(timers.dmTimer);
+    console.log = origLog; process.env.PATH = savedPath; rmSync(dir, { recursive: true, force: true });
+  }
+});