Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 21 additions & 2 deletions bin/bot-poll-loop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
}

Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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}"
Expand Down
208 changes: 208 additions & 0 deletions bin/poll-notifications.js
Original file line number Diff line number Diff line change
@@ -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 };
9 changes: 9 additions & 0 deletions bin/poll-notifications.sh
Original file line number Diff line number Diff line change
@@ -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}")
}
10 changes: 10 additions & 0 deletions docs/scheduled-dispatcher.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:

```
Expand Down
61 changes: 61 additions & 0 deletions tests/bot-poll-loop.sh
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion tests/bot-supervisor.sh
Original file line number Diff line number Diff line change
@@ -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"
9 changes: 9 additions & 0 deletions tests/bot-supervisor.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down
15 changes: 15 additions & 0 deletions tests/fixtures/poll-notifications.json
Original file line number Diff line number Diff line change
@@ -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"}
}
]
Loading
Loading