diff --git a/src/common/room-history.mjs b/src/common/room-history.mjs index 8f89d96..b2caaf7 100644 --- a/src/common/room-history.mjs +++ b/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,15 +46,21 @@ 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 = {}; } } @@ -55,7 +68,7 @@ export class RoomHistory { 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 { @@ -63,6 +76,35 @@ export class RoomHistory { } } + /** 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; diff --git a/src/team-relay/room-poller.mjs b/src/team-relay/room-poller.mjs index 93a165f..88bc246 100644 --- a/src/team-relay/room-poller.mjs +++ b/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); diff --git a/test/room-history.test.mjs b/test/room-history.test.mjs index d038ed9..18da060 100644 --- a/test/room-history.test.mjs +++ b/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'); + }); +}); diff --git a/test/seen-state.test.mjs b/test/seen-state.test.mjs index 9b06798..e9d5c22 100644 --- a/test/seen-state.test.mjs +++ b/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}`; diff --git a/test/watermark-live.test.mjs b/test/watermark-live.test.mjs new file mode 100644 index 0000000..e76a4a3 --- /dev/null +++ b/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 }); + } +});