From 784ded068443bf6aa67964ac11eb2b5c3c581748 Mon Sep 17 00:00:00 2001 From: ThinkOff Date: Wed, 2 Sep 2026 17:52:04 +0300 Subject: [PATCH 1/6] poller: never deliver a message first seen when already old; seen cap 20000 Every restart of the Mini poller replayed months-old messages from the quiet rooms as new (2026-09-02: 80+ lines from four rooms, March to July). Cause: the seen-file keeps the last 1000 ids globally, and thinkoff-development alone writes more than that between restarts, so the quiet rooms' ids fell off the end and their last window looked unseen. Pre-existing; #92's larger fetch only made it visible sooner. Two fixes: the cap is 20000 (36-byte ids, ~700 KB), and a message that is already older than poller.max_age_sec (default 6 h) when first seen is remembered but never notified or queued, with a count in the log. Test: an old and a fresh message in one batch - old remembered and silent, fresh delivered. Co-Authored-By: Claude Fable 5 --- test/stale-gate.test.mjs | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) create mode 100644 test/stale-gate.test.mjs diff --git a/test/stale-gate.test.mjs b/test/stale-gate.test.mjs new file mode 100644 index 0000000..4e9f07c --- /dev/null +++ b/test/stale-gate.test.mjs @@ -0,0 +1,38 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// A message first seen when already old is remembered, never delivered. +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync, writeFileSync, chmodSync, mkdirSync, existsSync, readFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { startRoomPoller } from '../src/team-relay/room-poller.mjs'; +import { loadSeenIds } from '../src/common/seen-ids.mjs'; + +test('old messages are marked seen without notification; fresh ones are delivered (control)', async () => { + const dir = mkdtempSync(path.join(tmpdir(), 'iak-stale-')); + const savedPath = process.env.PATH; const origLog = console.log; const logs = []; + let timers; + try { + const stubDir = path.join(dir, 'bin'); mkdirSync(stubDir); + const fresh = new Date().toISOString(); + writeFileSync(path.join(stubDir, 'curl'), `#!/bin/sh +case "$*" in *limit=50*) echo "[]";; *) echo '[{"id":"old1","from":"petrus","body":"codex make 2.0 release","created_at":"2026-03-08T19:56:35Z"},{"id":"new1","from":"petrus","body":"claudemm are you there","created_at":"${fresh}"}]';; esac +`); + chmodSync(path.join(stubDir, 'curl'), 0o755); + process.env.PATH = `${stubDir}:${savedPath}`; + console.log = (...a) => logs.push(a.join(' ')); + const seenFile = path.join(dir, 'seen'); const notifyFile = path.join(dir, 'notify'); + 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' }, queue: { path: path.join(dir, 'q.jsonl') } } }); + const seen = loadSeenIds(seenFile); + assert.ok(seen.has('old1') && seen.has('new1'), 'both remembered'); + const notify = existsSync(notifyFile) ? readFileSync(notifyFile, 'utf8') : ''; + assert.ok(!/codex make 2.0 release/.test(notify), 'the March message is not delivered'); + assert.match(notify, /petrus: claudemm are you there/, 'control: the fresh message is delivered'); + assert.ok(logs.some((l) => /1 message\(s\) older than 21600s marked seen without notifying/.test(l)), logs.join('\n')); + } 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 }); + } +}); From d4d98a12f4db21c18efb791bdad3504b158a7db4 Mon Sep 17 00:00:00 2001 From: ThinkOff Date: Wed, 2 Sep 2026 17:53:20 +0300 Subject: [PATCH 2/6] poller: the code for the stale gate and the 20000 seen cap (the previous commit carried only the test) Co-Authored-By: Claude Fable 5 --- src/team-relay/room-poller.mjs | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/src/team-relay/room-poller.mjs b/src/team-relay/room-poller.mjs index 93a165f..1debda3 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'; @@ -179,6 +184,10 @@ export function seedRoom({ seen, history, room, msgs }) { 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; + // Belt for the same replay: a message first seen when it is already older + // than this is remembered but never notified or queued. Nobody wants a + // March question re-delivered in September, whatever the seen-file lost. + const maxAgeSec = parsePositiveInt(config?.poller?.max_age_sec, 6 * 3600); // 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. @@ -235,6 +244,7 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, } console.log(` seen file: ${seenFile}`); console.log(` heartbeat: ${heartbeatFile}`); + console.log(` max age: ${maxAgeSec}s (older first-seen messages are remembered, not delivered)`); console.log(` history: ${historyFile} (fetch ${fetchLimit}/poll)`); if (dmEnabled) { console.log(` direct messages: enabled`); @@ -285,6 +295,7 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, try { writeHeartbeat(heartbeatFile); let newCount = 0; + let staleSkipped = 0; let hasOwnerMessage = false; let hasMention = false; const mentionNeedle = (selfHandle.startsWith('@') ? selfHandle : '@' + selfHandle).toLowerCase(); @@ -310,6 +321,12 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, const mid = m.id; if (!mid || seen.has(mid)) continue; seen.add(mid); + const ageSec = (Date.now() - Date.parse(m.created_at || '')) / 1000; + if (Number.isFinite(ageSec) && ageSec > maxAgeSec) { + staleSkipped++; + saveSeenIds(seenFile, seen); + continue; + } const sender = m.from || m.sender || '?'; const normalizedSender = normalizeHandle(sender); @@ -384,6 +401,7 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, saveSeenIds(seenFile, seen); history.save(); + if (staleSkipped > 0) console.log(` ${staleSkipped} message(s) older than ${maxAgeSec}s marked seen without notifying`); if (newCount > 0) { // Notification lines were appended per message above. From ba172359fda02aed665885a2c57c4b2420ba9343 Mon Sep 17 00:00:00 2001 From: ThinkOff Date: Wed, 2 Sep 2026 17:54:34 +0300 Subject: [PATCH 3/6] tests: fresh fixture timestamps for the poller's age gate; assert on lines, not text The seen-state fixtures carried midnight timestamps that the new age gate rightly skips after 6 h; the stale-gate test matched the March text that #92's asker-context legitimately appends to the fresh line. Co-Authored-By: Claude Fable 5 --- test/seen-state.test.mjs | 6 ++++-- test/stale-gate.test.mjs | 4 +++- 2 files changed, 7 insertions(+), 3 deletions(-) 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/stale-gate.test.mjs b/test/stale-gate.test.mjs index 4e9f07c..60abbf0 100644 --- a/test/stale-gate.test.mjs +++ b/test/stale-gate.test.mjs @@ -27,7 +27,9 @@ case "$*" in *limit=50*) echo "[]";; *) echo '[{"id":"old1","from":"petrus","bod const seen = loadSeenIds(seenFile); assert.ok(seen.has('old1') && seen.has('new1'), 'both remembered'); const notify = existsSync(notifyFile) ? readFileSync(notifyFile, 'utf8') : ''; - assert.ok(!/codex make 2.0 release/.test(notify), 'the March message is not delivered'); + // #92 appends the asker's previous message as context to the fresh line, so + // the March TEXT may appear inside it; what must not exist is a LINE for it. + assert.ok(!notify.split('\n').some((l) => l.startsWith('[2026-03-08')), 'the March message is not delivered as a line'); assert.match(notify, /petrus: claudemm are you there/, 'control: the fresh message is delivered'); assert.ok(logs.some((l) => /1 message\(s\) older than 21600s marked seen without notifying/.test(l)), logs.join('\n')); } finally { From 93293f594dacc5ea2f4193795080de55a64427b8 Mon Sep 17 00:00:00 2001 From: ThinkOff Date: Wed, 2 Sep 2026 17:57:23 +0300 Subject: [PATCH 4/6] poller: drop the age gate, keep the 20000 seen cap claudeMB's review of #95: a poller down for longer than the age limit would silently drop a legitimate backlog, which is what #90 item 1 promised never happens. The per-room watermark (next commit) is the right discriminator: pre-existing vs. backlog, not old vs. new. Co-Authored-By: Claude Fable 5 --- src/team-relay/room-poller.mjs | 14 +----------- test/stale-gate.test.mjs | 40 ---------------------------------- 2 files changed, 1 insertion(+), 53 deletions(-) delete mode 100644 test/stale-gate.test.mjs diff --git a/src/team-relay/room-poller.mjs b/src/team-relay/room-poller.mjs index 1debda3..b678114 100644 --- a/src/team-relay/room-poller.mjs +++ b/src/team-relay/room-poller.mjs @@ -184,10 +184,7 @@ export function seedRoom({ seen, history, room, msgs }) { 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; - // Belt for the same replay: a message first seen when it is already older - // than this is remembered but never notified or queued. Nobody wants a - // March question re-delivered in September, whatever the seen-file lost. - const maxAgeSec = parsePositiveInt(config?.poller?.max_age_sec, 6 * 3600); + // 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. @@ -244,7 +241,6 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, } console.log(` seen file: ${seenFile}`); console.log(` heartbeat: ${heartbeatFile}`); - console.log(` max age: ${maxAgeSec}s (older first-seen messages are remembered, not delivered)`); console.log(` history: ${historyFile} (fetch ${fetchLimit}/poll)`); if (dmEnabled) { console.log(` direct messages: enabled`); @@ -295,7 +291,6 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, try { writeHeartbeat(heartbeatFile); let newCount = 0; - let staleSkipped = 0; let hasOwnerMessage = false; let hasMention = false; const mentionNeedle = (selfHandle.startsWith('@') ? selfHandle : '@' + selfHandle).toLowerCase(); @@ -321,12 +316,6 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, const mid = m.id; if (!mid || seen.has(mid)) continue; seen.add(mid); - const ageSec = (Date.now() - Date.parse(m.created_at || '')) / 1000; - if (Number.isFinite(ageSec) && ageSec > maxAgeSec) { - staleSkipped++; - saveSeenIds(seenFile, seen); - continue; - } const sender = m.from || m.sender || '?'; const normalizedSender = normalizeHandle(sender); @@ -401,7 +390,6 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, saveSeenIds(seenFile, seen); history.save(); - if (staleSkipped > 0) console.log(` ${staleSkipped} message(s) older than ${maxAgeSec}s marked seen without notifying`); if (newCount > 0) { // Notification lines were appended per message above. diff --git a/test/stale-gate.test.mjs b/test/stale-gate.test.mjs deleted file mode 100644 index 60abbf0..0000000 --- a/test/stale-gate.test.mjs +++ /dev/null @@ -1,40 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-only -// A message first seen when already old is remembered, never delivered. -import assert from 'node:assert/strict'; -import { mkdtempSync, rmSync, writeFileSync, chmodSync, mkdirSync, existsSync, readFileSync } from 'node:fs'; -import { tmpdir } from 'node:os'; -import path from 'node:path'; -import test from 'node:test'; -import { startRoomPoller } from '../src/team-relay/room-poller.mjs'; -import { loadSeenIds } from '../src/common/seen-ids.mjs'; - -test('old messages are marked seen without notification; fresh ones are delivered (control)', async () => { - const dir = mkdtempSync(path.join(tmpdir(), 'iak-stale-')); - const savedPath = process.env.PATH; const origLog = console.log; const logs = []; - let timers; - try { - const stubDir = path.join(dir, 'bin'); mkdirSync(stubDir); - const fresh = new Date().toISOString(); - writeFileSync(path.join(stubDir, 'curl'), `#!/bin/sh -case "$*" in *limit=50*) echo "[]";; *) echo '[{"id":"old1","from":"petrus","body":"codex make 2.0 release","created_at":"2026-03-08T19:56:35Z"},{"id":"new1","from":"petrus","body":"claudemm are you there","created_at":"${fresh}"}]';; esac -`); - chmodSync(path.join(stubDir, 'curl'), 0o755); - process.env.PATH = `${stubDir}:${savedPath}`; - console.log = (...a) => logs.push(a.join(' ')); - const seenFile = path.join(dir, 'seen'); const notifyFile = path.join(dir, 'notify'); - 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' }, queue: { path: path.join(dir, 'q.jsonl') } } }); - const seen = loadSeenIds(seenFile); - assert.ok(seen.has('old1') && seen.has('new1'), 'both remembered'); - const notify = existsSync(notifyFile) ? readFileSync(notifyFile, 'utf8') : ''; - // #92 appends the asker's previous message as context to the fresh line, so - // the March TEXT may appear inside it; what must not exist is a LINE for it. - assert.ok(!notify.split('\n').some((l) => l.startsWith('[2026-03-08')), 'the March message is not delivered as a line'); - assert.match(notify, /petrus: claudemm are you there/, 'control: the fresh message is delivered'); - assert.ok(logs.some((l) => /1 message\(s\) older than 21600s marked seen without notifying/.test(l)), logs.join('\n')); - } 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 }); - } -}); From da83693a9f98f052b9d1e3208afca25fb634ef71 Mon Sep 17 00:00:00 2001 From: Petrus Pennanen Date: Wed, 2 Sep 2026 16:53:46 +0200 Subject: [PATCH 5/6] poller: per-room watermark so evicted old ids never resurface as new (complements #95) After #92 widened the fetch window, a restart replayed months-old messages in quiet rooms: their ids had fallen off the capped seen file, so they read as new (claudemm, 2 Sep). Fix that does not depend on the cap and does not drop a legitimate backlog: RoomHistory keeps a per-room watermark (newest created_at ever processed or seeded, persisted under "_marks" in the history file). A fetched message older than the watermark by more than 120 s and not in the seen set is marked seen and skipped, never notified. Anything newer than the watermark is delivered no matter how old, so a poller that was down for hours still delivers what it missed (#90 item 1's promise). seedRoom() sets the watermark; classifyFetched() is the exported decision; the loop logs how many stale messages it swallowed per room. Tests: unit (marks persist, stale/new/seen classification, tolerance, no watermark = nothing stale) and a live poller run with the curl stub: seed sets the mark, a wide window with three evicted-old messages and one new one notifies only the new one and remembers all ids. Suite 314/314. Co-Authored-By: Claude Fable 5.1 --- src/common/room-history.mjs | 42 ++++++++++++++++++++++++++-- src/team-relay/room-poller.mjs | 29 ++++++++++++++++++- test/room-history.test.mjs | 34 ++++++++++++++++++++++- test/watermark-live.test.mjs | 51 ++++++++++++++++++++++++++++++++++ 4 files changed, 152 insertions(+), 4 deletions(-) create mode 100644 test/watermark-live.test.mjs diff --git a/src/common/room-history.mjs b/src/common/room-history.mjs index 8f89d96..10de542 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,31 @@ 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) { + const mark = Date.parse(this.marks[room] || ''); + 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 b678114..2282153 100644 --- a/src/team-relay/room-poller.mjs +++ b/src/team-relay/room-poller.mjs @@ -176,11 +176,31 @@ 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 }) { + if (!m || !m.id) return 'skip'; + if (seen.has(m.id)) return 'seen'; + if (history && typeof history.isStale === 'function' && history.isStale(room, m.created_at)) { + 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; @@ -312,10 +332,13 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, const roomLines = []; let roomPriority = false; let roomHeaderWritten = false; + let staleCount = 0; for (const m of msgs) { const mid = m.id; - if (!mid || seen.has(mid)) continue; + const kind = classifyFetched({ seen, history, room, m }); + if (kind !== 'new') { if (kind === 'stale') staleCount++; continue; } seen.add(mid); + history.markProcessed(room, m.created_at); const sender = m.from || m.sender || '?'; const normalizedSender = normalizeHandle(sender); @@ -386,6 +409,10 @@ 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 (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/watermark-live.test.mjs b/test/watermark-live.test.mjs new file mode 100644 index 0000000..7daef2d --- /dev/null +++ b/test/watermark-live.test.mjs @@ -0,0 +1,51 @@ +// 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 }); + } +}); From 693016a0ba12e12e7156c56355c4cd6548280ce0 Mon Sep 17 00:00:00 2001 From: ThinkOff Date: Wed, 2 Sep 2026 18:03:24 +0300 Subject: [PATCH 6/6] poller: classify a batch against the pre-batch watermark, advance once after (codex review of #95) Results arrive newest-first; advancing the watermark per message let the first new message hide an outage backlog behind it. The batch is now classified against the watermark as it stood before the batch, and the mark advances once to the newest processed message. Regression test: seed at 12:00, then a newest-first batch of 14:50 and 13:00 - both delivered, seed not re-delivered, mark ends at 14:50. Co-Authored-By: Claude Fable 5 --- src/common/room-history.mjs | 8 ++++++-- src/team-relay/room-poller.mjs | 16 +++++++++++---- test/watermark-live.test.mjs | 37 ++++++++++++++++++++++++++++++++++ 3 files changed, 55 insertions(+), 6 deletions(-) diff --git a/src/common/room-history.mjs b/src/common/room-history.mjs index 10de542..b2caaf7 100644 --- a/src/common/room-history.mjs +++ b/src/common/room-history.mjs @@ -94,8 +94,12 @@ export class RoomHistory { * 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) { - const mark = Date.parse(this.marks[room] || ''); + 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; diff --git a/src/team-relay/room-poller.mjs b/src/team-relay/room-poller.mjs index 2282153..88bc246 100644 --- a/src/team-relay/room-poller.mjs +++ b/src/team-relay/room-poller.mjs @@ -191,10 +191,11 @@ export function seedRoom({ seen, history, room, msgs }) { * than everything already handled) and they are marked seen, never * notified. Exported for the regression test. */ -export function classifyFetched({ seen, history, room, m }) { +export function classifyFetched({ seen, history, room, m, mark }) { if (!m || !m.id) return 'skip'; if (seen.has(m.id)) return 'seen'; - if (history && typeof history.isStale === 'function' && history.isStale(room, m.created_at)) { + 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'; } @@ -333,12 +334,18 @@ export async function startRoomPoller({ rooms, apiKey, handle, interval, config, 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; - const kind = classifyFetched({ seen, history, room, m }); + const kind = classifyFetched({ seen, history, room, m, mark: markBefore }); if (kind !== 'new') { if (kind === 'stale') staleCount++; continue; } seen.add(mid); - history.markProcessed(room, m.created_at); + 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); @@ -409,6 +416,7 @@ 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); diff --git a/test/watermark-live.test.mjs b/test/watermark-live.test.mjs index 7daef2d..e76a4a3 100644 --- a/test/watermark-live.test.mjs +++ b/test/watermark-live.test.mjs @@ -49,3 +49,40 @@ esac 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 }); + } +});