diff --git a/.devcontainer/codespaces/README.md b/.devcontainer/codespaces/README.md index 50127ecf80a..41f595b8a25 100644 --- a/.devcontainer/codespaces/README.md +++ b/.devcontainer/codespaces/README.md @@ -86,6 +86,10 @@ If the Slack API fails, the turn still completes through the n8n resume URL. The worker does not export its dequeue or Slack credentials to OpenCode. Interactive sessions remove all worker-only credentials before they start. +An idle worker starts with a 3-second poll interval. After each empty dequeue, +it doubles the interval and limits it to 30 seconds. Work resets the interval +to 3 seconds. Queued turns run without a delay between them. + The log is at `/tmp/agent-worker.log`. The tmux session is `agent-worker`. To watch it, run `tmux attach -t agent-worker`. The worker does not start if a required secret or `$GITHUB_USER` is missing. diff --git a/.devcontainer/codespaces/agent-worker.mjs b/.devcontainer/codespaces/agent-worker.mjs index 995d83e0c37..adb4e652359 100644 --- a/.devcontainer/codespaces/agent-worker.mjs +++ b/.devcontainer/codespaces/agent-worker.mjs @@ -15,7 +15,8 @@ const GITHUB_USER = codespaceEnv('GITHUB_USER'); const BOX_ID = codespaceEnv('CODESPACE_NAME'); const ROOT = '/workspaces'; -const POLL_INTERVAL_MS = 3000; +const INITIAL_POLL_INTERVAL_MS = 3000; +const MAX_POLL_INTERVAL_MS = 30_000; const SLACK_UPDATE_INTERVAL_MS = 1500; const SLACK_TEXT_LIMIT = 3900; const OPENCODE_CONFIG_CONTENT = JSON.stringify({ @@ -31,6 +32,36 @@ function posNum(name, fallback) { return fallback; } +function nextIdlePollInterval(interval) { + return Math.min(interval * 2, MAX_POLL_INTERVAL_MS); +} + +export async function pollOnce( + interval, + { dequeueTurn = dequeue, handleTurn = handle, wait = sleep, logError = console.error } = {}, +) { + let turn; + try { + turn = await dequeueTurn(); + } catch (error) { + logError(`poll error: ${error.message}`); + await wait(interval); + return nextIdlePollInterval(interval); + } + if (!turn) { + await wait(interval); + return nextIdlePollInterval(interval); + } + try { + await handleTurn(turn); + return INITIAL_POLL_INTERVAL_MS; + } catch (error) { + logError(`poll error: ${error.message}`); + await wait(INITIAL_POLL_INTERVAL_MS); + return nextIdlePollInterval(INITIAL_POLL_INTERVAL_MS); + } +} + // This limit expires before n8n's Wait node so that the user receives a specific error. const TURN_TIMEOUT_MS = posNum('TURN_TIMEOUT_MS', 25 * 60_000); function turnTimeoutMessage(timeout) { @@ -328,6 +359,9 @@ export async function startSlackProgress( } async function handle(turn) { + console.log( + `${new Date().toISOString()} turn ${turn.turnId} by ${turn.author ?? 'unknown'}: ${turn.sessionId ? 'resume' : 'new'}`, + ); let result; let activeSessionId = turn.sessionId ?? ''; const progress = await startSlackProgress(turn); @@ -388,22 +422,11 @@ async function main() { 'SLACK_BOT_TOKEN is not set. Turns will complete without Slack progress updates.', ); - console.log(`agent-worker polling as ${GITHUB_USER} every ${POLL_INTERVAL_MS}ms`); - for (;;) { - try { - const turn = await dequeue(); - if (turn) { - console.log( - `${new Date().toISOString()} turn ${turn.turnId} by ${turn.author ?? 'unknown'}: ${turn.sessionId ? 'resume' : 'new'}`, - ); - await handle(turn); - continue; - } - } catch (error) { - console.error(`poll error: ${error.message}`); - } - await sleep(POLL_INTERVAL_MS); - } + console.log( + `agent-worker polling as ${GITHUB_USER} every ${INITIAL_POLL_INTERVAL_MS}-${MAX_POLL_INTERVAL_MS}ms`, + ); + let pollInterval = INITIAL_POLL_INTERVAL_MS; + for (;;) pollInterval = await pollOnce(pollInterval); } if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) await main(); diff --git a/.devcontainer/codespaces/agent-worker.test.mjs b/.devcontainer/codespaces/agent-worker.test.mjs index b0c8d3d5d51..c7a7e6a1f73 100644 --- a/.devcontainer/codespaces/agent-worker.test.mjs +++ b/.devcontainer/codespaces/agent-worker.test.mjs @@ -5,7 +5,7 @@ import { PassThrough } from 'node:stream'; import { test } from 'node:test'; import { setTimeout as sleep } from 'node:timers/promises'; -import { openCodeEnvironment, runOpenCode, startSlackProgress } from './agent-worker.mjs'; +import { openCodeEnvironment, pollOnce, runOpenCode, startSlackProgress } from './agent-worker.mjs'; const turn = { turnId: 'turn-1', @@ -60,6 +60,67 @@ function completedChild(lines) { return child; } +test('backs off idle polls and resets when work arrives', async () => { + let interval = 3000; + const waits = []; + const turns = [null, null, null, null, null, { turnId: '1' }, { turnId: '2' }]; + const handled = []; + for (let attempt = 0; attempt < 5; attempt++) { + interval = await pollOnce(interval, { + dequeueTurn: async () => turns.shift(), + handleTurn: async (turn) => handled.push(turn.turnId), + wait: async (delay) => waits.push(delay), + }); + } + assert.deepEqual(waits, [3000, 6000, 12_000, 24_000, 30_000]); + assert.equal(interval, 30_000); + + interval = await pollOnce(interval, { + dequeueTurn: async () => turns.shift(), + handleTurn: async (turn) => handled.push(turn.turnId), + wait: async (delay) => waits.push(delay), + }); + interval = await pollOnce(interval, { + dequeueTurn: async () => turns.shift(), + handleTurn: async (turn) => handled.push(turn.turnId), + wait: async (delay) => waits.push(delay), + }); + assert.equal(interval, 3000); + assert.deepEqual(handled, ['1', '2']); + assert.equal(waits.length, 5); +}); + +test('backs off after a dequeue error', async () => { + const waits = []; + const errors = []; + const interval = await pollOnce(3000, { + dequeueTurn: async () => { + throw new Error('unavailable'); + }, + wait: async (delay) => waits.push(delay), + logError: (error) => errors.push(error), + }); + assert.equal(interval, 6000); + assert.deepEqual(waits, [3000]); + assert.deepEqual(errors, ['poll error: unavailable']); +}); + +test('resets the interval before handling work', async () => { + const waits = []; + const errors = []; + const interval = await pollOnce(30_000, { + dequeueTurn: async () => ({ turnId: '1' }), + handleTurn: async () => { + throw new Error('failed'); + }, + wait: async (delay) => waits.push(delay), + logError: (error) => errors.push(error), + }); + assert.equal(interval, 6000); + assert.deepEqual(waits, [3000]); + assert.deepEqual(errors, ['poll error: failed']); +}); + test('streams tool progress but excludes reasoning', async () => { const slack = slackRecorder(); const progress = await startSlackProgress(turn, {