From e7cf80375f018fab0a2b7d2aeb49b2880911bfb3 Mon Sep 17 00:00:00 2001 From: Colin Walters Date: Fri, 9 Oct 2026 14:08:38 -0400 Subject: [PATCH] poll: Wake promptly for operator comments on active threads Operator replies without a mention should reach the dispatcher without waiting for the full sweep. Add a conditional notifications fast path to the existing loop, verifying the API author and tracked open thread before reporting a news wake. Keep the sweep as the safety net. Generated-by: AI Agent-run: https://github.com/bootc-dev/cgwalters-devspace-sandbox/actions/runs/37953108069 Agent-run: https://github.com/bootc-dev/cgwalters-devspace-sandbox/actions/runs/37969725194 --- bin/bot-poll-loop | 23 ++- bin/poll-notifications.js | 208 ++++++++++++++++++++++ bin/poll-notifications.sh | 9 + docs/scheduled-dispatcher.md | 10 ++ tests/bot-poll-loop.sh | 61 +++++++ tests/bot-supervisor.sh | 2 +- tests/bot-supervisor.test.js | 9 + tests/fixtures/poll-notifications.json | 15 ++ tests/poll-notifications.test.js | 233 +++++++++++++++++++++++++ 9 files changed, 567 insertions(+), 3 deletions(-) create mode 100644 bin/poll-notifications.js create mode 100644 bin/poll-notifications.sh create mode 100644 tests/fixtures/poll-notifications.json create mode 100644 tests/poll-notifications.test.js diff --git a/bin/bot-poll-loop b/bin/bot-poll-loop index 853e030..07e5365 100755 --- a/bin/bot-poll-loop +++ b/bin/bot-poll-loop @@ -23,7 +23,7 @@ readonly STALE_S=$((30 * 60)) SP_EVERY_S=3600 readonly HEARTBEAT_FRESH_MIN=30 # Cycles of FAST_S seconds after the first look (about 9 minutes), and # how long an unresolved action waits before it wakes again. -readonly CYCLES=${BOT_POLL_LOOP_CYCLES:-3} FAST_S=${BOT_POLL_LOOP_FAST_S:-180} +readonly CYCLES=${BOT_POLL_LOOP_CYCLES:-9} FAST_S=${BOT_POLL_LOOP_FAST_S:-60} readonly RESYNC_MIN=${BOT_POLL_LOOP_RESYNC_MIN:-120} # --until-actions: how long it waits at most by default (under the 30 # minutes a Claude Code background command gets by default), and after @@ -96,6 +96,12 @@ ${OP_FORGE_ORG} or ${OP_BOT_LOGIN}'s own, on the board (or in an item's Branch), or opened by ${OP_BOT_LOGIN}. The seen-sets and the actions state live in ${S}. +Each cycle also conditionally polls the bot's notifications, respecting +X-Poll-Interval across restarts. A latest comment verified as written by +${OP_OPERATOR_LOGIN} on an open thread authored by or assigned to the bot, +or on the board (including Branch PRs), wakes as news, fast, with its URL. +It never marks threads read; failed reads retain the window, and a fresh +state looks back ten minutes. The full sweep remains the safety net. EOF } @@ -142,7 +148,9 @@ mkdir -p "${S}" || exit 1 # reconcile, say) that holds the lock a little longer: wait for it. exec {lock_fd}>"${S}/lock" || exit 1 flock -w "${BOT_POLL_LOOP_LOCK_WAIT:-60}" "${lock_fd}" || die "another $(basename "$0") run holds ${S}/lock" -for k in ${KEYS} evapprove; do touch "${S}/${k}-now" "${S}/${k}-seen"; done +for k in ${KEYS} evapprove comment; do touch "${S}/${k}-now" "${S}/${k}-seen"; done +# shellcheck source=bin/poll-notifications.sh disable=SC1091 +source "${BIN}/poll-notifications.sh" # events: the operator's approving reviews in their public events feed. events() { @@ -345,6 +353,17 @@ while :; do done refresh reconcile + notification_news="" + # Do not spend the notification budget delaying already pending actions. + if test -z "${hits}"; then + if ! notifications; then + echo "warning: notifications fast path failed; retrying next poll" >&2 + fi + fi + if test -n "${notification_news}"; then + hits="news, fast${hits}" + wake "operator comments: $(tr '\n' ' ' <<<"${notification_news}")" + fi if why=$(sweep_problem); then hits="sweep-problem${hits}" wake "${why}" diff --git a/bin/poll-notifications.js b/bin/poll-notifications.js new file mode 100644 index 0000000..cbda185 --- /dev/null +++ b/bin/poll-notifications.js @@ -0,0 +1,208 @@ +// Read-only notification polling; the caller holds the poll-loop lock. +const fs = require('node:fs'); +const path = require('node:path'); +const { spawnSync } = require('node:child_process'); + +const MAX_REQUESTS = 24; +const MAX_PAGES = 4; +const MAX_BYTES = 1024 * 1024; +const MAX_THREADS = 100; +const MAX_RETRIES = 200; +const RETRY_LIFETIME = 24 * 60 * 60 * 1000; +const RETRY_DELAY = 60 * 1000; +const MAX_RETRY_DELAY = 60 * 60 * 1000; +const terminal = new Set([404, 410, 451]); +const commentURL = /^https:\/\/api\.github\.com\/repos\/[^/]+\/[^/]+\/(issues\/comments|pulls\/comments|pulls\/[0-9]+\/reviews)\/[0-9]+$/; +const issueURL = /^https:\/\/api\.github\.com\/repos\/([^/]+\/[^/]+)\/(issues|pulls)\/([0-9]+)$/; + +function poll({ dir, tools, bot, operator, seconds = 10, run = spawnSync, now = Date.now }) { + const read = (name, fallback = '') => { + try { return fs.readFileSync(path.join(dir, name), 'utf8').trim(); } + catch (error) { if (error.code === 'ENOENT') return fallback; throw error; } + }; + const write = (name, value) => { + fs.writeFileSync(path.join(dir, `${name}.tmp`), value); + fs.renameSync(path.join(dir, `${name}.tmp`), path.join(dir, name)); + }; + const deadline = Date.now() + Math.min(10, seconds) * 1000; + let requests = 0; + const invoke = (command, args) => { + if (requests >= MAX_REQUESTS || Date.now() >= deadline) throw new Error('notification batch budget exhausted'); + requests++; + const result = run(command, args, { encoding: 'utf8', timeout: Math.max(1, Math.min(2000, deadline - Date.now())), killSignal: 'SIGKILL', maxBuffer: MAX_BYTES }); + if (result.error) throw new Error(`notification request failed: ${result.error.message}`); + return result; + }; + const api = (url, headers = []) => { + const result = invoke('gh', ['api', '-i', url, ...headers]); + const raw = result.stdout.replace(/\r\n/g, '\n'); + const split = raw.indexOf('\n\n'); + const lines = raw.slice(0, split).split('\n'); + const status = Number(lines.shift()?.split(' ')[1]); + const values = {}; + for (const line of lines) { + const colon = line.indexOf(':'); + if (colon > 0) values[line.slice(0, colon).toLowerCase()] = line.slice(colon + 1).trim(); + } + if (terminal.has(status)) return { status, headers: values }; + if (status === 304) return { status, headers: values }; + if (result.status !== 0 || status !== 200 || split < 0) throw new Error(`notification API read failed (${status || 'no HTTP status'})`); + return { status, headers: values, body: JSON.parse(raw.slice(split + 2)) }; + }; + const fallback = reason => console.error(`warning: ${reason}; falling back to the full sweep`); + const emptyState = () => ({ queue: [], retry: [], next: null, visited: [] }); + let state; + try { + // Check before allocating or parsing an old, potentially unbounded backlog. + if (fs.statSync(path.join(dir, 'notification-work')).size > MAX_BYTES) { + fallback('notification retained state exceeds byte limit'); + } else state = JSON.parse(read('notification-work', 'null')); + } catch (error) { + if (error.code !== 'ENOENT') fallback(`invalid notification retained state: ${error.message}`); + } + if (!state) state = { queue: [], retry: [], next: null, visited: [] }; + const keyOf = thread => thread.subject?.url; + const coalesce = threads => { + const merged = new Map(); + for (const thread of threads) { + const key = keyOf(thread); + if (!key) continue; + const previous = merged.get(key); + merged.set(key, previous ? { + ...previous, subject: thread.subject, + pollSince: [previous.pollSince, thread.pollSince].filter(Boolean).sort()[0], + } : thread); + } + return [...merged.values()]; + }; + const boundRetries = () => { + const merged = coalesce(state.retry); + for (const thread of merged) thread.retryStarted ??= now(); + state.retry = merged.filter(thread => now() - thread.retryStarted < RETRY_LIFETIME).slice(0, MAX_RETRIES); + if (state.retry.length < merged.length) fallback('notification retries expired or exceeded count limit'); + }; + boundRetries(); + const seen = new Set(read('comment-seen').split('\n').filter(Boolean)); + const news = []; + let failed = false; + let board; + const since = read('notification-since', new Date(Date.now() - 600000).toISOString().replace('.000', '')); + if (!read('notification-since')) write('notification-since', since); + try { + if (Date.now() / 1000 < Number(read('notification-due', '0'))) return []; + const identity = invoke('gh', ['api', 'user', '--jq', '.login']); + if (identity.status !== 0 || identity.stdout.trim() !== bot) throw new Error('notification bot identity verification failed'); + // Acquisition and first attempts have priority over retries. Keep retries + // separate so a failing backlog cannot pin pagination or the next window. + let retriesRemaining = state.retry.length; + let pages = 0; + while (!state.queue.length && pages < MAX_PAGES) { + if (!state.next && state.visited.length) break; + const url = state.next || `notifications?all=true&since=${since}&per_page=100`; + if (state.visited.includes(url)) { + // A cyclic feed cannot be completed: restart the same window later. + state.next = null; + state.visited = []; + failed = true; + break; + } + if (state.next && !/^https:\/\/api\.github\.com\/notifications\?/.test(url)) throw new Error('unsafe notification pagination URL'); + const modified = read('notification-modified'); + const response = api(url, !state.next && modified ? ['-H', `If-Modified-Since: ${modified}`] : []); + const interval = /^\d+$/.test(response.headers['x-poll-interval'] || '') ? Number(response.headers['x-poll-interval']) : 60; + write('notification-due', String(Math.floor(Date.now() / 1000) + interval)); + if (response.status === 304) break; + if (!Array.isArray(response.body) || response.body.length > MAX_THREADS) throw new Error('invalid or oversized notification page'); + if (!state.visited.length) { + const date = new Date(response.headers.date || Date.now()); + state.until = date.toISOString().replace('.000', ''); + state.modified = response.headers['last-modified'] || ''; + } + state.visited.push(url); + state.next = response.headers.link?.match(/<([^>]+)>; rel="next"/)?.[1] || null; + state.queue = response.body.filter(t => ['Issue', 'PullRequest'].includes(t.subject?.type)).map(t => ({ ...t, pollSince: since })); + pages++; + } + state.queue = coalesce(state.queue); + const count = Math.min(MAX_THREADS, state.queue.length + retriesRemaining); + for (let index = 0; index < count; index++) { + if (requests + 3 > MAX_REQUESTS || Date.now() >= deadline) break; + let thread; + if (state.queue.length) thread = state.queue.shift(); + else { + if (!retriesRemaining) break; + retriesRemaining--; + thread = state.retry.shift(); + } + // New activity refreshes the endpoint, not the retry's age or backoff. + const existing = state.retry.findIndex(retry => keyOf(retry) === keyOf(thread)); + if (existing >= 0) { + const previous = state.retry.splice(existing, 1)[0]; + thread = coalesce([previous, thread])[0]; + retriesRemaining = Math.min(retriesRemaining, state.retry.length); + } + if (thread.retryAfter > now()) { + state.retry.push(thread); + continue; + } + try { + const latest = thread.subject?.latest_comment_url; + const match = thread.subject?.url?.match(issueURL); + if (!commentURL.test(latest) || !match) continue; + const response = api(latest); + if (terminal.has(response.status)) continue; + const comment = response.body; + const timestamp = comment.updated_at || comment.submitted_at || comment.created_at || ''; + if (comment.user?.login !== operator || comment.user?.type !== 'User' || timestamp < (thread.pollSince || since)) continue; + const issueResponse = api(`repos/${match[1]}/issues/${match[3]}`); + if (terminal.has(issueResponse.status)) continue; + const issue = issueResponse.body; + if (issue.state !== 'open') continue; + if (issue.user?.login !== bot && !issue.assignees?.some(user => user.login === bot)) { + if (!board) { + const result = invoke(path.join(tools, 'bot-board'), ['list', '--json']); + if (result.status !== 0) throw new Error('notification board read failed'); + board = JSON.parse(result.stdout); + } + if (!board.some(item => item.content?.url === issue.html_url || (item.branch || '').split(' ').includes(issue.html_url))) continue; + } + const key = `${comment.html_url} ${timestamp}`; + if (!seen.has(key)) { seen.add(key); news.push(key); } + } catch (error) { + thread.retryStarted ??= now(); + thread.retryAttempts = (thread.retryAttempts || 0) + 1; + thread.retryAfter = now() + Math.min(MAX_RETRY_DELAY, RETRY_DELAY * 2 ** Math.min(thread.retryAttempts - 1, 6)); + state.retry.push(thread); + failed = true; + } + } + if (!state.queue.length && !state.next && state.until && state.visited.length) { + // Retries carry their own threads; they do not pin the acquisition cursor. + write('notification-since', state.until); + write('notification-modified', state.modified); + state.until = null; + state.visited = []; + } + } catch (error) { + failed = true; + console.error(`warning: ${error.message}; notification work retained`); + } finally { + boundRetries(); + let retained = JSON.stringify(state); + if (Buffer.byteLength(retained) > MAX_BYTES) { + fallback('notification retained state exceeds byte limit'); + retained = JSON.stringify(emptyState()); + } + write('notification-work', retained); + write('comment-seen', [...seen].sort().join('\n') + '\n'); + } + if (failed) console.error('warning: incomplete notification reads; retrying without suppressing verified news'); + return news; +} + +if (require.main === module) { + const [dir, tools, bot, operator, seconds] = process.argv.slice(2); + console.log(poll({ dir, tools, bot, operator, seconds: Number(seconds) }).join('\n')); +} + +module.exports = { poll, MAX_REQUESTS, MAX_RETRIES, MAX_BYTES, RETRY_LIFETIME }; diff --git a/bin/poll-notifications.sh b/bin/poll-notifications.sh new file mode 100644 index 0000000..a104b19 --- /dev/null +++ b/bin/poll-notifications.sh @@ -0,0 +1,9 @@ +# shellcheck shell=bash +# bot-poll-loop owns the state lock. Node bounds each subprocess and the batch. +# shellcheck disable=SC2154,SC2034 # caller supplies config and consumes news +notifications() { + local budget=10 + test -z "${until_actions}" || budget=$((max_wait_s - SECONDS)) + test "${budget}" -gt 0 || return 0 + notification_news=$(node "${BIN}/poll-notifications.js" "${S}" "${TOOLS}" "${OP_BOT_LOGIN}" "${OP_OPERATOR_LOGIN}" "${budget}") +} diff --git a/docs/scheduled-dispatcher.md b/docs/scheduled-dispatcher.md index 5785460..fa5ab03 100644 --- a/docs/scheduled-dispatcher.md +++ b/docs/scheduled-dispatcher.md @@ -70,6 +70,16 @@ its previous run, and the rest is recomputable from GitHub. ## Local supervisor +The existing poll loop checks notifications about once a minute (respecting +GitHub's X-Poll-Interval), with If-Modified-Since for unchanged polls. It +reads the latest comment's API author, not mentions or quoted text, and +wakes as `news, fast` for the configured operator on open threads authored +by or assigned to the bot, or on its board (including Branch PRs). This +read-only path never marks notifications read; local comment URL/timestamp +seen-sets dedupe repeats and edits. Failed reads retain the polling window. +A fresh state looks back ten minutes. The slower sweep remains the safety +net for missed notifications and comments hidden by a later reply. + Launch from the checkout with an interpreter: ``` diff --git a/tests/bot-poll-loop.sh b/tests/bot-poll-loop.sh index a4a2a1a..6b764b6 100755 --- a/tests/bot-poll-loop.sh +++ b/tests/bot-poll-loop.sh @@ -28,6 +28,32 @@ fail() { # of the events feed (the last one after that), as its --jq filter would. cat >"${WORK}/bin/gh" <<'EOF' #!/usr/bin/env bash +if test "$2" = user; then echo cgwalters-bot; exit 0; fi +if test "$2" = -i; then + if [[ "$3" == https://api.github.com/repos/* ]]; then + printf 'HTTP/2.0 200 OK\n\n'; cat "${FAKE}/comment.json"; exit 0 + fi + if [[ "$3" == repos/*/issues/* ]]; then + printf 'HTTP/2.0 200 OK\n\n'; cat "${FAKE}/issue.json"; exit 0 + fi + echo "$*" >>"${FAKE}/notification-calls" + if test -f "${FAKE}/notifications.json"; then + printf 'HTTP/2.0 200 OK\nDate: Fri, 09 Oct 2026 12:00:00 GMT\nLast-Modified: Fri, 09 Oct 2026 12:00:00 GMT\nX-Poll-Interval: 0\n\n' + cat "${FAKE}/notifications.json" + else + printf 'HTTP/2.0 304 Not Modified\nX-Poll-Interval: 0\n\n' + exit 1 + fi + exit 0 +fi +if [[ "$2" == https://api.github.com/* ]]; then + cat "${FAKE}/comment.json" + exit 0 +fi +if [[ "$2" == repos/*/issues/* ]]; then + cat "${FAKE}/issue.json" + exit 0 +fi if [[ "$2" =~ ^repos/([^/]+)/([^/]+)/pulls/([0-9]+)$ ]]; then cat "${FAKE}/author-${BASH_REMATCH[1]}-${BASH_REMATCH[2]}-${BASH_REMATCH[3]}" 2>/dev/null || echo someone exit 0 @@ -365,6 +391,41 @@ done test "$("${LOOP}" --help | head -n1)" = "Usage: bot-poll-loop [--until-actions [--max-wait DURATION]]" || fail "--help" +# Recorded API shapes: operator comments need neither mention nor ask label. +for tracked in author assigned board branch unrelated closed stranger bot old; do + reset + sweep 20261002-100000-000 "" + cp "${TESTS}/fixtures/poll-notifications.json" "${FAKE}/notifications.json" + mkdir -p "${WORK}/state" + echo '2026-10-09T11:50:00Z' >"${WORK}/state/notification-since" + printf '%s\n' '{"user":{"login":"cgwalters","type":"User"},"html_url":"https://github.com/o/r/issues/1#issuecomment-2","updated_at":"2026-10-09T11:59:00Z","body":"B"}' >"${FAKE}/comment.json" + jq -n --arg mode "${tracked}" '{state: (if $mode == "closed" then "closed" else "open" end), html_url: "https://github.com/o/r/issues/1", user: {login: (if $mode == "author" or $mode == "closed" then "cgwalters-bot" else "someone" end)}, assignees: (if $mode == "assigned" then [{login:"cgwalters-bot"}] else [] end)}' >"${FAKE}/issue.json" + if test "${tracked}" = board; then + echo '[{"content":{"url":"https://github.com/o/r/issues/1"}}]' >"${FAKE}/board.json" + fi + if test "${tracked}" = branch; then + echo '[{"content":{"url":"https://github.com/o/r/issues/9"},"branch":"https://github.com/o/r/issues/1"}]' >"${FAKE}/board.json" + fi + case "${tracked}" in + stranger|bot|old) + jq --arg mode "${tracked}" '.user.login = (if $mode == "stranger" then "someone" else "cgwalters" end) | .user.type = (if $mode == "bot" then "Bot" else "User" end) | .updated_at = (if $mode == "old" then "2026-10-01T00:00:00Z" else .updated_at end)' "${FAKE}/comment.json" >"${FAKE}/comment.tmp" + mv "${FAKE}/comment.tmp" "${FAKE}/comment.json" ;; + esac + out=$(loop) + case "${tracked}" in + unrelated|closed|stranger|bot|old) expect_first "${out}" '^QUIET' ;; + *) expect_first "${out}" '^ACTIONS \(news, fast\)' ;; + esac + expect_first "$(loop)" '^QUIET' + grep -q 'If-Modified-Since:' "${FAKE}/notification-calls" || fail "no conditional request" +done + +# The server's minimum interval survives a process restart. +before=$(wc -l <"${FAKE}/notification-calls") +echo "$(($(date +%s) + 300))" >"${WORK}/state/notification-due" +expect_first "$(loop)" '^QUIET' +test "$(wc -l <"${FAKE}/notification-calls")" = "${before}" || fail "poll interval lost on restart" + if test "${failures}" -gt 0; then echo "${failures} failure(s)" 1>&2 exit 1 diff --git a/tests/bot-supervisor.sh b/tests/bot-supervisor.sh index a30a38b..a64a1d2 100644 --- a/tests/bot-supervisor.sh +++ b/tests/bot-supervisor.sh @@ -1,4 +1,4 @@ #!/usr/bin/env bash # Offline supervisor tests; CI discovers this nonexecutable wrapper. set -euo pipefail -exec node --test "$(dirname "$0")/bot-supervisor.test.js" +exec node --test "$(dirname "$0")/bot-supervisor.test.js" "$(dirname "$0")/poll-notifications.test.js" diff --git a/tests/bot-supervisor.test.js b/tests/bot-supervisor.test.js index 9bb4ce1..fa0afd9 100644 --- a/tests/bot-supervisor.test.js +++ b/tests/bot-supervisor.test.js @@ -118,6 +118,15 @@ test("event wakes are selected separately from reconcile payload", () => { assert.match(c.calls().find((x) => x.cmd === "start").brief, /Selected kinds: approval\n/); }); +test("operator comment fast wake dispatches immediately with its URL", () => { + const out = "ACTIONS (news, fast) at 1234: operator comments: https://github.com/o/r/issues/1#issuecomment-2\nObserved: nothing\n"; + const c = setup({ polls: [{ out }] }); + succeeded(c.once()); + const brief = c.calls().find((x) => x.cmd === "start").brief; + assert.match(brief, /Selected kinds: news\n/); + assert.ok(brief.endsWith(out)); +}); + test("URL-less fired heartbeat actions use the payload too", () => { const c = setup({ polls: [{ out: report([{ kind: "heartbeat", fired: "new", url: null, do: "publish one" }], "actions: heartbeat") }] }); succeeded(c.once()); diff --git a/tests/fixtures/poll-notifications.json b/tests/fixtures/poll-notifications.json new file mode 100644 index 0000000..9978d69 --- /dev/null +++ b/tests/fixtures/poll-notifications.json @@ -0,0 +1,15 @@ +[ + { + "id": "123", + "reason": "author", + "unread": true, + "updated_at": "2026-10-09T11:59:00Z", + "subject": { + "type": "Issue", + "title": "Choose an option", + "url": "https://api.github.com/repos/o/r/issues/1", + "latest_comment_url": "https://api.github.com/repos/o/r/issues/comments/2" + }, + "repository": {"full_name": "o/r"} + } +] diff --git a/tests/poll-notifications.test.js b/tests/poll-notifications.test.js new file mode 100644 index 0000000..6b1ff16 --- /dev/null +++ b/tests/poll-notifications.test.js @@ -0,0 +1,233 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); +const { poll, MAX_REQUESTS, MAX_RETRIES, MAX_BYTES, RETRY_LIFETIME } = require('../bin/poll-notifications.js'); + +function fixture(t, handler, tick = 60 * 1000) { + const dir = fs.mkdtempSync(path.join(os.homedir(), 'notification-test-')); + t.after(() => fs.rmSync(dir, { recursive: true, force: true })); + fs.writeFileSync(path.join(dir, 'notification-since'), '2026-10-09T11:00:00Z'); + const calls = []; + let clock = Date.now(); + const run = (command, args, options) => { + assert.ok(options.timeout > 0 && options.timeout <= 2000); + assert.equal(options.maxBuffer, 1024 * 1024); + calls.push(args); + if (args[1] === 'user') return { status: 0, stdout: 'bot\n' }; + const { status = 200, body = [], link = '' } = handler(args[2]); + return { status: status === 200 ? 0 : 1, stdout: `HTTP/2.0 ${status} Result\nDate: Fri, 09 Oct 2026 12:00:00 GMT\nX-Poll-Interval: 0\nLink: ${link}\n\n${JSON.stringify(body)}` }; + }; + return { calls, dir, advance: ms => { clock += ms; }, poll: () => { + clock += tick; + return poll({ dir, tools: dir, bot: 'bot', operator: 'human', run, now: () => clock }); + }, state: () => JSON.parse(fs.readFileSync(path.join(dir, 'notification-work'))) }; +} + +const thread = id => ({ subject: { type: 'Issue', url: `https://api.github.com/repos/o/r/issues/${id}`, latest_comment_url: `https://api.github.com/repos/o/r/issues/comments/${id}` } }); +const comment = id => ({ user: { login: 'human', type: 'User' }, html_url: `https://github.com/o/r/issues/1#issuecomment-${id}`, updated_at: '2026-10-09T11:30:00Z' }); +const issue = { state: 'open', user: { login: 'bot' } }; + +for (const failure of [404, 410, 451, 500, 403, 429]) { + test(`mixed success and ${failure}, with recovery`, t => { + let recovered = false; + const f = fixture(t, url => { + if (url.startsWith('notifications?')) return { body: [thread(1), thread(2)] }; + if (url.endsWith('comments/2') && !recovered) return { status: failure }; + if (url.includes('comments/')) return { body: comment(url.split('/').pop()) }; + return { body: issue }; + }); + assert.equal(f.poll().length, 1); + assert.equal(f.state().retry.length, failure >= 500 || [403, 429].includes(failure) ? 1 : 0); + assert.equal(fs.readFileSync(path.join(f.dir, 'notification-since'), 'utf8'), '2026-10-09T12:00:00Z'); + recovered = true; + assert.equal(f.poll().length, [403, 429, 500].includes(failure) ? 1 : 0); + assert.equal(f.state().retry.length, 0); + assert.deepEqual(f.poll(), []); + }); +} + +test('over-budget feed resumes without advancing past unprocessed work', t => { + const f = fixture(t, url => url.startsWith('notifications?') ? { body: Array.from({ length: 100 }, (_, i) => thread(i)) } : { body: url.includes('comments/') ? comment(url.split('/').pop()) : issue }); + let delivered = f.poll().length; + assert.ok(delivered > 0 && delivered < 100); + assert.ok(f.calls.length <= MAX_REQUESTS); + assert.equal(fs.readFileSync(path.join(f.dir, 'notification-since'), 'utf8'), '2026-10-09T11:00:00Z'); + while (f.state().queue.length) delivered += f.poll().length; + assert.equal(delivered, 100); +}); + +test('repeated next link stops and retains the acquisition window', t => { + const next = 'https://api.github.com/notifications?page=2'; + const f = fixture(t, () => ({ body: [], link: `<${next}>; rel="next"` })); + assert.deepEqual(f.poll(), []); + assert.equal(f.calls.length, 3); + assert.equal(fs.readFileSync(path.join(f.dir, 'notification-since'), 'utf8'), '2026-10-09T11:00:00Z'); +}); + +for (const failure of [403, 429, 503]) { + test(`large ${failure} backlog cannot starve later pages or windows`, t => { + const next = 'https://api.github.com/notifications?page=2'; + let firstPages = 0; + let laterPages = 0; + const f = fixture(t, url => { + if (url.startsWith('notifications?')) { + firstPages++; + return firstPages === 1 + ? { body: Array.from({ length: 100 }, (_, i) => thread(i)), link: `<${next}>; rel="next"` } + : { body: [thread(102)] }; + } + if (url === next) { + laterPages++; + return { body: [thread(101)] }; + } + if (url.includes('comments/')) { + const id = Number(url.split('/').pop()); + return id < 100 ? { status: failure } : { body: { ...comment(id), updated_at: id === 102 ? '2026-10-09T12:01:00Z' : comment(id).updated_at } }; + } + return { body: issue }; + }); + const delivered = []; + for (let iteration = 0; iteration < 30; iteration++) { + const before = f.calls.length; + delivered.push(...f.poll()); + assert.ok(f.calls.length - before <= MAX_REQUESTS); + if (f.state().queue.length) { + assert.equal(fs.readFileSync(path.join(f.dir, 'notification-since'), 'utf8'), '2026-10-09T11:00:00Z'); + } + } + assert.equal(laterPages, 1); + assert.ok(firstPages > 1, 'acquisition continues into later windows'); + assert.deepEqual(delivered, [ + `${comment(101).html_url} ${comment(101).updated_at}`, + `${comment(102).html_url} 2026-10-09T12:01:00Z`, + ]); + assert.equal(f.state().retry.length, 100, 'repeated failures stay separate and deduplicated'); + assert.equal(f.state().queue.length, 0); + }); +} + +test('page cap resumes at the next page', t => { + let page = 0; + const f = fixture(t, () => ({ body: [], link: `; rel="next"` })); + f.poll(); + assert.equal(page, 4); + assert.equal(f.state().next, 'https://api.github.com/notifications?page=4'); + f.poll(); + assert.equal(page, 8); +}); + +test('timeout retains work and still publishes prior verified news', t => { + const f = fixture(t, url => { + if (url.startsWith('notifications?')) return { body: [thread(1), thread(2)] }; + if (url.endsWith('comments/2')) throw new Error('request timed out'); + return { body: url.includes('comments/') ? comment(1) : issue }; + }); + assert.equal(f.poll().length, 1); + assert.equal(f.state().retry.length, 1); +}); + +test('expired deadline makes no network requests', t => { + const f = fixture(t, () => { throw new Error('must not run'); }); + assert.deepEqual(poll({ dir: f.dir, tools: f.dir, bot: 'bot', operator: 'human', seconds: 0, run: () => { throw new Error('must not run'); } }), []); + assert.equal(f.state().queue.length, 0); +}); + +test('304 still retries transient work', t => { + let recovery = false; + const f = fixture(t, url => { + if (url.startsWith('notifications?')) return recovery ? { status: 304 } : { body: [thread(1)] }; + if (url.includes('comments/')) return recovery ? { body: comment(1) } : { status: 503 }; + return { body: issue }; + }); + assert.deepEqual(f.poll(), []); + recovery = true; + assert.equal(f.poll().length, 1); + assert.equal(f.state().retry.length, 0); +}); + +test('unavailable issue does not suppress another verified thread', t => { + const f = fixture(t, url => { + if (url.startsWith('notifications?')) return { body: [thread(1), thread(2)] }; + if (url.includes('comments/')) return { body: comment(url.split('/').pop()) }; + return url.endsWith('/2') ? { status: 404 } : { body: issue }; + }); + assert.equal(f.poll().length, 1); + assert.equal(f.state().retry.length, 0); +}); + +test('changing failing threads coalesce across windows without resetting backoff or age', t => { + let window = 0; + let reads = 0; + const f = fixture(t, url => { + if (url.startsWith('notifications?')) return { body: [{ ...thread(1), updated_at: String(++window) }] }; + reads++; + return { status: 403 }; + }, 1000); + for (let i = 0; i < 100; i++) { + f.poll(); + assert.equal(f.state().retry.length, 1); + assert.equal(f.state().retry[0].pollSince, '2026-10-09T11:00:00Z'); + assert.ok(fs.statSync(path.join(f.dir, 'notification-work')).size <= MAX_BYTES); + } + assert.equal(window, 100); + assert.equal(reads, 2, 'new activity does not bypass exponential backoff'); + f.advance(RETRY_LIFETIME); + f.poll(); + assert.equal(f.state().retry[0].retryAttempts, 1, 'expiry starts fresh work, not an immortal retry'); +}); + +test('distinct failing threads remain count bounded across windows', t => { + let window = 0; + const f = fixture(t, url => url.startsWith('notifications?') + ? { body: Array.from({ length: 10 }, (_, i) => thread(window++)) } + : { status: 503 }, 1000); + for (let i = 0; i < 100; i++) { + f.poll(); + assert.ok(f.state().retry.length <= MAX_RETRIES); + assert.ok(fs.statSync(path.join(f.dir, 'notification-work')).size <= MAX_BYTES); + } + assert.equal(f.state().retry.length, MAX_RETRIES); +}); + +test('oversized legacy state is discarded before parsing and reported as sweep fallback', t => { + const f = fixture(t, () => ({ status: 304 })); + fs.writeFileSync(path.join(f.dir, 'notification-work'), 'x'.repeat(MAX_BYTES + 1)); + const warnings = []; + t.mock.method(console, 'error', message => warnings.push(message)); + assert.deepEqual(f.poll(), []); + assert.equal(f.state().retry.length, 0); + assert.ok(warnings.some(message => message.includes('falling back to the full sweep'))); +}); + +test('retry lifetime expires even without further activity', t => { + let first = true; + const warnings = []; + t.mock.method(console, 'error', message => warnings.push(message)); + const f = fixture(t, url => { + if (url.startsWith('notifications?')) { + if (!first) return { status: 304 }; + first = false; + return { body: [thread(1)] }; + } + return { status: 403 }; + }); + f.poll(); + f.advance(RETRY_LIFETIME); + f.poll(); + assert.equal(f.state().retry.length, 0); + assert.ok(warnings.some(message => message.includes('falling back to the full sweep'))); +}); + +test('oversized acquired work is not persisted', t => { + const warnings = []; + t.mock.method(console, 'error', message => warnings.push(message)); + const f = fixture(t, url => url.startsWith('notifications?') + ? { body: [{ ...thread(1), extra: 'x'.repeat(MAX_BYTES) }] } + : { status: 503 }); + f.poll(); + assert.ok(fs.statSync(path.join(f.dir, 'notification-work')).size <= MAX_BYTES); + assert.equal(f.state().retry.length, 0); + assert.ok(warnings.some(message => message.includes('falling back to the full sweep'))); +});