mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-12 02:19:43 +00:00
feat: Stream OpenCode session progress to Slack (no-changelog) (#37741)
This commit is contained in:
@@ -11,8 +11,9 @@ Playwright system deps, and Docker-in-Docker (for testcontainers and
|
||||
|
||||
## One-time setup (~5 min)
|
||||
|
||||
1. Add your key at [github.com/settings/codespaces](https://github.com/settings/codespaces)
|
||||
→ **New secret** → name `ANTHROPIC_API_KEY`, repository access `n8n-io/n8n`.
|
||||
1. Add provider keys at [github.com/settings/codespaces](https://github.com/settings/codespaces).
|
||||
Add `ANTHROPIC_API_KEY` for Claude Code. Add `OPENROUTER_API_KEY` for Flaky's
|
||||
OpenCode worker. Give both secrets access to `n8n-io/n8n`.
|
||||
(Alternative for Max subscriptions: `CLAUDE_CODE_OAUTH_TOKEN` from `claude setup-token`.)
|
||||
2. Give the GitHub CLI the codespace scope:
|
||||
|
||||
@@ -43,10 +44,12 @@ pnpm session rm # delete the codespace
|
||||
|
||||
## Agent worker (drive a session from n8n)
|
||||
|
||||
`agent-worker.mjs` lets an n8n workflow drive a Claude Code session on the
|
||||
`agent-worker.mjs` lets an n8n workflow drive an OpenCode session on the
|
||||
codespace. This is how Slack (the Flaky bot) steers a session that runs here.
|
||||
Each turn is one headless `claude -p`. The session state is on disk. So
|
||||
`--resume` continues a conversation across turns, and across a stop/start.
|
||||
Each turn runs `openrouter/openai/gpt-5.6-sol` through
|
||||
`opencode run --format json --auto`. The session state remains on disk. The
|
||||
returned session ID continues a conversation across turns and a Codespace
|
||||
restart.
|
||||
|
||||
**The worker polls outward. Nothing inbound is exposed.** GitHub sets every
|
||||
forwarded port to private on start. It gives no API to make a port public. So
|
||||
@@ -55,17 +58,27 @@ instead. It asks n8n for a turn addressed to this box's owner (`$GITHUB_USER`).
|
||||
It runs the turn. It sends the result to the turn's resume URL. It uses no
|
||||
tunnel, no open port, and no domain.
|
||||
|
||||
The worker starts on each container start (`postStartCommand`). It needs two
|
||||
secrets. Add them at
|
||||
The worker starts on each container start (`postStartCommand`). It needs three
|
||||
secrets and uses one optional secret. Add them at
|
||||
[github.com/settings/codespaces](https://github.com/settings/codespaces), the
|
||||
same way as `ANTHROPIC_API_KEY`:
|
||||
|
||||
- `AGENT_WORKER_TOKEN` — the shared bearer token. The worker sends it on each poll.
|
||||
- `N8N_DEQUEUE_URL` — the n8n webhook that returns a pending turn.
|
||||
- `OPENROUTER_API_KEY` — the model provider key for Sol.
|
||||
- `SLACK_BOT_TOKEN` — optional bot token for progress messages. It needs `chat:write` only.
|
||||
|
||||
The dequeue payload can include `slack.channel` and `slack.thread_ts`. The
|
||||
worker posts one placeholder in that thread. It coalesces completed tool calls.
|
||||
It updates the message at most once every 1.5 seconds. It does not send reasoning
|
||||
text. The worker replaces the placeholder with the final answer.
|
||||
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 Claude Code sessions unset all worker-only credentials before they start.
|
||||
|
||||
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
|
||||
secret or `$GITHUB_USER` is missing.
|
||||
required secret or `$GITHUB_USER` is missing.
|
||||
|
||||
A turn stops after about 25 minutes (`TURN_TIMEOUT_MS`). This limit is below the
|
||||
n8n Wait limit. So the worker reports a clear message before n8n reports a
|
||||
@@ -79,8 +92,8 @@ report back in — the turn's resume URL continues one waiting n8n execution and
|
||||
then spent, so nothing on the box can post to the thread unprompted. A session
|
||||
that backgrounds a build and signs off with "I'll verify once it finishes" is
|
||||
therefore describing something that cannot happen. The worker states this in
|
||||
`--append-system-prompt` on every turn (`turnContract`), together with a pointer
|
||||
to this file for the box-specific parts. This is only the n8n/Slack path: an
|
||||
each OpenCode prompt (`turnContract`), together with a pointer to this file for
|
||||
the box-specific parts. This is only the n8n/Slack path: an
|
||||
interactive session (`pnpm session`, tmux) is long-lived, so background work,
|
||||
monitors and notifications behave normally there.
|
||||
|
||||
|
||||
@@ -1,35 +1,27 @@
|
||||
#!/usr/bin/env node
|
||||
// Poll worker for per-turn Claude Code conversations on a codespace.
|
||||
// You cannot reach a codespace from outside. GitHub keeps forwarded ports
|
||||
// private. So this worker calls out. It polls n8n for a turn addressed to this
|
||||
// box's owner. It runs one `claude -p`. It sends the result to the turn's resume
|
||||
// URL. Every call is outbound HTTPS to n8n. It opens no inbound port.
|
||||
//
|
||||
// AGENT_WORKER_TOKEN=… N8N_DEQUEUE_URL=… node agent-worker.mjs
|
||||
//
|
||||
// Env:
|
||||
// N8N_DEQUEUE_URL n8n webhook that hands back one pending turn (required)
|
||||
// AGENT_WORKER_TOKEN shared bearer sent on every dequeue (required)
|
||||
// GITHUB_USER box owner's login; the bootstrap route for a new thread (see codespace-env.mjs)
|
||||
// CODESPACE_NAME stable box id; routes a thread back to the box holding its session (same source)
|
||||
// TURN_TIMEOUT_MS per-turn limit; keep below the n8n Wait limit (default 25 min)
|
||||
import { execFile } from 'node:child_process';
|
||||
// GitHub keeps Codespaces ports private, so inbound delivery is not available.
|
||||
import { spawn } from 'node:child_process';
|
||||
import { resolve as resolvePath, sep } from 'node:path';
|
||||
import { setTimeout as sleep } from 'node:timers/promises';
|
||||
import { pathToFileURL } from 'node:url';
|
||||
|
||||
import { codespaceEnv } from '../../scripts/codespace-env.mjs';
|
||||
|
||||
const DEQUEUE_URL = process.env.N8N_DEQUEUE_URL;
|
||||
const TOKEN = process.env.AGENT_WORKER_TOKEN;
|
||||
// Read both identities from the codespace. tmux can give an empty copy of either.
|
||||
const SLACK_TOKEN = process.env.SLACK_BOT_TOKEN;
|
||||
// tmux can retain empty identity values, but the Codespaces files stay current.
|
||||
const GITHUB_USER = codespaceEnv('GITHUB_USER');
|
||||
const BOX_ID = codespaceEnv('CODESPACE_NAME');
|
||||
const ROOT = '/workspaces';
|
||||
|
||||
const POLL_INTERVAL_MS = 3000;
|
||||
const SLACK_UPDATE_INTERVAL_MS = 1500;
|
||||
const SLACK_TEXT_LIMIT = 3900;
|
||||
const OPENCODE_CONFIG_CONTENT = JSON.stringify({
|
||||
provider: { openrouter: { options: { apiKey: '{env:OPENROUTER_API_KEY}' } } },
|
||||
});
|
||||
|
||||
// Warn and use the default on a bad value, so a config mistake cannot silently
|
||||
// disable the turn limit.
|
||||
function posNum(name, fallback) {
|
||||
const raw = process.env[name];
|
||||
if (raw === undefined) return fallback;
|
||||
@@ -39,46 +31,35 @@ function posNum(name, fallback) {
|
||||
return fallback;
|
||||
}
|
||||
|
||||
// Keep this below the n8n Wait-node limit. Then the worker reports a slow turn
|
||||
// before n8n's Wait ends with a generic message.
|
||||
// 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);
|
||||
|
||||
for (const [k, v] of Object.entries({
|
||||
N8N_DEQUEUE_URL: DEQUEUE_URL,
|
||||
AGENT_WORKER_TOKEN: TOKEN,
|
||||
GITHUB_USER,
|
||||
})) {
|
||||
if (!v) {
|
||||
console.error(`Refusing to start: ${k} is not set.`);
|
||||
process.exit(1);
|
||||
}
|
||||
function turnTimeoutMessage(timeout) {
|
||||
const duration = timeout % 60_000 === 0 ? `${timeout / 60_000}-minute` : `${timeout}-millisecond`;
|
||||
return `The turn passed the ${duration} limit and stopped. It may have been in a build. Do a smaller step, or run a long build in its own turn.`;
|
||||
}
|
||||
|
||||
// Not fatal, but box pinning needs it: without a box id every turn routes by
|
||||
// owner, so a thread cannot follow the box holding its session.
|
||||
if (!BOX_ID)
|
||||
console.error(
|
||||
'CODESPACE_NAME did not resolve — box pinning disabled; turns route by githubUser only.',
|
||||
);
|
||||
export function openCodeEnvironment(environment) {
|
||||
const childEnvironment = { ...environment };
|
||||
delete childEnvironment.AGENT_WORKER_TOKEN;
|
||||
delete childEnvironment.N8N_DEQUEUE_URL;
|
||||
delete childEnvironment.SLACK_BOT_TOKEN;
|
||||
childEnvironment.OPENCODE_CONFIG_CONTENT = OPENCODE_CONFIG_CONTENT;
|
||||
return childEnvironment;
|
||||
}
|
||||
|
||||
// A turn gets a copy of this environment. Put the correct values in it, because
|
||||
// `pnpm dev:up` and `gh -c $CODESPACE_NAME` in the session need them.
|
||||
const TURN_ENV = { ...process.env };
|
||||
// OpenCode does not need the credentials that control the broker.
|
||||
const TURN_ENV = openCodeEnvironment(process.env);
|
||||
if (BOX_ID) TURN_ENV.CODESPACE_NAME = BOX_ID;
|
||||
if (GITHUB_USER) TURN_ENV.GITHUB_USER = GITHUB_USER;
|
||||
|
||||
const CODESPACE_DOCS = '.devcontainer/codespaces/README.md';
|
||||
|
||||
// A session cannot be told any of this after its final message, and a system
|
||||
// prompt is not part of the resumed transcript, so send it on every turn. State
|
||||
// the turn's hard limits inline: a session that has to read a file to learn them
|
||||
// can reply before it gets there. Point at the box docs for the rest — they
|
||||
// already cover dev:up, ports, and build cost, and AGENTS.md does not.
|
||||
// The worker has no later turn, so each prompt must define the atomic runtime contract.
|
||||
function turnContract(author) {
|
||||
return [
|
||||
'# Your runtime',
|
||||
'You are one turn of a Slack thread, driven by an n8n workflow that runs you as a headless',
|
||||
'`claude -p` on a GitHub codespace. Your final message is the reply that reaches Slack, so keep',
|
||||
'OpenCode session on a GitHub codespace. Your final message is the reply that reaches Slack, so keep',
|
||||
'it short and skip heavy markdown.',
|
||||
author ? `You are replying to ${author}.` : '',
|
||||
'',
|
||||
@@ -102,38 +83,117 @@ function turnContract(author) {
|
||||
.join('\n');
|
||||
}
|
||||
|
||||
function runClaude({ message, sessionId, cwd, author }) {
|
||||
function safeCwd(cwd) {
|
||||
const safeCwd = resolvePath(typeof cwd === 'string' && cwd ? cwd : `${ROOT}/n8n`);
|
||||
if (safeCwd !== ROOT && !safeCwd.startsWith(ROOT + sep))
|
||||
throw new Error(`cwd must be under ${ROOT}`);
|
||||
const args = [
|
||||
'-p',
|
||||
'--output-format',
|
||||
'json',
|
||||
'--dangerously-skip-permissions',
|
||||
'--append-system-prompt',
|
||||
turnContract(typeof author === 'string' ? author : ''),
|
||||
];
|
||||
if (sessionId) args.push('--resume', sessionId);
|
||||
args.push(message);
|
||||
return new Promise((res, rej) => {
|
||||
execFile(
|
||||
'claude',
|
||||
args,
|
||||
{ cwd: safeCwd, env: TURN_ENV, timeout: TURN_TIMEOUT_MS, maxBuffer: 64 * 1024 * 1024 },
|
||||
(err, stdout, stderr) => {
|
||||
try {
|
||||
res(JSON.parse(stdout));
|
||||
} catch {
|
||||
if (err?.killed)
|
||||
rej(
|
||||
new Error(
|
||||
`The turn passed the ${Math.round(TURN_TIMEOUT_MS / 60_000)}-minute limit and stopped. It may have been in a build. Do a smaller step, or run a long build in its own turn.`,
|
||||
),
|
||||
);
|
||||
else rej(new Error(stderr?.trim() || err?.message || 'claude produced no output'));
|
||||
}
|
||||
},
|
||||
return safeCwd;
|
||||
}
|
||||
|
||||
function stopProcessTree(child, signal) {
|
||||
if (!child.pid) return;
|
||||
try {
|
||||
process.kill(-child.pid, signal);
|
||||
} catch {
|
||||
child.kill(signal);
|
||||
}
|
||||
}
|
||||
|
||||
function eventError(event) {
|
||||
return event.error?.data?.message ?? event.error?.message ?? event.error?.name;
|
||||
}
|
||||
|
||||
export function runOpenCode(
|
||||
{ message, sessionId, cwd, author },
|
||||
onEvent,
|
||||
onSession,
|
||||
{
|
||||
spawnProcess = spawn,
|
||||
stopProcess = stopProcessTree,
|
||||
timeout = TURN_TIMEOUT_MS,
|
||||
killDelay = 5000,
|
||||
} = {},
|
||||
) {
|
||||
const directory = safeCwd(cwd);
|
||||
const args = ['run', '--model', 'openrouter/openai/gpt-5.6-sol', '--format', 'json', '--auto'];
|
||||
if (sessionId) args.push('--session', sessionId);
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const child = spawnProcess('opencode', args, {
|
||||
cwd: directory,
|
||||
detached: true,
|
||||
env: TURN_ENV,
|
||||
stdio: ['pipe', 'pipe', 'pipe'],
|
||||
});
|
||||
let activeSessionId = sessionId ?? '';
|
||||
let buffer = '';
|
||||
let stderr = '';
|
||||
let error = '';
|
||||
const text = [];
|
||||
let timedOut = false;
|
||||
let forceTimer;
|
||||
const timer = setTimeout(() => {
|
||||
timedOut = true;
|
||||
stopProcess(child, 'SIGTERM');
|
||||
forceTimer = setTimeout(() => stopProcess(child, 'SIGKILL'), killDelay);
|
||||
forceTimer.unref?.();
|
||||
}, timeout);
|
||||
|
||||
const consume = (line) => {
|
||||
if (!line.trim()) return;
|
||||
let event;
|
||||
try {
|
||||
event = JSON.parse(line);
|
||||
} catch {
|
||||
console.error(`OpenCode output ignored: ${line.slice(0, 200)}`);
|
||||
return;
|
||||
}
|
||||
if (!activeSessionId && typeof event.sessionID === 'string') {
|
||||
activeSessionId = event.sessionID;
|
||||
onSession(activeSessionId);
|
||||
}
|
||||
if (event.sessionID !== activeSessionId) return;
|
||||
onEvent(event);
|
||||
if (event.type === 'text' && typeof event.part?.text === 'string') text.push(event.part.text);
|
||||
if (event.type === 'error') error = eventError(event) || 'OpenCode failed';
|
||||
};
|
||||
|
||||
child.stdout.setEncoding('utf8');
|
||||
child.stdout.on('data', (chunk) => {
|
||||
buffer += chunk;
|
||||
const lines = buffer.split(/\r?\n/);
|
||||
buffer = lines.pop() ?? '';
|
||||
for (const line of lines) consume(line);
|
||||
});
|
||||
child.stderr.setEncoding('utf8');
|
||||
child.stderr.on('data', (chunk) => (stderr = `${stderr}${chunk}`.slice(-4000)));
|
||||
child.once('error', (processError) => {
|
||||
clearTimeout(timer);
|
||||
if (forceTimer) clearTimeout(forceTimer);
|
||||
reject(processError);
|
||||
});
|
||||
child.once('close', (code) => {
|
||||
clearTimeout(timer);
|
||||
if (forceTimer) clearTimeout(forceTimer);
|
||||
consume(buffer);
|
||||
if (timedOut) {
|
||||
reject(new Error(turnTimeoutMessage(timeout)));
|
||||
return;
|
||||
}
|
||||
if (code !== 0 || error) {
|
||||
const failureMessage = error || stderr.trim() || `OpenCode exited with ${code}`;
|
||||
reject(new Error(failureMessage));
|
||||
return;
|
||||
}
|
||||
if (!activeSessionId) {
|
||||
reject(new Error('OpenCode did not return a session id'));
|
||||
return;
|
||||
}
|
||||
resolve({ result: text.join('\n').trim(), session_id: activeSessionId });
|
||||
});
|
||||
|
||||
child.stdin.end(
|
||||
`${turnContract(typeof author === 'string' ? author : '')}\n\n# Request\n${message}`,
|
||||
);
|
||||
});
|
||||
}
|
||||
@@ -151,20 +211,133 @@ async function dequeue() {
|
||||
const res = await post(DEQUEUE_URL, { githubUser: GITHUB_USER, boxId: BOX_ID, token: TOKEN });
|
||||
if (!res.ok) throw new Error(`dequeue HTTP ${res.status}`);
|
||||
const text = await res.text();
|
||||
if (!text.trim()) return null; // no pending turn
|
||||
if (!text.trim()) return null;
|
||||
const turn = JSON.parse(text);
|
||||
return turn?.turnId ? turn : null;
|
||||
}
|
||||
|
||||
async function slackApi(method, body) {
|
||||
const res = await fetch(`https://slack.com/api/${method}`, {
|
||||
method: 'POST',
|
||||
headers: { authorization: `Bearer ${SLACK_TOKEN}`, 'content-type': 'application/json' },
|
||||
body: JSON.stringify(body),
|
||||
signal: AbortSignal.timeout(20_000),
|
||||
});
|
||||
const result = await res.json();
|
||||
if (!res.ok || !result.ok) throw new Error(result.error || `HTTP ${res.status}`);
|
||||
return result;
|
||||
}
|
||||
|
||||
function progressText(tools) {
|
||||
const progress = [...tools.values()].slice(-6).map(({ status, title }) => {
|
||||
if (status === 'error') return `Failed: ${title}`;
|
||||
return `Done: ${title}`;
|
||||
});
|
||||
return (progress.length ? progress.join('\n') : 'Flaky is working…').slice(0, SLACK_TEXT_LIMIT);
|
||||
}
|
||||
|
||||
function finalSlackText(text, sessionId, boxId) {
|
||||
const metadata = [sessionId && `⟳session:${sessionId}`, boxId && `⟳box:${boxId}`]
|
||||
.filter(Boolean)
|
||||
.join(' ');
|
||||
const suffix = metadata ? `\n\n${metadata}` : '';
|
||||
const body = text || 'Flaky completed the turn';
|
||||
return `${body.slice(0, SLACK_TEXT_LIMIT - suffix.length)}${suffix}`;
|
||||
}
|
||||
|
||||
const NO_SLACK_PROGRESS = { event() {}, async finish() {} };
|
||||
|
||||
export async function startSlackProgress(
|
||||
turn,
|
||||
{
|
||||
callSlack = SLACK_TOKEN ? slackApi : undefined,
|
||||
updateInterval = SLACK_UPDATE_INTERVAL_MS,
|
||||
} = {},
|
||||
) {
|
||||
const channel = turn.slack?.channel;
|
||||
const threadTs = turn.slack?.thread_ts;
|
||||
if (!callSlack || typeof channel !== 'string' || typeof threadTs !== 'string')
|
||||
return NO_SLACK_PROGRESS;
|
||||
|
||||
let message;
|
||||
try {
|
||||
message = await callSlack('chat.postMessage', {
|
||||
channel,
|
||||
thread_ts: threadTs,
|
||||
text: 'Flaky is working…',
|
||||
});
|
||||
} catch (error) {
|
||||
console.error(`turn ${turn.turnId}: Slack placeholder failed: ${error.message}`);
|
||||
return NO_SLACK_PROGRESS;
|
||||
}
|
||||
|
||||
const tools = new Map();
|
||||
let timer;
|
||||
let inFlight;
|
||||
let pending;
|
||||
let lastUpdate = 0;
|
||||
let stopped = false;
|
||||
|
||||
const update = (text) =>
|
||||
callSlack('chat.update', { channel, ts: message.ts, text }).catch((error) =>
|
||||
console.error(`turn ${turn.turnId}: Slack update failed: ${error.message}`),
|
||||
);
|
||||
const flush = () => {
|
||||
if (stopped || timer || inFlight || pending === undefined) return;
|
||||
const delay = Math.max(0, lastUpdate + updateInterval - Date.now());
|
||||
timer = setTimeout(() => {
|
||||
timer = undefined;
|
||||
const text = pending;
|
||||
pending = undefined;
|
||||
lastUpdate = Date.now();
|
||||
inFlight = update(text).finally(() => {
|
||||
inFlight = undefined;
|
||||
flush();
|
||||
});
|
||||
}, delay);
|
||||
};
|
||||
const schedule = () => {
|
||||
pending = progressText(tools);
|
||||
flush();
|
||||
};
|
||||
|
||||
return {
|
||||
event(event) {
|
||||
if (!event.part?.id) return;
|
||||
if (event.type === 'tool_use') {
|
||||
tools.set(event.part.id, {
|
||||
status: event.part.state?.status,
|
||||
title: event.part.state?.title || event.part.tool,
|
||||
});
|
||||
schedule();
|
||||
}
|
||||
},
|
||||
async finish(text, sessionId, boxId) {
|
||||
stopped = true;
|
||||
pending = undefined;
|
||||
if (timer) {
|
||||
clearTimeout(timer);
|
||||
timer = undefined;
|
||||
}
|
||||
if (inFlight) await inFlight;
|
||||
const delay = Math.max(0, lastUpdate + updateInterval - Date.now());
|
||||
if (delay) await sleep(delay);
|
||||
await update(finalSlackText(text, sessionId, boxId));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function handle(turn) {
|
||||
let result;
|
||||
let activeSessionId = turn.sessionId ?? '';
|
||||
const progress = await startSlackProgress(turn);
|
||||
try {
|
||||
const r = await runClaude(turn);
|
||||
const r = await runOpenCode(turn, progress.event, (sessionId) => (activeSessionId = sessionId));
|
||||
result = {
|
||||
turnId: turn.turnId,
|
||||
status: 'done',
|
||||
output: r.result ?? '',
|
||||
sessionId: r.session_id ?? turn.sessionId ?? '',
|
||||
output: r.result,
|
||||
sessionId: r.session_id,
|
||||
boxId: BOX_ID,
|
||||
};
|
||||
} catch (error) {
|
||||
@@ -172,13 +345,12 @@ async function handle(turn) {
|
||||
turnId: turn.turnId,
|
||||
status: 'error',
|
||||
output: error.message,
|
||||
sessionId: turn.sessionId ?? '',
|
||||
sessionId: activeSessionId,
|
||||
boxId: BOX_ID,
|
||||
};
|
||||
}
|
||||
// Send the result to the turn's resume URL. This continues the waiting n8n
|
||||
// execution. Retry on a failed status or a network error, so a transient
|
||||
// failure does not drop the result and leave the turn to time out.
|
||||
await progress.finish(result.output, result.sessionId, result.boxId);
|
||||
// The resume POST is the delivery contract. Retry transient failures.
|
||||
for (let attempt = 1; attempt <= 3; attempt++) {
|
||||
try {
|
||||
const res = await post(turn.resumeUrl, result);
|
||||
@@ -194,19 +366,44 @@ async function handle(turn) {
|
||||
console.error(`turn ${turn.turnId}: result not delivered after 3 attempts`);
|
||||
}
|
||||
|
||||
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; // get the next turn now, with no delay
|
||||
async function main() {
|
||||
for (const [key, value] of Object.entries({
|
||||
N8N_DEQUEUE_URL: DEQUEUE_URL,
|
||||
AGENT_WORKER_TOKEN: TOKEN,
|
||||
OPENROUTER_API_KEY: process.env.OPENROUTER_API_KEY,
|
||||
GITHUB_USER,
|
||||
})) {
|
||||
if (!value) {
|
||||
console.error(`Refusing to start: ${key} is not set.`);
|
||||
process.exit(1);
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`poll error: ${error.message}`);
|
||||
}
|
||||
await sleep(POLL_INTERVAL_MS);
|
||||
|
||||
if (!BOX_ID)
|
||||
console.error(
|
||||
'CODESPACE_NAME did not resolve — box pinning disabled; turns route by githubUser only.',
|
||||
);
|
||||
if (!SLACK_TOKEN)
|
||||
console.error(
|
||||
'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);
|
||||
}
|
||||
}
|
||||
|
||||
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) await main();
|
||||
|
||||
@@ -0,0 +1,273 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import { EventEmitter } from 'node:events';
|
||||
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';
|
||||
|
||||
const turn = {
|
||||
turnId: 'turn-1',
|
||||
slack: { channel: 'C123', thread_ts: '123.456' },
|
||||
};
|
||||
|
||||
function slackRecorder() {
|
||||
const calls = [];
|
||||
return {
|
||||
calls,
|
||||
async call(method, body) {
|
||||
calls.push({ method, body });
|
||||
return method === 'chat.postMessage' ? { ok: true, ts: '456.789' } : { ok: true };
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function waitFor(check) {
|
||||
for (let attempt = 0; attempt < 100; attempt++) {
|
||||
if (check()) return;
|
||||
await sleep(5);
|
||||
}
|
||||
throw new Error('Condition was not met');
|
||||
}
|
||||
|
||||
function childProcess() {
|
||||
const child = new EventEmitter();
|
||||
child.pid = 123;
|
||||
child.stdin = new PassThrough();
|
||||
child.stdout = new PassThrough();
|
||||
child.stderr = new PassThrough();
|
||||
child.kill = () => {};
|
||||
return child;
|
||||
}
|
||||
|
||||
function completedChild(lines) {
|
||||
const child = childProcess();
|
||||
setImmediate(() => {
|
||||
for (const line of lines) child.stdout.write(`${line}\n`);
|
||||
child.stdout.end();
|
||||
child.stderr.end();
|
||||
child.emit('close', 0);
|
||||
});
|
||||
return child;
|
||||
}
|
||||
|
||||
test('streams tool progress but excludes reasoning', async () => {
|
||||
const slack = slackRecorder();
|
||||
const progress = await startSlackProgress(turn, {
|
||||
callSlack: slack.call,
|
||||
updateInterval: 50,
|
||||
});
|
||||
|
||||
progress.event({
|
||||
type: 'reasoning',
|
||||
sessionID: 'ses_1',
|
||||
part: { id: 'prt_reason', type: 'reasoning', text: 'private reasoning' },
|
||||
});
|
||||
progress.event({
|
||||
type: 'tool_use',
|
||||
sessionID: 'ses_1',
|
||||
part: {
|
||||
id: 'prt_tool',
|
||||
type: 'tool',
|
||||
tool: 'read',
|
||||
state: { status: 'completed', title: 'Read worker source' },
|
||||
},
|
||||
});
|
||||
await sleep(100);
|
||||
|
||||
const streamed = slack.calls.find((call) => call.method === 'chat.update')?.body.text;
|
||||
assert.match(streamed, /Done: Read worker source/);
|
||||
assert.doesNotMatch(streamed, /private reasoning/);
|
||||
});
|
||||
|
||||
test('coalesces Slack updates within the configured interval', async () => {
|
||||
const slack = slackRecorder();
|
||||
const progress = await startSlackProgress(turn, {
|
||||
callSlack: slack.call,
|
||||
updateInterval: 50,
|
||||
});
|
||||
|
||||
for (const [id, title] of ['One', 'Two', 'Three'].entries()) {
|
||||
progress.event({
|
||||
type: 'tool_use',
|
||||
sessionID: 'ses_1',
|
||||
part: {
|
||||
id: `prt_${id}`,
|
||||
type: 'tool',
|
||||
tool: 'read',
|
||||
state: { status: 'completed', title },
|
||||
},
|
||||
});
|
||||
}
|
||||
await sleep(100);
|
||||
assert.equal(slack.calls.filter((call) => call.method === 'chat.update').length, 1);
|
||||
|
||||
progress.event({
|
||||
type: 'tool_use',
|
||||
sessionID: 'ses_1',
|
||||
part: {
|
||||
id: 'prt_4',
|
||||
type: 'tool',
|
||||
tool: 'read',
|
||||
state: { status: 'completed', title: 'Four' },
|
||||
},
|
||||
});
|
||||
assert.equal(slack.calls.filter((call) => call.method === 'chat.update').length, 1);
|
||||
await sleep(100);
|
||||
assert.equal(slack.calls.filter((call) => call.method === 'chat.update').length, 2);
|
||||
});
|
||||
|
||||
test('replaces progress with the final answer', async () => {
|
||||
const calls = [];
|
||||
let releaseProgress;
|
||||
const callSlack = async (method, body) => {
|
||||
calls.push({ method, body });
|
||||
if (method === 'chat.postMessage') return { ok: true, ts: '456.789' };
|
||||
if (calls.filter((call) => call.method === 'chat.update').length === 1) {
|
||||
return await new Promise((resolve) => (releaseProgress = () => resolve({ ok: true })));
|
||||
}
|
||||
return { ok: true };
|
||||
};
|
||||
const progress = await startSlackProgress(turn, {
|
||||
callSlack,
|
||||
updateInterval: 0,
|
||||
});
|
||||
progress.event({
|
||||
type: 'tool_use',
|
||||
sessionID: 'ses_1',
|
||||
part: {
|
||||
id: 'prt_tool',
|
||||
type: 'tool',
|
||||
tool: 'read',
|
||||
state: { status: 'completed', title: 'Read worker source' },
|
||||
},
|
||||
});
|
||||
await waitFor(() => calls.filter((call) => call.method === 'chat.update').length === 1);
|
||||
for (const title of ['Second tool', 'Third tool', 'Fourth tool']) {
|
||||
progress.event({
|
||||
type: 'tool_use',
|
||||
sessionID: 'ses_1',
|
||||
part: {
|
||||
id: title,
|
||||
type: 'tool',
|
||||
tool: 'read',
|
||||
state: { status: 'completed', title },
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const finish = progress.finish('Final answer', 'ses_1', 'box-1');
|
||||
await sleep(10);
|
||||
assert.equal(calls.filter((call) => call.method === 'chat.update').length, 1);
|
||||
releaseProgress();
|
||||
await finish;
|
||||
|
||||
assert.deepEqual(calls.at(-1), {
|
||||
method: 'chat.update',
|
||||
body: {
|
||||
channel: 'C123',
|
||||
ts: '456.789',
|
||||
text: 'Final answer\n\n⟳session:ses_1 ⟳box:box-1',
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
test('streams OpenCode CLI events and resumes an OpenCode session', async () => {
|
||||
let invocation;
|
||||
let prompt = '';
|
||||
const events = [];
|
||||
const result = await runOpenCode(
|
||||
{
|
||||
message: 'Test message',
|
||||
sessionId: 'ses_existing',
|
||||
cwd: '/workspaces/n8n',
|
||||
author: 'Tester',
|
||||
},
|
||||
(event) => events.push(event),
|
||||
() => {},
|
||||
{
|
||||
spawnProcess(command, args, options) {
|
||||
invocation = { command, args, options };
|
||||
const child = completedChild([
|
||||
JSON.stringify({ type: 'step_start', sessionID: 'ses_existing', part: {} }),
|
||||
JSON.stringify({
|
||||
type: 'tool_use',
|
||||
sessionID: 'ses_existing',
|
||||
part: { id: 'prt_1', type: 'tool', tool: 'read', state: { status: 'completed' } },
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: 'text',
|
||||
sessionID: 'ses_existing',
|
||||
part: { id: 'prt_2', type: 'text', text: 'Completed response' },
|
||||
}),
|
||||
]);
|
||||
child.stdin.on('data', (chunk) => (prompt += chunk));
|
||||
return child;
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
assert.equal(result.result, 'Completed response');
|
||||
assert.equal(result.session_id, 'ses_existing');
|
||||
assert.deepEqual(invocation.args, [
|
||||
'run',
|
||||
'--model',
|
||||
'openrouter/openai/gpt-5.6-sol',
|
||||
'--format',
|
||||
'json',
|
||||
'--auto',
|
||||
'--session',
|
||||
'ses_existing',
|
||||
]);
|
||||
assert.equal(invocation.options.detached, true);
|
||||
assert.match(prompt, /# Request\nTest message/);
|
||||
assert.deepEqual(
|
||||
events.map((event) => event.type),
|
||||
['step_start', 'tool_use', 'text'],
|
||||
);
|
||||
});
|
||||
|
||||
test('stops the OpenCode process group at the turn limit', async () => {
|
||||
const signals = [];
|
||||
let child;
|
||||
const run = runOpenCode(
|
||||
{ message: 'Test message', cwd: '/workspaces/n8n', author: 'Tester' },
|
||||
() => {},
|
||||
() => {},
|
||||
{
|
||||
spawnProcess() {
|
||||
child = childProcess();
|
||||
return child;
|
||||
},
|
||||
stopProcess(_child, signal) {
|
||||
signals.push(signal);
|
||||
if (signal === 'SIGTERM') setImmediate(() => child.emit('close', null));
|
||||
},
|
||||
killDelay: 5,
|
||||
timeout: 1,
|
||||
},
|
||||
);
|
||||
|
||||
await assert.rejects(run, /The turn passed the 1-millisecond limit/);
|
||||
assert.deepEqual(signals, ['SIGTERM']);
|
||||
});
|
||||
|
||||
test('keeps broker credentials out of the OpenCode process', () => {
|
||||
const environment = openCodeEnvironment({
|
||||
AGENT_WORKER_TOKEN: 'worker',
|
||||
N8N_DEQUEUE_URL: 'https://example.com',
|
||||
SLACK_BOT_TOKEN: 'slack',
|
||||
ANTHROPIC_API_KEY: 'model',
|
||||
OPENROUTER_API_KEY: 'openrouter',
|
||||
GITHUB_TOKEN: 'github',
|
||||
});
|
||||
assert.deepEqual(JSON.parse(environment.OPENCODE_CONFIG_CONTENT), {
|
||||
provider: { openrouter: { options: { apiKey: '{env:OPENROUTER_API_KEY}' } } },
|
||||
});
|
||||
delete environment.OPENCODE_CONFIG_CONTENT;
|
||||
assert.deepEqual(environment, {
|
||||
ANTHROPIC_API_KEY: 'model',
|
||||
OPENROUTER_API_KEY: 'openrouter',
|
||||
GITHUB_TOKEN: 'github',
|
||||
});
|
||||
});
|
||||
@@ -38,12 +38,18 @@
|
||||
"ANTHROPIC_API_KEY": {
|
||||
"description": "Anthropic API key for Claude Code / OpenCode sessions"
|
||||
},
|
||||
"OPENROUTER_API_KEY": {
|
||||
"description": "OpenRouter API key for Flaky's OpenCode sessions"
|
||||
},
|
||||
"AGENT_WORKER_TOKEN": {
|
||||
"description": "The shared bearer token. The poll worker sends it to n8n on each dequeue. See .devcontainer/codespaces/agent-worker.mjs."
|
||||
},
|
||||
"N8N_DEQUEUE_URL": {
|
||||
"description": "n8n webhook the poll worker calls to claim a pending turn"
|
||||
},
|
||||
"SLACK_BOT_TOKEN": {
|
||||
"description": "Slack bot token for agent progress messages. It needs the chat:write scope."
|
||||
},
|
||||
"CLAUDE_CODE_OAUTH_TOKEN": {
|
||||
"description": "Alternative to an API key: token from `claude setup-token` (Max subscription)"
|
||||
},
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
},
|
||||
"packageManager": "[email protected]",
|
||||
"scripts": {
|
||||
"test": "node --test --experimental-test-module-mocks ./*.test.mjs ./docker/*.test.mjs ./owners/*.test.mjs ./quality/*.test.mjs ./slack/*.test.mjs ./stale/*.test.mjs ../../scripts/licenses/*.test.mjs ../../scripts/mutation-health/*.test.mjs",
|
||||
"test": "node --test --experimental-test-module-mocks ./*.test.mjs ./docker/*.test.mjs ./owners/*.test.mjs ./quality/*.test.mjs ./slack/*.test.mjs ./stale/*.test.mjs ../../.devcontainer/codespaces/*.test.mjs ../../scripts/licenses/*.test.mjs ../../scripts/mutation-health/*.test.mjs",
|
||||
"generate-sbom": "FETCH_LICENSE=true cdxgen -t pnpm --no-install-deps --profile license-compliance --spec-version 1.6 -o ../../sbom-source.cdx.json ../../compiled/",
|
||||
"enrich-sbom": "node ../../scripts/licenses/enrich-sbom.mjs ../../sbom-source.cdx.json",
|
||||
"render-licenses-md": "node ../../scripts/licenses/render-licenses-md.mjs ../../sbom-source.cdx.json ../../packages/cli/THIRD_PARTY_LICENSES.md ../../compiled/node_modules",
|
||||
|
||||
@@ -114,6 +114,7 @@ jobs:
|
||||
workflows: .github/**
|
||||
workflow-scripts:
|
||||
.github/scripts/**
|
||||
.devcontainer/codespaces/agent-worker*.mjs
|
||||
OWNERS
|
||||
scripts/licenses/**
|
||||
scripts/mutation-health/**
|
||||
|
||||
@@ -71,7 +71,8 @@ function ensureCodespace() {
|
||||
// The preludes go inside tmux's '…' argument: do not use single quotes in them.
|
||||
// tmux commands are not login shells. Source the shared secrets so Claude Code
|
||||
// finds ${FLAKY_MCP_TOKEN} in its process env.
|
||||
const SECRETS = '. /usr/local/lib/codespaces-env.sh 2>/dev/null || true';
|
||||
const SECRETS =
|
||||
'. /usr/local/lib/codespaces-env.sh 2>/dev/null || true; unset AGENT_WORKER_TOKEN N8N_DEQUEUE_URL OPENROUTER_API_KEY SLACK_BOT_TOKEN';
|
||||
// Worktrees share the pnpm store but not the turbo cache; a shared TURBO_CACHE_DIR
|
||||
// (seeded from the main checkout) keeps new-worktree builds at cache-hit speed.
|
||||
const CACHE = 'export TURBO_CACHE_DIR=/workspaces/.turbo-cache; [ -d "$TURBO_CACHE_DIR" ] || cp -r /workspaces/n8n/.turbo/cache "$TURBO_CACHE_DIR" 2>/dev/null || mkdir -p "$TURBO_CACHE_DIR"';
|
||||
|
||||
Reference in New Issue
Block a user