openvibers/openvibe.codes Commits
A ticket follows its pull request (AL5) (#375)
Changes
@@@ -30,6 +30,16 @@ OV_OAUTH_CLIENT_SECRET=# optional: defaults to true in production# COOKIE_SECURE=true++# OpenVibe.Events (AL5): Codes subscribes to run.job.* and, when PR follows are on, github.* at boot, and reads++# the signed deliveries at POST /internal/events (CODES_EVENTS_SECRET verifies them; 32 characters or more,++# comma-separated while rotating). Codes publishes codes.task.* here.++#CODES_EVENTS_URL=https://openvibe.events++#CODES_EVENTS_SECRET=++# A ticket follows its pull request (AL5), off by default: with 1, a finished task that pushed a PR is bound to it++# and a red check, a merge conflict or a review comment wakes a follow-up on the PR branch (at most 3 automatic++# fixes per PR; a person's "stop" ends it; the worker never merges). Also requires the Events secret above.++#CODES_PR_FOLLOW_ENABLED=0++# OpenVibe.AI (Codes C3). CODES_AI_URL is what a harness on a person's machine calls; without a values file it is# https://ai.openvibe.services.#CODES_AI_URL=https://ai.openvibe.services@@@ -5,14 +5,15 @@* worker runs it in its sandbox as artifact `[email protected]`, JOB_PROTOCOL below). The job's input is one JSON* line on stdin:*-- * { repo, ref?, prompt, agent?, model?, permission?: 'edit'|'full', push?: { branch?, pull_request? } }++ * { repo, ref?, prompt, agent?, model?, permission?: 'edit'|'full', push?: { branch?, pull_request?, update? } }** It clones `repo` (https only) into a scratch directory, runs the harness there with hand-offs, and prints one compact* JSON line per event on stdout (what the page shows live). The result goes to fd 3 (the Link's result channel), or as* a last `{ t: 'result_doc' }` line when fd 3 is not open: what the agents said, the changed files and the patch, the* per-turn checkpoints (refs and the authored/imported turn diff, `checkpoints` in the document), and, when `push` is* asked and the machine holds GITHUB_TOKEN, the new branch it pushed (never the base branch) and its pull request.-- * Nothing here ever prints a token.++ * A follow-up (AL5) sets push.update with branch === ref: the checked-out PR branch is committed and pushed again++ * (never forced, no new PR) so the same pull request turns green. Nothing here ever prints a token.*/const fs = require('fs');const os = require('os');@@@ -60,14 +61,16 @@ const PROTECTED = /^(main|master|trunk|dev|develop|development|staging|stage|pro/*** The branch a task may push: a new one. → { branch } or { refused }. `exists` is GitHub's answer for that name (true-- * when it already exists, null when GitHub could not be asked: refused too, never assumed new).++ * when it already exists, null when GitHub could not be asked: refused too, never assumed new). A follow-up (AL5)++ * passes update with branch === ref: the PR branch already exists and is pushed again, never forced.*/--function pushTarget(requested, base, { exists = false, now = Date.now() } = {}) {++function pushTarget(requested, base, { exists = false, now = Date.now(), update = false } = {}) {const branch = requested || `openvibe-codes/${now.toString(36)}`;-- if (branch === base) return { refused: 'a task never pushes to the branch it started from' };++ if (branch === base && !update) return { refused: 'a task never pushes to the branch it started from' };if (PROTECTED.test(branch)) return { refused: `a task pushes only a new branch of its own, never ${branch}` };if (exists === null) return { refused: 'GitHub could not say whether the branch exists, so nothing was pushed' };-- if (exists) return { refused: `${branch} already exists: a task pushes only a new branch` };++ if (exists && !update) return { refused: `${branch} already exists: a task pushes only a new branch` };++ if (update && !exists) return { refused: `${branch} is gone: nothing to update` };return { branch };}@@@ -235,9 +238,15 @@ function parseArgs(raw, env = process.env) {const push = a.push && typeof a.push === 'object' ? {branch: a.push.branch == null ? null : String(a.push.branch),pull_request: a.push.pull_request === true,++ // Only a follow-up (AL5) sets update; a normal task's args keep their old shape.++ ...(a.push.update === true ? { update: true } : {}),} : null;if (push && push.branch && (!BRANCH.test(push.branch) || push.branch.startsWith('-'))) throw new Error('push.branch must be a branch name');if (push && push.branch && PROTECTED.test(push.branch)) throw new Error(`push.branch must be a new branch of the task's own, never ${push.branch}`);++ // A follow-up (AL5) pushes the checked-out PR branch again: it names the branch it cloned (branch === ref),++ // never asks for a new pull request, and is never forced. Anything else with update is rejected.++ if (push && push.update && push.pull_request) throw new Error('push.update never opens a new pull request');++ if (push && push.update && (!push.branch || push.branch !== ref)) throw new Error('push.update needs branch === ref: a follow-up pushes the branch it cloned');// The session a resume continues: the job's own input, or RESUME in the environment (a task carried on in the// same conversation). It is a harness's session id, never a path and never a credential.const resumeIn = a.resume == null || a.resume === '' ? env.RESUME : a.resume;@@@ -606,19 +615,22 @@ async function pushBranch(dir, job, { env, fetchImpl, base }) {if (host && giteaHosts(env).has(host)) return pushForge(dir, job, { env, fetchImpl, base });}if (!token || !gh) return { pushed: false, reason: !gh ? 'push is for github.com repositories' : 'this machine has no GITHUB_TOKEN' };-- const first = pushTarget(job.push.branch, base);++ const update = job.push.update === true;++ const first = pushTarget(job.push.branch, base, { update });if (first.refused) return { pushed: false, reason: first.refused };const branch = first.branch;const remoteUrl = `https://x-access-token:${token}@github.com/${gh.owner}/${gh.name}.git`;let exists = null;try { exists = (await git(['ls-remote', '--heads', remoteUrl, `refs/heads/${branch}`], { cwd: dir, timeout: 60_000 })).trim() !== ''; } catch { exists = null; }-- const checked = pushTarget(branch, base, { exists });++ const checked = pushTarget(branch, base, { exists, update });if (checked.refused) return { pushed: false, reason: checked.refused };const who = {GIT_AUTHOR_NAME: env.GIT_AUTHOR_NAME || 'OpenVibe.Codes', GIT_AUTHOR_EMAIL: env.GIT_AUTHOR_EMAIL || '[email protected]',GIT_COMMITTER_NAME: env.GIT_AUTHOR_NAME || 'OpenVibe.Codes', GIT_COMMITTER_EMAIL: env.GIT_AUTHOR_EMAIL || '[email protected]',};-- await git(['checkout', '-b', branch], { cwd: dir });++ // A follow-up is already on the PR branch it cloned (branch === ref === base): committing there and pushing++ // (never forced, so a branch someone rewrote meanwhile is rejected, not overwritten).++ if (branch !== base) await git(['checkout', '-b', branch], { cwd: dir });await git(['commit', '-m', job.prompt.split('\n')[0].slice(0, 72) || 'OpenVibe.Codes task'], { cwd: dir, env: who });await git(['push', remoteUrl, `HEAD:refs/heads/${branch}`], { cwd: dir, timeout: 180_000 });const out = { pushed: true, branch, branch_url: `https://github.com/${gh.owner}/${gh.name}/tree/${encodeURIComponent(branch)}` };nnew file mode 100644@@@ -0,0 +1,23 @@++-- phase: expand++-- AL5: a ticket follows its pull request. One row per ticket (task) bound to a PR: which repo and PR number, the++-- role it plays there, how the PR was found, and the automatic fixes so far. A red check, a conflict or a review++-- comment on the PR wakes the same worker (a follow-up Run job on the PR branch); at most 3 automatic fixes per++-- PR; a person's "stop" or a merged/closed PR ends the binding. Nothing here holds a secret.++CREATE TABLE pr_bindings (++ id text COLLATE "C" PRIMARY KEY, -- prb_<ULID>++ task_id text COLLATE "C" NOT NULL, -- tsk_… the ticket's current task (moves to each follow-up)++ repo text COLLATE "C" NOT NULL, -- owner/name, lowercased++ pr_number integer NOT NULL CHECK (pr_number > 0),++ role text COLLATE "C" NOT NULL, -- review|verify|fix++ how_found text COLLATE "C" NOT NULL, -- created|worktree|review++ branch text COLLATE "C", -- the PR branch a follow-up pushes to (update, never a new PR)++ state text COLLATE "C" NOT NULL DEFAULT 'active', -- active|ended++ fix_count integer NOT NULL DEFAULT 0, -- automatic fixes this binding's task lineage used++ ended_reason text COLLATE "C", -- merged|closed|stopped_by_person++ last_wake_at bigint,++ last_wake_reason text COLLATE "C",++ created_at bigint NOT NULL,++ updated_at bigint NOT NULL++);++CREATE INDEX IF NOT EXISTS pr_bindings_pr ON pr_bindings (repo, pr_number);++CREATE INDEX IF NOT EXISTS pr_bindings_task ON pr_bindings (task_id);@@@ -157,6 +157,10 @@ async function createApp(opts = {}) {fetchImpl: fetchImpl || globalThis.fetch, fleet, userSettings, harnesses, attachments,});tasks.setConductor(ctx.conductor);++ // A ticket follows its pull request (AL5): bindings need tasks (a follow-up is a task), and tasks need++ // bindings (a terminal result with a PR binds it) — wired to each other here, not in a constructor.++ ctx.prBindings = require('./domain/pr-bindings').createPrBindings({ store, tasks, outbox, config, log });++ tasks.attachBindings(ctx.prBindings);const app = express();aiCatalogConfig.mount(app);@@@ -233,7 +237,7 @@ async function createApp(opts = {}) {const accountSend = opts.accountSend || (config.oauth.clientSecret? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl || globalThis.fetch }): null);-- const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, log });++ const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, bindings: ctx.prBindings, log });ctx.accountData = accountData;ctx.eventsConsumer = consumer;ctx.publisher = createPublisher({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });@@@ -127,6 +127,12 @@ function load(env = process.env) {deadAfterRefusals: Math.max(1, int(env.CODES_EVENTS_DEAD_AFTER_REFUSALS, 3)),deadAfterAttempts: Math.max(1, int(env.CODES_EVENTS_DEAD_AFTER_ATTEMPTS, 20)),},++ // A ticket follows its pull request (AL5): a red check, a conflict or a review comment wakes the bound++ // ticket's worker with a follow-up Run job, at most 3 automatic fixes per PR. Default off: set++ // CODES_PR_FOLLOW_ENABLED=1 to let Codes start follow-ups on github.* deliveries.++ prFollow: {++ enabled: env.CODES_PR_FOLLOW_ENABLED === '1',++ },// Session lines in PostgreSQL: no total cap; finished tasks keep theirs for this many days (default 90).sessions: {retentionDays: Math.max(1, int(env.CODES_SESSION_RETENTION_DAYS, 90)),nnew file mode 100644@@@ -0,0 +1,279 @@++'use strict';++++/**++ * A ticket follows its pull request (AL5).++ *++ * A binding is one row per ticket (task): the repo, the PR number, the role this ticket plays there++ * (review|verify|fix) and how the PR was found (created|worktree|review). A red check, a merge conflict or++ * a review comment on the PR wakes the SAME worker: a follow-up Run job on the PR branch, with the failing++ * evidence in its prompt. At most 3 automatic fixes per PR; a person's "stop" ends the binding; a merged or++ * closed PR ends it too. A bot's comment or a CI result never wakes an ended binding; a person's comment can++ * (it re-opens the binding and wakes it once). The worker never merges its own PR: a follow-up pushes the++ * same branch again (push.update, never a force push) and opens no new pull request; the harness has no merge++ * step at all.++ *++ * No polling: GitHub webhooks arrive through OpenVibe.Events (server/github.js there) as github.* envelopes,++ * delivered to Codes' /internal/events consumer, which calls handleGithubEvent here. The binding follows the++ * newest task: after a follow-up starts, the binding's task_id moves to it, so the next wake carries the same++ * ticket forward. Every bind, follow-up and end also writes a codes.pr.* outbox row (bound|followup|ended),++ * so the portal can show the ticket without reading this table.++ *++ * The switch is default-off: with CODES_PR_FOLLOW_ENABLED unset, red checks are recorded (ignored:++ * follow_disabled) and no follow-up starts.++ *++ * const bindings = createPrBindings({ store, tasks, outbox, config, log });++ * await bindings.maybeBind(taskRow); // after a task ends with a PR: idempotent, in-tx safe++ * await bindings.handleGithubEvent(envelope); // → { outcome, task_ids, followed_up }++ */++const { parseArgs } = require('../../harness/job');++++const ROLES = new Set(['review', 'verify', 'fix']);++const MAX_FIXES_PER_PR = 3;++/** Check conclusions that wake a verify|fix binding (a cancelled run is usually superseded, not broken). */++const RED_CONCLUSIONS = new Set(['failure', 'timed_out', 'action_required']);++/** A person's comment that is only "stop" (or starts "stop:" / "stop —") ends the binding instead of waking it.++ * A sentence that happens to begin with the word ("stop the flaky retry loop") is a normal comment. */++const STOP_RE = /^\s*stop\s*(?:[:—–-]\s*.*)?[.!]?\s*$/i;++const MAX_PROMPT = 16000;++const MAX_BODY = 4000;++const MAX_EVIDENCE = 6000;++++const slug = (v) => (typeof v === 'string' ? v.toLowerCase() : null);++const num = (v) => (Number.isInteger(v) && v > 0 ? v : null);++const str = (v, n) => (typeof v === 'string' ? v.slice(0, n) : null);++++/** owner/name and the PR number of a github.com pull URL, or null. */++function parsePrUrl(url) {++ if (typeof url !== 'string') return null;++ const m = /^https:\/\/github\.com\/([A-Za-z0-9-]+)\/([A-Za-z0-9._-]+)\/pull\/(\d+)(?:\/|$)/.exec(url.trim());++ if (!m) return null;++ return { repo: `${m[1]}/${m[2]}`.toLowerCase(), number: Number(m[3]) };++}++++function createPrBindings({ store, tasks, outbox = null, config = {}, log = console } = {}) {++ if (!store) throw new TypeError('pr-bindings need a store');++ const db = store.db;++ const enabled = Boolean(config.prFollow && config.prFollow.enabled);++++ const emit = async (topic, subject, payload) => {++ if (!outbox || typeof outbox.emitPr !== 'function' || !subject) return null;++ try { return await outbox.emitPr(topic, subject, payload); } catch (e) { log.warn(`[Codes] pr-bindings outbox: ${e.message}`); return null; }++ };++++ /**++ * Bind a finished task to the PR its result pushed (role fix, found when created). Idempotent: a task++ * binds its PR once. Uses the ambient transaction when called from run-tasks' recordState.++ */++ async function maybeBind(row) {++ if (!row || !row.id || !row.result) return null;++ let result = row.result;++ if (typeof result === 'string') { try { result = JSON.parse(result); } catch { return null; } }++ const push = result && typeof result.push === 'object' ? result.push : null;++ if (!push || push.pushed !== true || typeof push.pull_request !== 'string') return null;++ const pr = parsePrUrl(push.pull_request);++ if (!pr) return null;++ const branch = str(push.branch, 200);++ const seen = await db.maybe(++ 'SELECT id FROM pr_bindings WHERE task_id = $1 AND repo = $2 AND pr_number = $3 AND state = $4',++ [row.id, pr.repo, pr.number, 'active'],++ );++ if (seen) return seen;++ const now = store.now();++ const bound = await db.one(++ `INSERT INTO pr_bindings (id, task_id, repo, pr_number, role, how_found, branch, state, fix_count, created_at, updated_at)++ VALUES ($1, $2, $3, $4, 'fix', 'created', $5, 'active', 0, $6, $6) RETURNING *`,++ [store.newId('prb'), row.id, pr.repo, pr.number, branch, now],++ );++ await emit('codes.pr.bound',++ row.subject, { task_id: row.id, repo: pr.repo, pr_number: pr.number, role: 'fix', branch });++ return bound;++ }++++ const forPr = (repo, number, { activeOnly = false } = {}) => db.many(++ `SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2${activeOnly ? ` AND state = 'active'` : ''} ORDER BY created_at ASC`,++ [repo, number],++ );++++ async function endAll(rows, reason) {++ for (const b of rows) {++ if (b.state === 'ended') continue;++ await db.exec(`UPDATE pr_bindings SET state = 'ended', ended_reason = $2, updated_at = $3 WHERE id = $1`,++ [b.id, reason, store.now()]);++ const owner = await db.maybe('SELECT subject FROM run_tasks WHERE id = $1', [b.task_id]);++ await emit('codes.pr.ended', owner && owner.subject,++ { task_id: b.task_id, repo: b.repo, pr_number: b.pr_number, reason });++ }++ }++++ const fixesForPr = async (repo, number) => {++ const r = await db.maybe('SELECT COALESCE(SUM(fix_count), 0) AS n FROM pr_bindings WHERE repo = $1 AND pr_number = $2', [repo, number]);++ return Number((r && r.n) || 0);++ };++++ /** The follow-up prompt: the failing evidence first, the original task after, the no-merge rule in both. */++ function followupPrompt({ binding, parent, evidence, fixN, automatic = true }) {++ const lead = automatic++ ? `Automatic follow-up ${fixN} of ${MAX_FIXES_PER_PR}`++ : 'Follow-up requested by a person';++ const head = `${lead} for ${binding.repo} pull request #${binding.pr_number} (ticket task ${parent.id}).\n\nEvidence from the PR:\n${evidence}\n\nThe branch is ${binding.branch || '(the PR branch)'}: it is already checked out, fix it there, and push the same branch again — no new branch, no new pull request, and never merge anything yourself.`;++ const tail = `\n\nThe original task was:\n${typeof parent.prompt === 'string' ? parent.prompt : ''}`;++ return (head + tail).slice(0, MAX_PROMPT);++ }++++ /**++ * Start one follow-up Run job for the binding's ticket: same repo, the PR branch checked out and pushed++ * again (update, non-force), no new PR. The binding moves to the new task. Throws when Run refuses.++ */++ async function startFollowup(binding, evidence, { automatic = true } = {}) {++ const parent = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [binding.task_id]);++ if (!parent) throw new Error(`ticket task ${binding.task_id} is gone`);++ if (!binding.branch) throw new Error('the binding names no branch to push again');++ const fixes = await fixesForPr(binding.repo, binding.pr_number);++ const prompt = followupPrompt({ binding, parent, evidence, fixN: fixes + 1, automatic });++ // The worker's own harness session, when the task's last attempt recorded one: the follow-up continues the++ // same conversation instead of re-reading the repository from nothing. A machine without it falls back to++ // the prompt (harness/job.js treats an unknown resume as the agent's own error and hands off normally).++ const last = await db.maybe('SELECT agent, session FROM codes_attempts WHERE task_id = $1 ORDER BY n DESC LIMIT 1', [parent.id]);++ const agent = (last && last.agent) || parent.agent || null;++ const resume = last && typeof last.session === 'string' && last.session ? last.session : null;++ const args = parseArgs(JSON.stringify({++ repo: parent.repo, ref: binding.branch, prompt, agent,++ permission: 'edit',++ push: { branch: binding.branch, pull_request: false, update: true },++ ...(resume ? { resume } : {}),++ }));++ let actorRef = null;++ if (parent.actor_ref) { try { const p = typeof parent.actor_ref === 'string' ? JSON.parse(parent.actor_ref) : parent.actor_ref; if (p && typeof p === 'object') actorRef = p; } catch { actorRef = null; } }++ const task = await tasks.start({ subject: parent.subject, project: parent.project_id, args, via: parent.via, key: null, actorRef });++ await db.exec(`UPDATE pr_bindings SET task_id = $2, fix_count = fix_count + $5, last_wake_at = $3, last_wake_reason = $4, updated_at = $3 WHERE id = $1`,++ [binding.id, task.id, store.now(), evidence.slice(0, 120), automatic ? 1 : 0]);++ await emit('codes.pr.followup', parent.subject,++ { task_id: task.id, parent_task_id: parent.id, repo: binding.repo, pr_number: binding.pr_number, fix_n: fixes + 1, automatic });++ return task;++ }++++ /** One wake across the PR's active bindings of the given roles. → { started, skipped } of task ids. */++ async function wake(rows, { roles, evidence, reason, eventId, automatic = true }) {++ const started = [];++ const skipped = [];++ if (!enabled) return { started, skipped, disabled: true };++ // The cap bounds automatic fixes only: a person's comment is a new instruction and may start a manual one.++ if (automatic && await fixesForPr(rows[0].repo, rows[0].pr_number) >= MAX_FIXES_PER_PR) return { started, skipped, capped: true };++ for (const b of rows) {++ if (!roles.has(b.role)) { skipped.push(b.task_id); continue; }++ const current = await db.maybe('SELECT id, state FROM run_tasks WHERE id = $1', [b.task_id]);++ if (current && !['succeeded', 'failed', 'cancelled', 'expired'].includes(current.state)) { skipped.push(b.task_id); continue; }++ try {++ const task = await startFollowup(b, evidence, { automatic });++ started.push(task.id);++ log.info(`[Codes] pr-follow ${b.repo}#${b.pr_number} woke ${task.id} (${reason}, ${eventId})`);++ } catch (e) {++ log.warn(`[Codes] pr-follow ${b.repo}#${b.pr_number} failed to wake: ${e.message}`);++ skipped.push(b.task_id);++ }++ }++ return { started, skipped };++ }++++ /**++ * A github.* envelope (already inbox-deduped by the consumer). → { outcome, task_ids, followed_up }.++ * outcome is what the consumer records: woke, stopped, reopened, ended:<reason>, ignored:<why>.++ */++ async function handleGithubEvent(event) {++ if (!event || typeof event.event_type !== 'string') return { outcome: 'ignored:payload', task_ids: [], followed_up: [] };++ const subject = event.subject && typeof event.subject === 'object' ? event.subject : null;++ const m = subject && typeof subject.id === 'string' ? /^([a-z0-9-]+(?:\/[a-z0-9._-]+))#(\d+)$/.exec(subject.id) : null;++ if (!m) return { outcome: 'ignored:payload', task_ids: [], followed_up: [] };++ const repo = m[1];++ const number = Number(m[2]);++ const payload = event.payload && typeof event.payload === 'object' ? event.payload : {};++ // The subject is the routing key (a check fans out one envelope per PR, so its payload names no number);++ // a payload that names another repo or PR is not ours.++ if (payload.repo !== undefined && slug(payload.repo) !== repo) return { outcome: 'ignored:payload', task_ids: [], followed_up: [] };++ if (payload.pr_number !== undefined && num(payload.pr_number) !== number) return { outcome: 'ignored:payload', task_ids: [], followed_up: [] };++ const rows = await forPr(repo, number);++ if (!rows.length) return { outcome: 'ignored:no_binding', task_ids: [], followed_up: [] };++ const active = rows.filter((b) => b.state === 'active');++ const isPerson = payload.author_type === 'User';++++ if (event.event_type === 'github.check.completed') {++ if (!RED_CONCLUSIONS.has(payload.conclusion)) return { outcome: 'ignored:green', task_ids: rows.map((b) => b.task_id), followed_up: [] };++ if (!active.length) return { outcome: 'ignored:ended', task_ids: rows.map((b) => b.task_id), followed_up: [] };++ const out = payload.output && typeof payload.output === 'object' ? payload.output : null;++ const parts = [`check "${payload.check_name || 'ci'}" concluded "${payload.conclusion}"`++ + `${payload.head_sha ? ` at ${String(payload.head_sha).slice(0, 12)}` : ''}`++ + `${payload.html_url ? ` (${payload.html_url})` : ''}`];++ if (out && out.title) parts.push(`title: ${String(out.title).slice(0, 500)}`);++ if (out && out.summary) parts.push(`summary: ${String(out.summary).slice(0, 2000)}`);++ if (out && out.text) parts.push(`log tail:\n${String(out.text).slice(0, 4000)}`);++ if (payload.details_url) parts.push(`details: ${String(payload.details_url).slice(0, 500)}`);++ parts.push('— find why it failed on this branch and fix it.');++ const evidence = parts.join('\n').slice(0, MAX_EVIDENCE);++ const r = await wake(active, { roles: new Set(['verify', 'fix']), evidence, reason: 'red-check', eventId: event.event_id });++ if (r.disabled) return { outcome: 'ignored:follow_disabled', task_ids: active.map((b) => b.task_id), followed_up: [] };++ if (r.capped) return { outcome: 'ignored:max_fixes', task_ids: active.map((b) => b.task_id), followed_up: [] };++ return { outcome: r.started.length ? 'woke' : 'ignored:fix_running', task_ids: active.map((b) => b.task_id), followed_up: r.started };++ }++++ if (event.event_type === 'github.pr.updated') {++ if (payload.action === 'closed') {++ const reason = payload.merged === true ? 'merged' : 'closed';++ await endAll(rows, reason);++ return { outcome: `ended:${reason}`, task_ids: rows.map((b) => b.task_id), followed_up: [] };++ }++ if (payload.mergeable_state === 'dirty' || payload.mergeable === false) {++ if (!active.length) return { outcome: 'ignored:ended', task_ids: rows.map((b) => b.task_id), followed_up: [] };++ const where = [payload.head_ref ? `branch ${payload.head_ref}` : null, payload.base ? `onto ${payload.base}` : null].filter(Boolean).join(' ');++ const evidence = [`the pull request reports mergeable_state "${payload.mergeable_state || 'dirty'}"`++ + `${where ? ` (${where})` : ''}`++ + `${payload.changed_files ? `, ${payload.changed_files} file${payload.changed_files === 1 ? '' : 's'} changed` : ''}`++ + `${payload.html_url ? ` (${payload.html_url})` : ''}`,++ payload.files_url ? `the changed files: ${payload.files_url}` : null,++ '— fetch the base branch, run `git merge-tree` (or a local merge) to list the conflicted files, resolve them on this branch, and push it again.']++ .filter(Boolean).join('\n').slice(0, MAX_EVIDENCE);++ const r = await wake(active, { roles: new Set(['fix']), evidence, reason: 'conflict', eventId: event.event_id });++ if (r.disabled) return { outcome: 'ignored:follow_disabled', task_ids: active.map((b) => b.task_id), followed_up: [] };++ if (r.capped) return { outcome: 'ignored:max_fixes', task_ids: active.map((b) => b.task_id), followed_up: [] };++ return { outcome: r.started.length ? 'woke' : 'ignored:fix_running', task_ids: active.map((b) => b.task_id), followed_up: r.started };++ }++ return { outcome: 'ignored:no_conflict', task_ids: rows.map((b) => b.task_id), followed_up: [] };++ }++++ if (event.event_type === 'github.pr.commented') {++ const body = str(payload.body, MAX_BODY) || '';++ if (STOP_RE.test(body) && isPerson) {++ if (!active.length) return { outcome: 'ignored:ended', task_ids: rows.map((b) => b.task_id), followed_up: [] };++ await endAll(active, 'stopped_by_person');++ return { outcome: 'stopped', task_ids: rows.map((b) => b.task_id), followed_up: [] };++ }++ if (!isPerson) return { outcome: active.length ? 'ignored:bot' : 'ignored:ended', task_ids: rows.map((b) => b.task_id), followed_up: [] };++ // A person's comment: on an ended binding it re-opens the ticket and wakes it once; otherwise it is++ // review feedback for the active bindings. Never past the per-PR fix cap.++ let targets = active;++ let reopened = false;++ if (!targets.length) {++ const last = rows[rows.length - 1];++ await db.exec(`UPDATE pr_bindings SET state = 'active', ended_reason = NULL, updated_at = $2 WHERE id = $1`, [last.id, store.now()]);++ targets = [{ ...last, state: 'active' }];++ reopened = true;++ }++ const where = payload.path ? ` on ${payload.path}${payload.line ? `:${payload.line}` : ''}` : '';++ const evidence = `${payload.author || 'someone'} commented${where} on the pull request${payload.html_url ? ` (${payload.html_url})` : ''}:\n${body}\n— address it on the branch and push it again.`;++ // A person's comment is manual: it can start a fix past the automatic cap.++ const r = await wake(targets, { roles: new Set(['review', 'fix']), evidence, reason: 'comment', eventId: event.event_id, automatic: false });++ if (r.disabled) return { outcome: 'ignored:follow_disabled', task_ids: targets.map((b) => b.task_id), followed_up: [] };++ if (r.capped) return { outcome: 'ignored:max_fixes', task_ids: targets.map((b) => b.task_id), followed_up: [] };++ if (!r.started.length) return { outcome: 'ignored:fix_running', task_ids: targets.map((b) => b.task_id), followed_up: [] };++ return { outcome: reopened ? 'reopened' : 'woke', task_ids: targets.map((b) => b.task_id), followed_up: r.started };++ }++++ return { outcome: 'ignored:type', task_ids: rows.map((b) => b.task_id), followed_up: [] };++ }++++ /** This ticket's bindings, newest first (for pages and the portal). */++ const listForTask = (taskId) => db.many('SELECT * FROM pr_bindings WHERE task_id = $1 ORDER BY created_at DESC', [taskId]);++++ return { maybeBind, handleGithubEvent, listForTask, startFollowup, forPr, parsePrUrl, MAX_FIXES_PER_PR, RED_CONCLUSIONS };++}++++module.exports = { createPrBindings, parsePrUrl, MAX_FIXES_PER_PR, RED_CONCLUSIONS, STOP_RE };@@@ -164,6 +164,10 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou*/let conductorTurns = conductor;const setConductor = (turns) => { conductorTurns = turns; };++ // A ticket follows its pull request (AL5): wired by the app after both exist, because a follow-up is a task++ // started here. A terminal task whose result pushed a PR binds it, in the same transaction.++ let prBindings = null;++ const attachBindings = (b) => { prBindings = b || null; };/** An attempt's state when it ended: a limit reached is a budget-stop, never a plain failure. */const ENDED = new Set(['done', 'failed', 'stopped', 'budget-stopped', 'interrupted']);@@@ -501,6 +505,9 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou});}}++ if (TERMINAL.has(updated.state) && prBindings) {++ try { await prBindings.maybeBind(updated); } catch (e) { log.warn(`[Codes] pr-bind ${updated.id}: ${String((e && e.message) || e).slice(0, 200)}`); }++ }return updated;});// What the refused run really spent: OpenVibe.AI counted it (the harness may have stopped mid-turn).@@@ -1074,7 +1081,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, oureturn {start, refresh, applyJob, applyEvent, recordState, afterEnd, continueTask, stopTask, proof: taskProof,-- latestProject, present, who, attempts: attemptStore, setConductor, fundTurn,++ latestProject, present, who, attempts: attemptStore, setConductor, attachBindings, fundTurn,setForge(f) { forge = f || null; },};}@@@ -15,7 +15,11 @@* backoff; it is off without EVENTS_URL/CODES_EVENTS_SECRET/the OAuth client secret, under CODES_DRILL, or when* CODES_EVENTS_SUBSCRIBE=0.*-- * const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, log });++ * github.check.completed | github.pr.updated | github.pr.commented++ * GitHub webhooks through OpenVibe.Events (AL5): routed to the pr-bindings (server/domain/++ * pr-bindings.js), which wake the bound ticket's worker. The inbox dedupes redeliveries.++ *++ * const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, bindings, log });* app.use('/internal/events', consumer.router);*/const express = require('express');@@@ -27,7 +31,9 @@ const { TERMINAL, STATES } = require('./domain/run-tasks');const RUN_TOPICS = Object.freeze(['run.job.queued', 'run.job.started', 'run.job.succeeded', 'run.job.failed', 'run.job.cancelled', 'run.job.expired',]);--const TOPICS = Object.freeze([...RUN_TOPICS, ...accountDataLib.TOPICS]);++/** GitHub through OpenVibe.Events (AL5): one envelope per pull request (server/github.js there). */++const GITHUB_TOPICS = Object.freeze(['github.check.completed', 'github.pr.updated', 'github.pr.commented']);++const TOPICS = Object.freeze([...RUN_TOPICS, ...accountDataLib.TOPICS, ...GITHUB_TOPICS]);const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;const JOB_RE = /^job_[0-9A-HJKMNP-TV-Z]{26}$/;const STATE_OF_TOPIC = {@@@ -42,7 +48,7 @@ function errorText(error) {return null;}--function createEventsConsumer({ config, store, tasks, run, accountData, accountSend = null, log = console, now = store.now }) {++function createEventsConsumer({ config, store, tasks, run, accountData, accountSend = null, bindings = null, log = console, now = store.now }) {const db = store.db;const secrets = (config.events && config.events.secrets ? config.events.secrets : []).filter((x) => typeof x === 'string' && x.length >= 32);const stats = { received: 0, applied: 0, duplicates: 0, ignored: 0, refused: 0, failed: 0, last_at: null };@@@ -128,6 +134,39 @@ function createEventsConsumer({ config, store, tasks, run, accountData, accountSreturn res.json({ event_id: event.event_id, outcome });}++ if (GITHUB_TOPICS.includes(event.event_type)) {++ // A ticket follows its pull request (AL5). The inbox claims the id first: a redelivery answers++ // duplicate, and the wake below runs once.++ const seen = await db.maybe('SELECT task_id, outcome FROM codes_events_inbox WHERE event_id = $1', [event.event_id]);++ if (seen) { stats.duplicates++; return res.json({ event_id: event.event_id, duplicate: true, outcome: null }); }++ if (!bindings) {++ const result = await ignore(event, 'ignored:no_bindings');++ if (result.duplicate) stats.duplicates++; else stats.ignored++;++ return res.json({ event_id: event.event_id, duplicate: result.duplicate, outcome: result.outcome });++ }++ let result;++ try {++ result = await store.tx(async () => {++ const inserted = await db.maybe(++ `INSERT INTO codes_events_inbox (event_id, event_type, task_id, outcome, processed_at)++ VALUES ($1, $2, NULL, 'wake', $3::timestamptz) ON CONFLICT (event_id) DO NOTHING RETURNING event_id`,++ [event.event_id, event.event_type, store.iso()],++ );++ if (!inserted) return { duplicate: true, outcome: null, task_ids: [], followed_up: [] };++ const r = await bindings.handleGithubEvent(event);++ await db.exec('UPDATE codes_events_inbox SET outcome = $2 WHERE event_id = $1', [event.event_id, r.outcome]);++ return { duplicate: false, ...r };++ });++ } catch (err) {++ stats.failed++;++ log.error(`[Codes] github event ${event.event_id} (${event.event_type}) failed:`, err.message);++ return problem(503, 'codes.event_failed', 'processing failed; it will be retried');++ }++ if (result.duplicate) { stats.duplicates++; return res.json({ event_id: event.event_id, duplicate: true, outcome: null }); }++ if (String(result.outcome).startsWith('ignored:')) stats.ignored++; else stats.applied++;++ return res.json({ event_id: event.event_id, duplicate: false, outcome: result.outcome, ...(result.followed_up && result.followed_up.length ? { task_ids: result.followed_up } : {}) });++ }++if (!RUN_TOPICS.includes(event.event_type)) {const result = await ignore(event, 'ignored:type');if (result.duplicate) stats.duplicates++; else stats.ignored++;@@@ -145,7 +184,7 @@ function createEventsConsumer({ config, store, tasks, run, accountData, accountSreturn res.json({ event_id: event.event_id, duplicate: false, outcome: result.outcome, ...(result.task_id ? { task_id: result.task_id } : {}) });});-- return { router, topics: TOPICS, runTopics: RUN_TOPICS, stats: () => ({ ...stats, enabled: secrets.length > 0 }) };++ return { router, topics: TOPICS, runTopics: RUN_TOPICS, githubTopics: GITHUB_TOPICS, stats: () => ({ ...stats, enabled: secrets.length > 0 }) };}/** Create any missing subscription for `topics` at Events (idempotent; retried in the background at boot). */@@@ -206,4 +245,4 @@ function startSubscriptions(opts = {}) {return { topics, endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };}--module.exports = { createEventsConsumer, startSubscriptions, RUN_TOPICS, TOPICS, EVENT_ID_RE };++module.exports = { createEventsConsumer, startSubscriptions, RUN_TOPICS, GITHUB_TOPICS, TOPICS, EVENT_ID_RE };@@@ -21,6 +21,11 @@ const TOPICS = Object.freeze(['codes.task.queued', 'codes.task.running', 'codes.task.succeeded', 'codes.task.failed', 'codes.task.cancelled',]);const TOPIC_SET = new Set(TOPICS);++/** A ticket follows its pull request (AL5): bound, follow-up started, ended. Payloads are plain JSON; the++ * service manifest's eventsProduced stays the codes.task.* five (contracts.test.js), so these are the++ * portal's view, not a new contract. */++const PR_TOPICS = Object.freeze(['codes.pr.bound', 'codes.pr.followup', 'codes.pr.ended']);++const PR_TOPIC_SET = new Set(PR_TOPICS);const MAX_BACKOFF_MS = 5 * 60_000;// The envelope contract (events.event-envelope@1): the envelope Codes sends has to match it, or OpenVibe.Events// refuses the event with a 422 and the row is retried forever.@@@ -160,7 +165,23 @@ function createOutbox({ store, log = console }) {return eventId;}-- return { emitTask, emitAgentStep, payloadOf, topicFor, groupOf };++ /**++ * A pr-follow event in the same outbox (the publisher needs no topic filter: it sends every unpublished++ * row). subject is the ticket owner's usr_…; the payload names the ticket, the repo, the PR and the fix.++ */++ async function emitPr(topic, subject, payload) {++ if (!PR_TOPIC_SET.has(topic) || !subject || !payload || typeof payload !== 'object') return null;++ const eventId = ids.newId('event', store.now());++ const at = store.iso();++ await db.exec(++ `INSERT INTO codes_outbox (event_id, topic, subject, payload, created_at, next_attempt_at)++ VALUES ($1, $2, $3, $4::jsonb, $5::timestamptz, $5::timestamptz)`,++ [eventId, topic, subject, JSON.stringify(payload), at],++ );++ return eventId;++ }++++ return { emitTask, emitAgentStep, emitPr, payloadOf, topicFor, groupOf, PR_TOPICS };}function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = console, intervalMs = (config.events && config.events.intervalMs) || 5000 }) {@@@ -333,4 +354,4 @@ function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = co};}--module.exports = { createOutbox, createPublisher, TOPICS, TOPIC_SET, AGENT_TOPICS, agentStepOf, topicFor, groupOf, payloadOf, repoSlug, MAX_BACKOFF_MS, DEAD_AFTER_REFUSALS, DEAD_AFTER_ATTEMPTS };++module.exports = { createOutbox, createPublisher, TOPICS, TOPIC_SET, PR_TOPICS, AGENT_TOPICS, agentStepOf, topicFor, groupOf, payloadOf, repoSlug, MAX_BACKOFF_MS, DEAD_AFTER_REFUSALS, DEAD_AFTER_ATTEMPTS };nnew file mode 100644@@@ -0,0 +1,285 @@++'use strict';++/**++ * A ticket follows its pull request (AL5): a binding is stored when a task's result pushes a PR; a failed++ * check, a conflict or a review comment on that PR wakes the same worker with the failing evidence (at most++ * 3 automatic fixes per PR); a person's "stop" ends it; a bot's comment on an ended ticket does nothing.++ *++ * The github.* envelopes below are the shape OpenVibe.Events stores for the recorded GitHub payloads in its++ * own test/fixtures/github (same fake repo and PR: OpenVibers/OpenVibe.Codes#999), delivered to /internal/events++ * signed the way Events signs (openvibe-sdk/events, the CODES_EVENTS_SECRET fixture).++ */++const assert = require('assert');++const { signDeliveryHeaders } = require('openvibe-sdk/events');++const { ids } = require('openvibe-contracts');++const { parseArgs } = require('../harness/job');++const { parsePrUrl, MAX_FIXES_PER_PR, RED_CONCLUSIONS } = require('../server/domain/pr-bindings');++const { boot, check, done } = require('./helpers/boot');++++const SECRET = `fixture_${'pr-follow'.repeat(4)}`;++const NOW = Date.now();++const REPO = 'https://github.com/OpenVibers/OpenVibe.Codes';++const PR_URL = 'https://github.com/OpenVibers/OpenVibe.Codes/pull/999';++const BRANCH = 'al5-scratch';++const PROJECT = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';++++const terminalJob = () => ({++ state: 'succeeded',++ result: { ok: true, summary: 'did the thing', push: { pushed: true, branch: BRANCH, pull_request: PR_URL } },++ timings: { started_at: new Date(NOW).toISOString(), finished_at: new Date(NOW).toISOString() },++ error: null,++});++++const ghEvent = (eventType, subjectId, payload) => ({++ event_id: ids.newId('event'), event_type: eventType, version: 1, source: 'github',++ actor: { type: 'service', id: 'github' }, timestamp: new Date(NOW).toISOString(),++ subject: { type: 'pr', id: subjectId }, payload,++});++const SUBJECT = 'openvibers/openvibe.codes#999';++// The real ingress fans a check out one envelope per PR: the number lives in the subject, not the payload.++const redCheck = () => ghEvent('github.check.completed', SUBJECT, {++ repo: 'openvibers/openvibe.codes', kind: 'check_run', conclusion: 'failure',++ check_name: 'codes-ci / tests', head_sha: 'abc123',++ html_url: 'https://github.com/OpenVibers/OpenVibe.Codes/actions/runs/111/job/222',++ details_url: 'https://github.com/OpenVibers/OpenVibe.Codes/actions/runs/111',++ output: {++ title: '1 failing test', summary: 'test/pr-bindings.test.js failed on the branch',++ text: 'not ok 3 - a red check wakes the worker once\n AssertionError: expected 0 to equal 1\n# fail 1\n',++ },++});++const wrongPr = () => ghEvent('github.check.completed', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 1000, kind: 'check_run', conclusion: 'failure', check_name: 'x', head_sha: 'abc123', html_url: 'https://example.com' });++const greenCheck = () => ghEvent('github.check.completed', SUBJECT, { repo: 'openvibers/openvibe.codes', kind: 'check_run', conclusion: 'success', check_name: 'codes-ci / tests', head_sha: 'abc123', html_url: 'https://github.com/OpenVibers/OpenVibe.Codes/actions/runs/112' });++const conflict = () => ghEvent('github.pr.updated', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 999, action: 'synchronize', merged: false, mergeable_state: 'dirty', head_sha: 'abc123', base: 'main', html_url: PR_URL });++const personComment = (body) => ghEvent('github.pr.commented', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 999, author: 'owner-person', author_type: 'User', state: 'commented', body, html_url: `${PR_URL}#issuecomment-1` });++const botComment = () => ghEvent('github.pr.commented', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 999, author: 'github-actions[bot]', author_type: 'Bot', state: 'commented', body: 'All checks have passed.', html_url: `${PR_URL}#issuecomment-2` });++const merged = () => ghEvent('github.pr.updated', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 999, action: 'closed', merged: true, mergeable_state: 'clean', head_sha: 'abc123', base: 'main', html_url: PR_URL });++++(async () => {++ const submissions = [];++ let nextJob = 1;++ const run = {++ enabled: true,++ async submit({ subject, project, args }) {++ submissions.push({ subject, project, args });++ return { id: `job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J${nextJob++}`, state: 'queued' };++ },++ async read() { throw new Error('not used'); },++ async cancel() { return { state: 'cancelled' }; },++ async streamUrl() { throw new Error('not used'); },++ };++ const t = await boot({ run, env: { CODES_EVENTS_SECRET: SECRET, CODES_PR_FOLLOW_ENABLED: '1' } });++ const db = t.ctx.store.db;++ const ada = t.network.addUser('ada');++ const args = parseArgs(JSON.stringify({ repo: REPO, prompt: 'Fix the typo on the page', permission: 'edit', push: { pull_request: true } }));++++ const deliver = async (event) => {++ const body = JSON.stringify({ event, seq: 1 });++ return await t.get('/internal/events', { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });++ };++ const bindingOf = (taskId) => db.maybe('SELECT * FROM pr_bindings WHERE task_id = $1', [taskId]);++ const outbox = () => db.many('SELECT topic, subject, payload FROM codes_outbox ORDER BY created_at ASC');++++ try {++ await check('a finished task that pushed a PR binds it (role fix, found on create)', async () => {++ const task = await t.ctx.tasks.start({ subject: ada.subject, project: PROJECT, args });++ const ended = await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [task.id]), { state: 'succeeded', job: terminalJob(), timings: terminalJob().timings });++ assert.strictEqual(ended.state, 'succeeded');++ const b = await bindingOf(task.id);++ assert.ok(b, 'no binding stored');++ assert.strictEqual(b.repo, 'openvibers/openvibe.codes');++ assert.strictEqual(b.pr_number, 999);++ assert.strictEqual(b.role, 'fix');++ assert.strictEqual(b.how_found, 'created');++ assert.strictEqual(b.branch, BRANCH);++ assert.strictEqual(b.state, 'active');++ assert.strictEqual(b.fix_count, 0);++ const topics = (await outbox()).map((r) => r.topic);++ assert.ok(topics.includes('codes.pr.bound'), `no codes.pr.bound in ${topics.join(',')}`);++ });++++ let parent;++ await check('a red check wakes the worker once, with the log tail and the same session, and no merge', async () => {++ parent = await db.maybe('SELECT * FROM run_tasks ORDER BY created_at DESC LIMIT 1');++ // The worker's last attempt recorded its harness session; the follow-up continues that conversation.++ await db.exec("UPDATE codes_attempts SET session = 'sess_pr_follow_1', agent = 'claude-code' WHERE task_id = $1", [parent.id]);++ const before = submissions.length;++ const r = await deliver(redCheck());++ assert.strictEqual(r.status, 200, r.text);++ assert.strictEqual(r.json().outcome, 'woke', r.text);++ assert.strictEqual(submissions.length, before + 1, 'the red check must start exactly one follow-up');++ const job = submissions[submissions.length - 1];++ assert.strictEqual(job.args.repo, REPO);++ assert.strictEqual(job.args.ref, BRANCH);++ assert.strictEqual(job.args.push.branch, BRANCH);++ assert.strictEqual(job.args.push.pull_request, false);++ assert.strictEqual(job.args.push.update, true);++ assert.strictEqual(job.args.agent, 'claude-code');++ assert.strictEqual(job.args.resume, 'sess_pr_follow_1');++ assert.match(job.args.prompt, /Automatic follow-up 1 of 3/);++ assert.match(job.args.prompt, /concluded "failure"/);++ assert.match(job.args.prompt, /not ok 3 - a red check wakes the worker once/);++ assert.match(job.args.prompt, /\/actions\/runs\/111/);++ assert.match(job.args.prompt, /never merge/);++ assert.ok(!/merge( |ing|d) /.test(job.args.prompt.replace('never merge', '')), 'the follow-up must not be told to merge');++ const b = await db.maybe('SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);++ assert.strictEqual(b.fix_count, 1);++ assert.notStrictEqual(b.task_id, parent.id, 'the binding follows the newest task');++ const child = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [b.task_id]);++ assert.ok(child && child.state === 'queued');++ const topics = (await outbox()).map((r) => r.topic);++ assert.ok(topics.includes('codes.pr.followup'), `no codes.pr.followup in ${topics.join(',')}`);++ });++++ await check('the fourth red check does not wake (at most 3 automatic fixes per PR)', async () => {++ // Fixes 2 and 3: end each follow-up first so the fix_running guard lets the next one through.++ for (const n of [2, 3]) {++ const b = await db.maybe('SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);++ await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [b.task_id]), { state: 'succeeded', job: terminalJob(), timings: terminalJob().timings });++ const before = submissions.length;++ const r = await deliver(redCheck());++ assert.strictEqual(r.json().outcome, 'woke', r.text);++ assert.strictEqual(submissions.length, before + 1);++ const last = submissions[submissions.length - 1];++ assert.match(last.args.prompt, new RegExp(`Automatic follow-up ${n} of 3`));++ }++ const b = await db.maybe('SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);++ await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [b.task_id]), { state: 'succeeded', job: terminalJob(), timings: terminalJob().timings });++ const before = submissions.length;++ const r = await deliver(redCheck());++ assert.strictEqual(r.json().outcome, 'ignored:max_fixes', r.text);++ assert.strictEqual(submissions.length, before, 'the fourth red must not start a follow-up');++++ // But a person's comment is a new instruction: it may start a manual fix past the automatic cap.++ const manual = await deliver(personComment('please fix the last failure by hand'));++ assert.strictEqual(manual.json().outcome, 'woke', manual.text);++ assert.strictEqual(submissions.length, before + 1);++ assert.match(submissions[submissions.length - 1].args.prompt, /Follow-up requested by a person/);++ const after = await db.maybe('SELECT fix_count, task_id FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);++ assert.strictEqual(after.fix_count, 3, 'a manual fix does not count against the automatic cap');++ });++++ await check('a green check changes nothing; a payload naming another PR is not ours', async () => {++ const before = submissions.length;++ const r = await deliver(greenCheck());++ assert.strictEqual(r.json().outcome, 'ignored:green', r.text);++ assert.strictEqual(submissions.length, before);++ const w = await deliver(wrongPr());++ assert.strictEqual(w.json().outcome, 'ignored:payload', w.text);++ assert.strictEqual(submissions.length, before);++ });++++ await check('a conflict wakes the fix binding with the evidence', async () => {++ // A fresh ticket on another PR, so the fix cap of #999 does not apply.++ const other = await t.ctx.tasks.start({ subject: ada.subject, project: PROJECT, args: parseArgs(JSON.stringify({ repo: REPO, prompt: 'Another typo', permission: 'edit', push: { pull_request: true } })) });++ await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [other.id]), {++ state: 'succeeded',++ job: { ...terminalJob(), result: { ...terminalJob().result, push: { pushed: true, branch: 'al5-other', pull_request: 'https://github.com/OpenVibers/OpenVibe.Codes/pull/1000' } } },++ timings: terminalJob().timings,++ });++ const dirty = ghEvent('github.pr.updated', 'openvibers/openvibe.codes#1000', { repo: 'openvibers/openvibe.codes', pr_number: 1000, action: 'synchronize', merged: false, mergeable_state: 'dirty', head_sha: 'def456', base: 'main', html_url: 'https://github.com/OpenVibers/OpenVibe.Codes/pull/1000' });++ const before = submissions.length;++ const r = await deliver(dirty);++ assert.strictEqual(r.json().outcome, 'woke', r.text);++ assert.strictEqual(submissions.length, before + 1);++ assert.match(submissions[submissions.length - 1].args.prompt, /mergeable_state "dirty"/);++ });++++ await check('a merged PR ends the binding', async () => {++ const r = await deliver(merged());++ assert.strictEqual(r.json().outcome, 'ended:merged', r.text);++ const b = await db.maybe('SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);++ assert.strictEqual(b.state, 'ended');++ assert.strictEqual(b.ended_reason, 'merged');++ const topics = (await outbox()).map((r) => r.topic);++ assert.ok(topics.includes('codes.pr.ended'), `no codes.pr.ended in ${topics.join(',')}`);++ });++++ await check('a bot comment on an ended ticket does nothing', async () => {++ const before = submissions.length;++ const r = await deliver(botComment());++ assert.strictEqual(r.json().outcome, 'ignored:ended', r.text);++ assert.strictEqual(submissions.length, before);++ });++++ await check('a person\'s "stop" ends the binding; a person\'s comment re-opens and wakes it', async () => {++ // Re-openable ticket on PR #1000 (one fix used, cap not reached).++ const stop1000 = personComment('stop — for now, I will finish this myself');++ stop1000.subject = { type: 'pr', id: 'openvibers/openvibe.codes#1000' };++ stop1000.payload = { ...stop1000.payload, pr_number: 1000 };++ const wake1000 = personComment('actually the test still fails, please fix it');++ wake1000.subject = { type: 'pr', id: 'openvibers/openvibe.codes#1000' };++ wake1000.payload = { ...wake1000.payload, pr_number: 1000 };++ // End #1000's follow-up with a plain result (no PR: it binds nothing new) so the next wake may start.++ {++ const b = await db.maybe("SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 AND state = 'active'", ['openvibers/openvibe.codes', 1000]);++ await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [b.task_id]), {++ state: 'succeeded',++ job: { state: 'succeeded', result: { ok: true, summary: 'fixed the conflict' }, timings: terminalJob().timings, error: null },++ timings: terminalJob().timings,++ });++ }++ let r = await deliver(stop1000);++ assert.strictEqual(r.json().outcome, 'stopped', r.text);++ let b = await db.maybe("SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 AND state = 'active'", ['openvibers/openvibe.codes', 1000]);++ assert.strictEqual(b, null, 'stop must end the active binding');++ const before = submissions.length;++ r = await deliver(wake1000);++ assert.strictEqual(r.json().outcome, 'reopened', r.text);++ assert.strictEqual(submissions.length, before + 1, 'a person can wake an ended ticket');++ b = await db.maybe("SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 AND state = 'active'", ['openvibers/openvibe.codes', 1000]);++ assert.ok(b, 'the comment re-opens the binding');++ assert.match(submissions[submissions.length - 1].args.prompt, /actually the test still fails/);++ });++++ await check('with the switch off, a red check is recorded and no follow-up starts', async () => {++ const t2 = await boot({ run, env: { CODES_EVENTS_SECRET: SECRET } });++ try {++ const db2 = t2.ctx.store.db;++ const ada2 = t2.network.addUser('bob');++ const task = await t2.ctx.tasks.start({ subject: ada2.subject, project: PROJECT, args });++ await t2.ctx.tasks.applyEvent(await db2.maybe('SELECT * FROM run_tasks WHERE id = $1', [task.id]), { state: 'succeeded', job: terminalJob(), timings: terminalJob().timings });++ assert.ok(await db2.maybe('SELECT id FROM pr_bindings WHERE task_id = $1', [task.id]), 'binding is stored even with the switch off');++ const before = submissions.length;++ const body = JSON.stringify({ event: redCheck(), seq: 1 });++ const res = await t2.get('/internal/events', { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });++ assert.strictEqual(res.json().outcome, 'ignored:follow_disabled', res.text);++ assert.strictEqual(submissions.length, before);++ } finally { await t2.close(); }++ });++++ await check('parsePrUrl and the update guards', async () => {++ assert.deepStrictEqual(parsePrUrl(PR_URL), { repo: 'openvibers/openvibe.codes', number: 999 });++ assert.strictEqual(parsePrUrl('https://github.com/o/n/issues/1'), null);++ assert.strictEqual(parsePrUrl('not a url'), null);++ assert.deepStrictEqual([...RED_CONCLUSIONS].sort(), ['action_required', 'failure', 'timed_out']);++ assert.strictEqual(MAX_FIXES_PER_PR, 3);++ const { STOP_RE } = require('../server/domain/pr-bindings');++ assert.ok(STOP_RE.test('stop'));++ assert.ok(STOP_RE.test('Stop.'));++ assert.ok(STOP_RE.test('stop — I will take this one from here'));++ assert.ok(!STOP_RE.test('stop the flaky retry loop, it is noisy'));++ assert.throws(() => parseArgs(JSON.stringify({ repo: REPO, prompt: 'x', push: { branch: 'b', ref: undefined, pull_request: true, update: true } })), /never opens a new pull request/);++ assert.throws(() => parseArgs(JSON.stringify({ repo: REPO, ref: 'main', prompt: 'x', push: { branch: 'other', update: true } })), /branch === ref/);++ const ok = parseArgs(JSON.stringify({ repo: REPO, ref: 'b', prompt: 'x', push: { branch: 'b', update: true } }));++ assert.strictEqual(ok.push.update, true);++ assert.strictEqual(ok.push.pull_request, false);++ });++++ await check('the worker never merges its own PR', async () => {++ const { pushTarget } = require('../harness/job');++ // An update push goes only to the branch it cloned; anything else is refused.++ assert.deepStrictEqual(pushTarget('b', 'b', { update: true, exists: true }), { branch: 'b' });++ assert.ok(pushTarget('b', 'b', { update: true, exists: false }).refused, 'a follow-up refuses a vanished branch');++ assert.deepStrictEqual(pushTarget('b', 'main', { exists: false }), { branch: 'b' }, 'a new branch still pushes as before');++ assert.ok(pushTarget('b', 'b', { exists: false }).refused, 'without update the old refusal stands');++ // The harness has no merge step: no git merge, no merge API call anywhere in the job path.++ const fs = require('fs');++ const jobSrc = fs.readFileSync(require('path').join(__dirname, '..', 'harness', 'job.js'), 'utf8');++ assert.ok(!/git\(\s*\['merge'/.test(jobSrc), 'no git merge in harness/job.js');++ assert.ok(!/\/merges?\b/.test(jobSrc.replace(/mergeable_state/g, '')), 'no merge API in harness/job.js');++ });++++ } finally {++ await t.close();++ }++ done();++})();
@@ -30,6 +30,16 @@ OV_OAUTH_CLIENT_SECRET=
@@ -30,6 +30,16 @@ OV_OAUTH_CLIENT_SECRET=
# optional: defaults to true in production
# optional: defaults to true in production
# COOKIE_SECURE=true
# COOKIE_SECURE=true
+# OpenVibe.Events (AL5): Codes subscribes to run.job.* and, when PR follows are on, github.* at boot, and reads
+# the signed deliveries at POST /internal/events (CODES_EVENTS_SECRET verifies them; 32 characters or more,
+# comma-separated while rotating). Codes publishes codes.task.* here.
+#CODES_EVENTS_URL=https://openvibe.events
+#CODES_EVENTS_SECRET=
+# A ticket follows its pull request (AL5), off by default: with 1, a finished task that pushed a PR is bound to it
+# and a red check, a merge conflict or a review comment wakes a follow-up on the PR branch (at most 3 automatic
+# fixes per PR; a person's "stop" ends it; the worker never merges). Also requires the Events secret above.
+#CODES_PR_FOLLOW_ENABLED=0
+
# OpenVibe.AI (Codes C3). CODES_AI_URL is what a harness on a person's machine calls; without a values file it is
# OpenVibe.AI (Codes C3). CODES_AI_URL is what a harness on a person's machine calls; without a values file it is
# https://ai.openvibe.services.
# https://ai.openvibe.services.
#CODES_AI_URL=https://ai.openvibe.services
#CODES_AI_URL=https://ai.openvibe.services
@@ -5,14 +5,15 @@
@@ -5,14 +5,15 @@
* worker runs it in its sandbox as artifact `[email protected]`, JOB_PROTOCOL below). The job's input is one JSON
* worker runs it in its sandbox as artifact `[email protected]`, JOB_PROTOCOL below). The job's input is one JSON
* line on stdin:
* line on stdin:
*
*
- * { repo, ref?, prompt, agent?, model?, permission?: 'edit'|'full', push?: { branch?, pull_request? } }
+ * { repo, ref?, prompt, agent?, model?, permission?: 'edit'|'full', push?: { branch?, pull_request?, update? } }
*
*
* It clones `repo` (https only) into a scratch directory, runs the harness there with hand-offs, and prints one compact
* It clones `repo` (https only) into a scratch directory, runs the harness there with hand-offs, and prints one compact
* JSON line per event on stdout (what the page shows live). The result goes to fd 3 (the Link's result channel), or as
* JSON line per event on stdout (what the page shows live). The result goes to fd 3 (the Link's result channel), or as
* a last `{ t: 'result_doc' }` line when fd 3 is not open: what the agents said, the changed files and the patch, the
* a last `{ t: 'result_doc' }` line when fd 3 is not open: what the agents said, the changed files and the patch, the
* per-turn checkpoints (refs and the authored/imported turn diff, `checkpoints` in the document), and, when `push` is
* per-turn checkpoints (refs and the authored/imported turn diff, `checkpoints` in the document), and, when `push` is
* asked and the machine holds GITHUB_TOKEN, the new branch it pushed (never the base branch) and its pull request.
* asked and the machine holds GITHUB_TOKEN, the new branch it pushed (never the base branch) and its pull request.
- * Nothing here ever prints a token.
+ * A follow-up (AL5) sets push.update with branch === ref: the checked-out PR branch is committed and pushed again
+ * (never forced, no new PR) so the same pull request turns green. Nothing here ever prints a token.
*/
*/
const fs = require('fs');
const fs = require('fs');
const os = require('os');
const os = require('os');
@@ -60,14 +61,16 @@ const PROTECTED = /^(main|master|trunk|dev|develop|development|staging|stage|pro
@@ -60,14 +61,16 @@ const PROTECTED = /^(main|master|trunk|dev|develop|development|staging|stage|pro
/**
/**
* The branch a task may push: a new one. → { branch } or { refused }. `exists` is GitHub's answer for that name (true
* The branch a task may push: a new one. → { branch } or { refused }. `exists` is GitHub's answer for that name (true
- * when it already exists, null when GitHub could not be asked: refused too, never assumed new).
+ * when it already exists, null when GitHub could not be asked: refused too, never assumed new). A follow-up (AL5)
+ * passes update with branch === ref: the PR branch already exists and is pushed again, never forced.
*/
*/
-function pushTarget(requested, base, { exists = false, now = Date.now() } = {}) {
+function pushTarget(requested, base, { exists = false, now = Date.now(), update = false } = {}) {
const branch = requested || `openvibe-codes/${now.toString(36)}`;
const branch = requested || `openvibe-codes/${now.toString(36)}`;
- if (branch === base) return { refused: 'a task never pushes to the branch it started from' };
+ if (branch === base && !update) return { refused: 'a task never pushes to the branch it started from' };
if (PROTECTED.test(branch)) return { refused: `a task pushes only a new branch of its own, never ${branch}` };
if (PROTECTED.test(branch)) return { refused: `a task pushes only a new branch of its own, never ${branch}` };
if (exists === null) return { refused: 'GitHub could not say whether the branch exists, so nothing was pushed' };
if (exists === null) return { refused: 'GitHub could not say whether the branch exists, so nothing was pushed' };
- if (exists) return { refused: `${branch} already exists: a task pushes only a new branch` };
+ if (exists && !update) return { refused: `${branch} already exists: a task pushes only a new branch` };
+ if (update && !exists) return { refused: `${branch} is gone: nothing to update` };
return { branch };
return { branch };
}
}
@@ -235,9 +238,15 @@ function parseArgs(raw, env = process.env) {
@@ -235,9 +238,15 @@ function parseArgs(raw, env = process.env) {
const push = a.push && typeof a.push === 'object' ? {
const push = a.push && typeof a.push === 'object' ? {
branch: a.push.branch == null ? null : String(a.push.branch),
branch: a.push.branch == null ? null : String(a.push.branch),
pull_request: a.push.pull_request === true,
pull_request: a.push.pull_request === true,
+ // Only a follow-up (AL5) sets update; a normal task's args keep their old shape.
+ ...(a.push.update === true ? { update: true } : {}),
} : null;
} : null;
if (push && push.branch && (!BRANCH.test(push.branch) || push.branch.startsWith('-'))) throw new Error('push.branch must be a branch name');
if (push && push.branch && (!BRANCH.test(push.branch) || push.branch.startsWith('-'))) throw new Error('push.branch must be a branch name');
if (push && push.branch && PROTECTED.test(push.branch)) throw new Error(`push.branch must be a new branch of the task's own, never ${push.branch}`);
if (push && push.branch && PROTECTED.test(push.branch)) throw new Error(`push.branch must be a new branch of the task's own, never ${push.branch}`);
+ // A follow-up (AL5) pushes the checked-out PR branch again: it names the branch it cloned (branch === ref),
+ // never asks for a new pull request, and is never forced. Anything else with update is rejected.
+ if (push && push.update && push.pull_request) throw new Error('push.update never opens a new pull request');
+ if (push && push.update && (!push.branch || push.branch !== ref)) throw new Error('push.update needs branch === ref: a follow-up pushes the branch it cloned');
// The session a resume continues: the job's own input, or RESUME in the environment (a task carried on in the
// The session a resume continues: the job's own input, or RESUME in the environment (a task carried on in the
// same conversation). It is a harness's session id, never a path and never a credential.
// same conversation). It is a harness's session id, never a path and never a credential.
const resumeIn = a.resume == null || a.resume === '' ? env.RESUME : a.resume;
const resumeIn = a.resume == null || a.resume === '' ? env.RESUME : a.resume;
@@ -606,19 +615,22 @@ async function pushBranch(dir, job, { env, fetchImpl, base }) {
@@ -606,19 +615,22 @@ async function pushBranch(dir, job, { env, fetchImpl, base }) {
if (host && giteaHosts(env).has(host)) return pushForge(dir, job, { env, fetchImpl, base });
if (host && giteaHosts(env).has(host)) return pushForge(dir, job, { env, fetchImpl, base });
}
}
if (!token || !gh) return { pushed: false, reason: !gh ? 'push is for github.com repositories' : 'this machine has no GITHUB_TOKEN' };
if (!token || !gh) return { pushed: false, reason: !gh ? 'push is for github.com repositories' : 'this machine has no GITHUB_TOKEN' };
- const first = pushTarget(job.push.branch, base);
+ const update = job.push.update === true;
+ const first = pushTarget(job.push.branch, base, { update });
if (first.refused) return { pushed: false, reason: first.refused };
if (first.refused) return { pushed: false, reason: first.refused };
const branch = first.branch;
const branch = first.branch;
const remoteUrl = `https://x-access-token:${token}@github.com/${gh.owner}/${gh.name}.git`;
const remoteUrl = `https://x-access-token:${token}@github.com/${gh.owner}/${gh.name}.git`;
let exists = null;
let exists = null;
try { exists = (await git(['ls-remote', '--heads', remoteUrl, `refs/heads/${branch}`], { cwd: dir, timeout: 60_000 })).trim() !== ''; } catch { exists = null; }
try { exists = (await git(['ls-remote', '--heads', remoteUrl, `refs/heads/${branch}`], { cwd: dir, timeout: 60_000 })).trim() !== ''; } catch { exists = null; }
- const checked = pushTarget(branch, base, { exists });
+ const checked = pushTarget(branch, base, { exists, update });
if (checked.refused) return { pushed: false, reason: checked.refused };
if (checked.refused) return { pushed: false, reason: checked.refused };
const who = {
const who = {
GIT_AUTHOR_NAME: env.GIT_AUTHOR_NAME || 'OpenVibe.Codes', GIT_AUTHOR_EMAIL: env.GIT_AUTHOR_EMAIL || '[email protected]',
GIT_AUTHOR_NAME: env.GIT_AUTHOR_NAME || 'OpenVibe.Codes', GIT_AUTHOR_EMAIL: env.GIT_AUTHOR_EMAIL || '[email protected]',
GIT_COMMITTER_NAME: env.GIT_AUTHOR_NAME || 'OpenVibe.Codes', GIT_COMMITTER_EMAIL: env.GIT_AUTHOR_EMAIL || '[email protected]',
GIT_COMMITTER_NAME: env.GIT_AUTHOR_NAME || 'OpenVibe.Codes', GIT_COMMITTER_EMAIL: env.GIT_AUTHOR_EMAIL || '[email protected]',
};
};
- await git(['checkout', '-b', branch], { cwd: dir });
+ // A follow-up is already on the PR branch it cloned (branch === ref === base): committing there and pushing
+ // (never forced, so a branch someone rewrote meanwhile is rejected, not overwritten).
+ if (branch !== base) await git(['checkout', '-b', branch], { cwd: dir });
await git(['commit', '-m', job.prompt.split('\n')[0].slice(0, 72) || 'OpenVibe.Codes task'], { cwd: dir, env: who });
await git(['commit', '-m', job.prompt.split('\n')[0].slice(0, 72) || 'OpenVibe.Codes task'], { cwd: dir, env: who });
await git(['push', remoteUrl, `HEAD:refs/heads/${branch}`], { cwd: dir, timeout: 180_000 });
await git(['push', remoteUrl, `HEAD:refs/heads/${branch}`], { cwd: dir, timeout: 180_000 });
const out = { pushed: true, branch, branch_url: `https://github.com/${gh.owner}/${gh.name}/tree/${encodeURIComponent(branch)}` };
const out = { pushed: true, branch, branch_url: `https://github.com/${gh.owner}/${gh.name}/tree/${encodeURIComponent(branch)}` };
new file mode 100644
new file mode 100644
@@ -0,0 +1,23 @@
@@ -0,0 +1,23 @@
+-- phase: expand
+-- AL5: a ticket follows its pull request. One row per ticket (task) bound to a PR: which repo and PR number, the
+-- role it plays there, how the PR was found, and the automatic fixes so far. A red check, a conflict or a review
+-- comment on the PR wakes the same worker (a follow-up Run job on the PR branch); at most 3 automatic fixes per
+-- PR; a person's "stop" or a merged/closed PR ends the binding. Nothing here holds a secret.
+CREATE TABLE pr_bindings (
+ id text COLLATE "C" PRIMARY KEY, -- prb_<ULID>
+ task_id text COLLATE "C" NOT NULL, -- tsk_… the ticket's current task (moves to each follow-up)
+ repo text COLLATE "C" NOT NULL, -- owner/name, lowercased
+ pr_number integer NOT NULL CHECK (pr_number > 0),
+ role text COLLATE "C" NOT NULL, -- review|verify|fix
+ how_found text COLLATE "C" NOT NULL, -- created|worktree|review
+ branch text COLLATE "C", -- the PR branch a follow-up pushes to (update, never a new PR)
+ state text COLLATE "C" NOT NULL DEFAULT 'active', -- active|ended
+ fix_count integer NOT NULL DEFAULT 0, -- automatic fixes this binding's task lineage used
+ ended_reason text COLLATE "C", -- merged|closed|stopped_by_person
+ last_wake_at bigint,
+ last_wake_reason text COLLATE "C",
+ created_at bigint NOT NULL,
+ updated_at bigint NOT NULL
+);
+CREATE INDEX IF NOT EXISTS pr_bindings_pr ON pr_bindings (repo, pr_number);
+CREATE INDEX IF NOT EXISTS pr_bindings_task ON pr_bindings (task_id);
@@ -157,6 +157,10 @@ async function createApp(opts = {}) {
@@ -157,6 +157,10 @@ async function createApp(opts = {}) {
fetchImpl: fetchImpl || globalThis.fetch, fleet, userSettings, harnesses, attachments,
fetchImpl: fetchImpl || globalThis.fetch, fleet, userSettings, harnesses, attachments,
});
});
tasks.setConductor(ctx.conductor);
tasks.setConductor(ctx.conductor);
+ // A ticket follows its pull request (AL5): bindings need tasks (a follow-up is a task), and tasks need
+ // bindings (a terminal result with a PR binds it) — wired to each other here, not in a constructor.
+ ctx.prBindings = require('./domain/pr-bindings').createPrBindings({ store, tasks, outbox, config, log });
+ tasks.attachBindings(ctx.prBindings);
const app = express();
const app = express();
aiCatalogConfig.mount(app);
aiCatalogConfig.mount(app);
@@ -233,7 +237,7 @@ async function createApp(opts = {}) {
@@ -233,7 +237,7 @@ async function createApp(opts = {}) {
const accountSend = opts.accountSend || (config.oauth.clientSecret
const accountSend = opts.accountSend || (config.oauth.clientSecret
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl || globalThis.fetch })
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl || globalThis.fetch })
: null);
: null);
- const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, log });
+ const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, bindings: ctx.prBindings, log });
ctx.accountData = accountData;
ctx.accountData = accountData;
ctx.eventsConsumer = consumer;
ctx.eventsConsumer = consumer;
ctx.publisher = createPublisher({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });
ctx.publisher = createPublisher({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });
@@ -127,6 +127,12 @@ function load(env = process.env) {
@@ -127,6 +127,12 @@ function load(env = process.env) {
deadAfterRefusals: Math.max(1, int(env.CODES_EVENTS_DEAD_AFTER_REFUSALS, 3)),
deadAfterRefusals: Math.max(1, int(env.CODES_EVENTS_DEAD_AFTER_REFUSALS, 3)),
deadAfterAttempts: Math.max(1, int(env.CODES_EVENTS_DEAD_AFTER_ATTEMPTS, 20)),
deadAfterAttempts: Math.max(1, int(env.CODES_EVENTS_DEAD_AFTER_ATTEMPTS, 20)),
},
},
+ // A ticket follows its pull request (AL5): a red check, a conflict or a review comment wakes the bound
+ // ticket's worker with a follow-up Run job, at most 3 automatic fixes per PR. Default off: set
+ // CODES_PR_FOLLOW_ENABLED=1 to let Codes start follow-ups on github.* deliveries.
+ prFollow: {
+ enabled: env.CODES_PR_FOLLOW_ENABLED === '1',
+ },
// Session lines in PostgreSQL: no total cap; finished tasks keep theirs for this many days (default 90).
// Session lines in PostgreSQL: no total cap; finished tasks keep theirs for this many days (default 90).
sessions: {
sessions: {
retentionDays: Math.max(1, int(env.CODES_SESSION_RETENTION_DAYS, 90)),
retentionDays: Math.max(1, int(env.CODES_SESSION_RETENTION_DAYS, 90)),
new file mode 100644
new file mode 100644
@@ -0,0 +1,279 @@
@@ -0,0 +1,279 @@
+'use strict';
+
+/**
+ * A ticket follows its pull request (AL5).
+ *
+ * A binding is one row per ticket (task): the repo, the PR number, the role this ticket plays there
+ * (review|verify|fix) and how the PR was found (created|worktree|review). A red check, a merge conflict or
+ * a review comment on the PR wakes the SAME worker: a follow-up Run job on the PR branch, with the failing
+ * evidence in its prompt. At most 3 automatic fixes per PR; a person's "stop" ends the binding; a merged or
+ * closed PR ends it too. A bot's comment or a CI result never wakes an ended binding; a person's comment can
+ * (it re-opens the binding and wakes it once). The worker never merges its own PR: a follow-up pushes the
+ * same branch again (push.update, never a force push) and opens no new pull request; the harness has no merge
+ * step at all.
+ *
+ * No polling: GitHub webhooks arrive through OpenVibe.Events (server/github.js there) as github.* envelopes,
+ * delivered to Codes' /internal/events consumer, which calls handleGithubEvent here. The binding follows the
+ * newest task: after a follow-up starts, the binding's task_id moves to it, so the next wake carries the same
+ * ticket forward. Every bind, follow-up and end also writes a codes.pr.* outbox row (bound|followup|ended),
+ * so the portal can show the ticket without reading this table.
+ *
+ * The switch is default-off: with CODES_PR_FOLLOW_ENABLED unset, red checks are recorded (ignored:
+ * follow_disabled) and no follow-up starts.
+ *
+ * const bindings = createPrBindings({ store, tasks, outbox, config, log });
+ * await bindings.maybeBind(taskRow); // after a task ends with a PR: idempotent, in-tx safe
+ * await bindings.handleGithubEvent(envelope); // → { outcome, task_ids, followed_up }
+ */
+const { parseArgs } = require('../../harness/job');
+
+const ROLES = new Set(['review', 'verify', 'fix']);
+const MAX_FIXES_PER_PR = 3;
+/** Check conclusions that wake a verify|fix binding (a cancelled run is usually superseded, not broken). */
+const RED_CONCLUSIONS = new Set(['failure', 'timed_out', 'action_required']);
+/** A person's comment that is only "stop" (or starts "stop:" / "stop —") ends the binding instead of waking it.
+ * A sentence that happens to begin with the word ("stop the flaky retry loop") is a normal comment. */
+const STOP_RE = /^\s*stop\s*(?:[:—–-]\s*.*)?[.!]?\s*$/i;
+const MAX_PROMPT = 16000;
+const MAX_BODY = 4000;
+const MAX_EVIDENCE = 6000;
+
+const slug = (v) => (typeof v === 'string' ? v.toLowerCase() : null);
+const num = (v) => (Number.isInteger(v) && v > 0 ? v : null);
+const str = (v, n) => (typeof v === 'string' ? v.slice(0, n) : null);
+
+/** owner/name and the PR number of a github.com pull URL, or null. */
+function parsePrUrl(url) {
+ if (typeof url !== 'string') return null;
+ const m = /^https:\/\/github\.com\/([A-Za-z0-9-]+)\/([A-Za-z0-9._-]+)\/pull\/(\d+)(?:\/|$)/.exec(url.trim());
+ if (!m) return null;
+ return { repo: `${m[1]}/${m[2]}`.toLowerCase(), number: Number(m[3]) };
+}
+
+function createPrBindings({ store, tasks, outbox = null, config = {}, log = console } = {}) {
+ if (!store) throw new TypeError('pr-bindings need a store');
+ const db = store.db;
+ const enabled = Boolean(config.prFollow && config.prFollow.enabled);
+
+ const emit = async (topic, subject, payload) => {
+ if (!outbox || typeof outbox.emitPr !== 'function' || !subject) return null;
+ try { return await outbox.emitPr(topic, subject, payload); } catch (e) { log.warn(`[Codes] pr-bindings outbox: ${e.message}`); return null; }
+ };
+
+ /**
+ * Bind a finished task to the PR its result pushed (role fix, found when created). Idempotent: a task
+ * binds its PR once. Uses the ambient transaction when called from run-tasks' recordState.
+ */
+ async function maybeBind(row) {
+ if (!row || !row.id || !row.result) return null;
+ let result = row.result;
+ if (typeof result === 'string') { try { result = JSON.parse(result); } catch { return null; } }
+ const push = result && typeof result.push === 'object' ? result.push : null;
+ if (!push || push.pushed !== true || typeof push.pull_request !== 'string') return null;
+ const pr = parsePrUrl(push.pull_request);
+ if (!pr) return null;
+ const branch = str(push.branch, 200);
+ const seen = await db.maybe(
+ 'SELECT id FROM pr_bindings WHERE task_id = $1 AND repo = $2 AND pr_number = $3 AND state = $4',
+ [row.id, pr.repo, pr.number, 'active'],
+ );
+ if (seen) return seen;
+ const now = store.now();
+ const bound = await db.one(
+ `INSERT INTO pr_bindings (id, task_id, repo, pr_number, role, how_found, branch, state, fix_count, created_at, updated_at)
+ VALUES ($1, $2, $3, $4, 'fix', 'created', $5, 'active', 0, $6, $6) RETURNING *`,
+ [store.newId('prb'), row.id, pr.repo, pr.number, branch, now],
+ );
+ await emit('codes.pr.bound',
+ row.subject, { task_id: row.id, repo: pr.repo, pr_number: pr.number, role: 'fix', branch });
+ return bound;
+ }
+
+ const forPr = (repo, number, { activeOnly = false } = {}) => db.many(
+ `SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2${activeOnly ? ` AND state = 'active'` : ''} ORDER BY created_at ASC`,
+ [repo, number],
+ );
+
+ async function endAll(rows, reason) {
+ for (const b of rows) {
+ if (b.state === 'ended') continue;
+ await db.exec(`UPDATE pr_bindings SET state = 'ended', ended_reason = $2, updated_at = $3 WHERE id = $1`,
+ [b.id, reason, store.now()]);
+ const owner = await db.maybe('SELECT subject FROM run_tasks WHERE id = $1', [b.task_id]);
+ await emit('codes.pr.ended', owner && owner.subject,
+ { task_id: b.task_id, repo: b.repo, pr_number: b.pr_number, reason });
+ }
+ }
+
+ const fixesForPr = async (repo, number) => {
+ const r = await db.maybe('SELECT COALESCE(SUM(fix_count), 0) AS n FROM pr_bindings WHERE repo = $1 AND pr_number = $2', [repo, number]);
+ return Number((r && r.n) || 0);
+ };
+
+ /** The follow-up prompt: the failing evidence first, the original task after, the no-merge rule in both. */
+ function followupPrompt({ binding, parent, evidence, fixN, automatic = true }) {
+ const lead = automatic
+ ? `Automatic follow-up ${fixN} of ${MAX_FIXES_PER_PR}`
+ : 'Follow-up requested by a person';
+ const head = `${lead} for ${binding.repo} pull request #${binding.pr_number} (ticket task ${parent.id}).\n\nEvidence from the PR:\n${evidence}\n\nThe branch is ${binding.branch || '(the PR branch)'}: it is already checked out, fix it there, and push the same branch again — no new branch, no new pull request, and never merge anything yourself.`;
+ const tail = `\n\nThe original task was:\n${typeof parent.prompt === 'string' ? parent.prompt : ''}`;
+ return (head + tail).slice(0, MAX_PROMPT);
+ }
+
+ /**
+ * Start one follow-up Run job for the binding's ticket: same repo, the PR branch checked out and pushed
+ * again (update, non-force), no new PR. The binding moves to the new task. Throws when Run refuses.
+ */
+ async function startFollowup(binding, evidence, { automatic = true } = {}) {
+ const parent = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [binding.task_id]);
+ if (!parent) throw new Error(`ticket task ${binding.task_id} is gone`);
+ if (!binding.branch) throw new Error('the binding names no branch to push again');
+ const fixes = await fixesForPr(binding.repo, binding.pr_number);
+ const prompt = followupPrompt({ binding, parent, evidence, fixN: fixes + 1, automatic });
+ // The worker's own harness session, when the task's last attempt recorded one: the follow-up continues the
+ // same conversation instead of re-reading the repository from nothing. A machine without it falls back to
+ // the prompt (harness/job.js treats an unknown resume as the agent's own error and hands off normally).
+ const last = await db.maybe('SELECT agent, session FROM codes_attempts WHERE task_id = $1 ORDER BY n DESC LIMIT 1', [parent.id]);
+ const agent = (last && last.agent) || parent.agent || null;
+ const resume = last && typeof last.session === 'string' && last.session ? last.session : null;
+ const args = parseArgs(JSON.stringify({
+ repo: parent.repo, ref: binding.branch, prompt, agent,
+ permission: 'edit',
+ push: { branch: binding.branch, pull_request: false, update: true },
+ ...(resume ? { resume } : {}),
+ }));
+ let actorRef = null;
+ if (parent.actor_ref) { try { const p = typeof parent.actor_ref === 'string' ? JSON.parse(parent.actor_ref) : parent.actor_ref; if (p && typeof p === 'object') actorRef = p; } catch { actorRef = null; } }
+ const task = await tasks.start({ subject: parent.subject, project: parent.project_id, args, via: parent.via, key: null, actorRef });
+ await db.exec(`UPDATE pr_bindings SET task_id = $2, fix_count = fix_count + $5, last_wake_at = $3, last_wake_reason = $4, updated_at = $3 WHERE id = $1`,
+ [binding.id, task.id, store.now(), evidence.slice(0, 120), automatic ? 1 : 0]);
+ await emit('codes.pr.followup', parent.subject,
+ { task_id: task.id, parent_task_id: parent.id, repo: binding.repo, pr_number: binding.pr_number, fix_n: fixes + 1, automatic });
+ return task;
+ }
+
+ /** One wake across the PR's active bindings of the given roles. → { started, skipped } of task ids. */
+ async function wake(rows, { roles, evidence, reason, eventId, automatic = true }) {
+ const started = [];
+ const skipped = [];
+ if (!enabled) return { started, skipped, disabled: true };
+ // The cap bounds automatic fixes only: a person's comment is a new instruction and may start a manual one.
+ if (automatic && await fixesForPr(rows[0].repo, rows[0].pr_number) >= MAX_FIXES_PER_PR) return { started, skipped, capped: true };
+ for (const b of rows) {
+ if (!roles.has(b.role)) { skipped.push(b.task_id); continue; }
+ const current = await db.maybe('SELECT id, state FROM run_tasks WHERE id = $1', [b.task_id]);
+ if (current && !['succeeded', 'failed', 'cancelled', 'expired'].includes(current.state)) { skipped.push(b.task_id); continue; }
+ try {
+ const task = await startFollowup(b, evidence, { automatic });
+ started.push(task.id);
+ log.info(`[Codes] pr-follow ${b.repo}#${b.pr_number} woke ${task.id} (${reason}, ${eventId})`);
+ } catch (e) {
+ log.warn(`[Codes] pr-follow ${b.repo}#${b.pr_number} failed to wake: ${e.message}`);
+ skipped.push(b.task_id);
+ }
+ }
+ return { started, skipped };
+ }
+
+ /**
+ * A github.* envelope (already inbox-deduped by the consumer). → { outcome, task_ids, followed_up }.
+ * outcome is what the consumer records: woke, stopped, reopened, ended:<reason>, ignored:<why>.
+ */
+ async function handleGithubEvent(event) {
+ if (!event || typeof event.event_type !== 'string') return { outcome: 'ignored:payload', task_ids: [], followed_up: [] };
+ const subject = event.subject && typeof event.subject === 'object' ? event.subject : null;
+ const m = subject && typeof subject.id === 'string' ? /^([a-z0-9-]+(?:\/[a-z0-9._-]+))#(\d+)$/.exec(subject.id) : null;
+ if (!m) return { outcome: 'ignored:payload', task_ids: [], followed_up: [] };
+ const repo = m[1];
+ const number = Number(m[2]);
+ const payload = event.payload && typeof event.payload === 'object' ? event.payload : {};
+ // The subject is the routing key (a check fans out one envelope per PR, so its payload names no number);
+ // a payload that names another repo or PR is not ours.
+ if (payload.repo !== undefined && slug(payload.repo) !== repo) return { outcome: 'ignored:payload', task_ids: [], followed_up: [] };
+ if (payload.pr_number !== undefined && num(payload.pr_number) !== number) return { outcome: 'ignored:payload', task_ids: [], followed_up: [] };
+ const rows = await forPr(repo, number);
+ if (!rows.length) return { outcome: 'ignored:no_binding', task_ids: [], followed_up: [] };
+ const active = rows.filter((b) => b.state === 'active');
+ const isPerson = payload.author_type === 'User';
+
+ if (event.event_type === 'github.check.completed') {
+ if (!RED_CONCLUSIONS.has(payload.conclusion)) return { outcome: 'ignored:green', task_ids: rows.map((b) => b.task_id), followed_up: [] };
+ if (!active.length) return { outcome: 'ignored:ended', task_ids: rows.map((b) => b.task_id), followed_up: [] };
+ const out = payload.output && typeof payload.output === 'object' ? payload.output : null;
+ const parts = [`check "${payload.check_name || 'ci'}" concluded "${payload.conclusion}"`
+ + `${payload.head_sha ? ` at ${String(payload.head_sha).slice(0, 12)}` : ''}`
+ + `${payload.html_url ? ` (${payload.html_url})` : ''}`];
+ if (out && out.title) parts.push(`title: ${String(out.title).slice(0, 500)}`);
+ if (out && out.summary) parts.push(`summary: ${String(out.summary).slice(0, 2000)}`);
+ if (out && out.text) parts.push(`log tail:\n${String(out.text).slice(0, 4000)}`);
+ if (payload.details_url) parts.push(`details: ${String(payload.details_url).slice(0, 500)}`);
+ parts.push('— find why it failed on this branch and fix it.');
+ const evidence = parts.join('\n').slice(0, MAX_EVIDENCE);
+ const r = await wake(active, { roles: new Set(['verify', 'fix']), evidence, reason: 'red-check', eventId: event.event_id });
+ if (r.disabled) return { outcome: 'ignored:follow_disabled', task_ids: active.map((b) => b.task_id), followed_up: [] };
+ if (r.capped) return { outcome: 'ignored:max_fixes', task_ids: active.map((b) => b.task_id), followed_up: [] };
+ return { outcome: r.started.length ? 'woke' : 'ignored:fix_running', task_ids: active.map((b) => b.task_id), followed_up: r.started };
+ }
+
+ if (event.event_type === 'github.pr.updated') {
+ if (payload.action === 'closed') {
+ const reason = payload.merged === true ? 'merged' : 'closed';
+ await endAll(rows, reason);
+ return { outcome: `ended:${reason}`, task_ids: rows.map((b) => b.task_id), followed_up: [] };
+ }
+ if (payload.mergeable_state === 'dirty' || payload.mergeable === false) {
+ if (!active.length) return { outcome: 'ignored:ended', task_ids: rows.map((b) => b.task_id), followed_up: [] };
+ const where = [payload.head_ref ? `branch ${payload.head_ref}` : null, payload.base ? `onto ${payload.base}` : null].filter(Boolean).join(' ');
+ const evidence = [`the pull request reports mergeable_state "${payload.mergeable_state || 'dirty'}"`
+ + `${where ? ` (${where})` : ''}`
+ + `${payload.changed_files ? `, ${payload.changed_files} file${payload.changed_files === 1 ? '' : 's'} changed` : ''}`
+ + `${payload.html_url ? ` (${payload.html_url})` : ''}`,
+ payload.files_url ? `the changed files: ${payload.files_url}` : null,
+ '— fetch the base branch, run `git merge-tree` (or a local merge) to list the conflicted files, resolve them on this branch, and push it again.']
+ .filter(Boolean).join('\n').slice(0, MAX_EVIDENCE);
+ const r = await wake(active, { roles: new Set(['fix']), evidence, reason: 'conflict', eventId: event.event_id });
+ if (r.disabled) return { outcome: 'ignored:follow_disabled', task_ids: active.map((b) => b.task_id), followed_up: [] };
+ if (r.capped) return { outcome: 'ignored:max_fixes', task_ids: active.map((b) => b.task_id), followed_up: [] };
+ return { outcome: r.started.length ? 'woke' : 'ignored:fix_running', task_ids: active.map((b) => b.task_id), followed_up: r.started };
+ }
+ return { outcome: 'ignored:no_conflict', task_ids: rows.map((b) => b.task_id), followed_up: [] };
+ }
+
+ if (event.event_type === 'github.pr.commented') {
+ const body = str(payload.body, MAX_BODY) || '';
+ if (STOP_RE.test(body) && isPerson) {
+ if (!active.length) return { outcome: 'ignored:ended', task_ids: rows.map((b) => b.task_id), followed_up: [] };
+ await endAll(active, 'stopped_by_person');
+ return { outcome: 'stopped', task_ids: rows.map((b) => b.task_id), followed_up: [] };
+ }
+ if (!isPerson) return { outcome: active.length ? 'ignored:bot' : 'ignored:ended', task_ids: rows.map((b) => b.task_id), followed_up: [] };
+ // A person's comment: on an ended binding it re-opens the ticket and wakes it once; otherwise it is
+ // review feedback for the active bindings. Never past the per-PR fix cap.
+ let targets = active;
+ let reopened = false;
+ if (!targets.length) {
+ const last = rows[rows.length - 1];
+ await db.exec(`UPDATE pr_bindings SET state = 'active', ended_reason = NULL, updated_at = $2 WHERE id = $1`, [last.id, store.now()]);
+ targets = [{ ...last, state: 'active' }];
+ reopened = true;
+ }
+ const where = payload.path ? ` on ${payload.path}${payload.line ? `:${payload.line}` : ''}` : '';
+ const evidence = `${payload.author || 'someone'} commented${where} on the pull request${payload.html_url ? ` (${payload.html_url})` : ''}:\n${body}\n— address it on the branch and push it again.`;
+ // A person's comment is manual: it can start a fix past the automatic cap.
+ const r = await wake(targets, { roles: new Set(['review', 'fix']), evidence, reason: 'comment', eventId: event.event_id, automatic: false });
+ if (r.disabled) return { outcome: 'ignored:follow_disabled', task_ids: targets.map((b) => b.task_id), followed_up: [] };
+ if (r.capped) return { outcome: 'ignored:max_fixes', task_ids: targets.map((b) => b.task_id), followed_up: [] };
+ if (!r.started.length) return { outcome: 'ignored:fix_running', task_ids: targets.map((b) => b.task_id), followed_up: [] };
+ return { outcome: reopened ? 'reopened' : 'woke', task_ids: targets.map((b) => b.task_id), followed_up: r.started };
+ }
+
+ return { outcome: 'ignored:type', task_ids: rows.map((b) => b.task_id), followed_up: [] };
+ }
+
+ /** This ticket's bindings, newest first (for pages and the portal). */
+ const listForTask = (taskId) => db.many('SELECT * FROM pr_bindings WHERE task_id = $1 ORDER BY created_at DESC', [taskId]);
+
+ return { maybeBind, handleGithubEvent, listForTask, startFollowup, forPr, parsePrUrl, MAX_FIXES_PER_PR, RED_CONCLUSIONS };
+}
+
+module.exports = { createPrBindings, parsePrUrl, MAX_FIXES_PER_PR, RED_CONCLUSIONS, STOP_RE };
@@ -164,6 +164,10 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -164,6 +164,10 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
*/
*/
let conductorTurns = conductor;
let conductorTurns = conductor;
const setConductor = (turns) => { conductorTurns = turns; };
const setConductor = (turns) => { conductorTurns = turns; };
+ // A ticket follows its pull request (AL5): wired by the app after both exist, because a follow-up is a task
+ // started here. A terminal task whose result pushed a PR binds it, in the same transaction.
+ let prBindings = null;
+ const attachBindings = (b) => { prBindings = b || null; };
/** An attempt's state when it ended: a limit reached is a budget-stop, never a plain failure. */
/** An attempt's state when it ended: a limit reached is a budget-stop, never a plain failure. */
const ENDED = new Set(['done', 'failed', 'stopped', 'budget-stopped', 'interrupted']);
const ENDED = new Set(['done', 'failed', 'stopped', 'budget-stopped', 'interrupted']);
@@ -501,6 +505,9 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -501,6 +505,9 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
});
});
}
}
}
}
+ if (TERMINAL.has(updated.state) && prBindings) {
+ try { await prBindings.maybeBind(updated); } catch (e) { log.warn(`[Codes] pr-bind ${updated.id}: ${String((e && e.message) || e).slice(0, 200)}`); }
+ }
return updated;
return updated;
});
});
// What the refused run really spent: OpenVibe.AI counted it (the harness may have stopped mid-turn).
// What the refused run really spent: OpenVibe.AI counted it (the harness may have stopped mid-turn).
@@ -1074,7 +1081,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -1074,7 +1081,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
return {
return {
start, refresh, applyJob, applyEvent, recordState, afterEnd, continueTask, stopTask, proof: taskProof,
start, refresh, applyJob, applyEvent, recordState, afterEnd, continueTask, stopTask, proof: taskProof,
- latestProject, present, who, attempts: attemptStore, setConductor, fundTurn,
+ latestProject, present, who, attempts: attemptStore, setConductor, attachBindings, fundTurn,
setForge(f) { forge = f || null; },
setForge(f) { forge = f || null; },
};
};
}
}
@@ -15,7 +15,11 @@
@@ -15,7 +15,11 @@
* backoff; it is off without EVENTS_URL/CODES_EVENTS_SECRET/the OAuth client secret, under CODES_DRILL, or when
* backoff; it is off without EVENTS_URL/CODES_EVENTS_SECRET/the OAuth client secret, under CODES_DRILL, or when
* CODES_EVENTS_SUBSCRIBE=0.
* CODES_EVENTS_SUBSCRIBE=0.
*
*
- * const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, log });
+ * github.check.completed | github.pr.updated | github.pr.commented
+ * GitHub webhooks through OpenVibe.Events (AL5): routed to the pr-bindings (server/domain/
+ * pr-bindings.js), which wake the bound ticket's worker. The inbox dedupes redeliveries.
+ *
+ * const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, bindings, log });
* app.use('/internal/events', consumer.router);
* app.use('/internal/events', consumer.router);
*/
*/
const express = require('express');
const express = require('express');
@@ -27,7 +31,9 @@ const { TERMINAL, STATES } = require('./domain/run-tasks');
@@ -27,7 +31,9 @@ const { TERMINAL, STATES } = require('./domain/run-tasks');
const RUN_TOPICS = Object.freeze([
const RUN_TOPICS = Object.freeze([
'run.job.queued', 'run.job.started', 'run.job.succeeded', 'run.job.failed', 'run.job.cancelled', 'run.job.expired',
'run.job.queued', 'run.job.started', 'run.job.succeeded', 'run.job.failed', 'run.job.cancelled', 'run.job.expired',
]);
]);
-const TOPICS = Object.freeze([...RUN_TOPICS, ...accountDataLib.TOPICS]);
+/** GitHub through OpenVibe.Events (AL5): one envelope per pull request (server/github.js there). */
+const GITHUB_TOPICS = Object.freeze(['github.check.completed', 'github.pr.updated', 'github.pr.commented']);
+const TOPICS = Object.freeze([...RUN_TOPICS, ...accountDataLib.TOPICS, ...GITHUB_TOPICS]);
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
const JOB_RE = /^job_[0-9A-HJKMNP-TV-Z]{26}$/;
const JOB_RE = /^job_[0-9A-HJKMNP-TV-Z]{26}$/;
const STATE_OF_TOPIC = {
const STATE_OF_TOPIC = {
@@ -42,7 +48,7 @@ function errorText(error) {
@@ -42,7 +48,7 @@ function errorText(error) {
return null;
return null;
}
}
-function createEventsConsumer({ config, store, tasks, run, accountData, accountSend = null, log = console, now = store.now }) {
+function createEventsConsumer({ config, store, tasks, run, accountData, accountSend = null, bindings = null, log = console, now = store.now }) {
const db = store.db;
const db = store.db;
const secrets = (config.events && config.events.secrets ? config.events.secrets : []).filter((x) => typeof x === 'string' && x.length >= 32);
const secrets = (config.events && config.events.secrets ? config.events.secrets : []).filter((x) => typeof x === 'string' && x.length >= 32);
const stats = { received: 0, applied: 0, duplicates: 0, ignored: 0, refused: 0, failed: 0, last_at: null };
const stats = { received: 0, applied: 0, duplicates: 0, ignored: 0, refused: 0, failed: 0, last_at: null };
@@ -128,6 +134,39 @@ function createEventsConsumer({ config, store, tasks, run, accountData, accountS
@@ -128,6 +134,39 @@ function createEventsConsumer({ config, store, tasks, run, accountData, accountS
return res.json({ event_id: event.event_id, outcome });
return res.json({ event_id: event.event_id, outcome });
}
}
+ if (GITHUB_TOPICS.includes(event.event_type)) {
+ // A ticket follows its pull request (AL5). The inbox claims the id first: a redelivery answers
+ // duplicate, and the wake below runs once.
+ const seen = await db.maybe('SELECT task_id, outcome FROM codes_events_inbox WHERE event_id = $1', [event.event_id]);
+ if (seen) { stats.duplicates++; return res.json({ event_id: event.event_id, duplicate: true, outcome: null }); }
+ if (!bindings) {
+ const result = await ignore(event, 'ignored:no_bindings');
+ if (result.duplicate) stats.duplicates++; else stats.ignored++;
+ return res.json({ event_id: event.event_id, duplicate: result.duplicate, outcome: result.outcome });
+ }
+ let result;
+ try {
+ result = await store.tx(async () => {
+ const inserted = await db.maybe(
+ `INSERT INTO codes_events_inbox (event_id, event_type, task_id, outcome, processed_at)
+ VALUES ($1, $2, NULL, 'wake', $3::timestamptz) ON CONFLICT (event_id) DO NOTHING RETURNING event_id`,
+ [event.event_id, event.event_type, store.iso()],
+ );
+ if (!inserted) return { duplicate: true, outcome: null, task_ids: [], followed_up: [] };
+ const r = await bindings.handleGithubEvent(event);
+ await db.exec('UPDATE codes_events_inbox SET outcome = $2 WHERE event_id = $1', [event.event_id, r.outcome]);
+ return { duplicate: false, ...r };
+ });
+ } catch (err) {
+ stats.failed++;
+ log.error(`[Codes] github event ${event.event_id} (${event.event_type}) failed:`, err.message);
+ return problem(503, 'codes.event_failed', 'processing failed; it will be retried');
+ }
+ if (result.duplicate) { stats.duplicates++; return res.json({ event_id: event.event_id, duplicate: true, outcome: null }); }
+ if (String(result.outcome).startsWith('ignored:')) stats.ignored++; else stats.applied++;
+ return res.json({ event_id: event.event_id, duplicate: false, outcome: result.outcome, ...(result.followed_up && result.followed_up.length ? { task_ids: result.followed_up } : {}) });
+ }
+
if (!RUN_TOPICS.includes(event.event_type)) {
if (!RUN_TOPICS.includes(event.event_type)) {
const result = await ignore(event, 'ignored:type');
const result = await ignore(event, 'ignored:type');
if (result.duplicate) stats.duplicates++; else stats.ignored++;
if (result.duplicate) stats.duplicates++; else stats.ignored++;
@@ -145,7 +184,7 @@ function createEventsConsumer({ config, store, tasks, run, accountData, accountS
@@ -145,7 +184,7 @@ function createEventsConsumer({ config, store, tasks, run, accountData, accountS
return res.json({ event_id: event.event_id, duplicate: false, outcome: result.outcome, ...(result.task_id ? { task_id: result.task_id } : {}) });
return res.json({ event_id: event.event_id, duplicate: false, outcome: result.outcome, ...(result.task_id ? { task_id: result.task_id } : {}) });
});
});
- return { router, topics: TOPICS, runTopics: RUN_TOPICS, stats: () => ({ ...stats, enabled: secrets.length > 0 }) };
+ return { router, topics: TOPICS, runTopics: RUN_TOPICS, githubTopics: GITHUB_TOPICS, stats: () => ({ ...stats, enabled: secrets.length > 0 }) };
}
}
/** Create any missing subscription for `topics` at Events (idempotent; retried in the background at boot). */
/** Create any missing subscription for `topics` at Events (idempotent; retried in the background at boot). */
@@ -206,4 +245,4 @@ function startSubscriptions(opts = {}) {
@@ -206,4 +245,4 @@ function startSubscriptions(opts = {}) {
return { topics, endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };
return { topics, endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };
}
}
-module.exports = { createEventsConsumer, startSubscriptions, RUN_TOPICS, TOPICS, EVENT_ID_RE };
+module.exports = { createEventsConsumer, startSubscriptions, RUN_TOPICS, GITHUB_TOPICS, TOPICS, EVENT_ID_RE };
@@ -21,6 +21,11 @@ const TOPICS = Object.freeze([
@@ -21,6 +21,11 @@ const TOPICS = Object.freeze([
'codes.task.queued', 'codes.task.running', 'codes.task.succeeded', 'codes.task.failed', 'codes.task.cancelled',
'codes.task.queued', 'codes.task.running', 'codes.task.succeeded', 'codes.task.failed', 'codes.task.cancelled',
]);
]);
const TOPIC_SET = new Set(TOPICS);
const TOPIC_SET = new Set(TOPICS);
+/** A ticket follows its pull request (AL5): bound, follow-up started, ended. Payloads are plain JSON; the
+ * service manifest's eventsProduced stays the codes.task.* five (contracts.test.js), so these are the
+ * portal's view, not a new contract. */
+const PR_TOPICS = Object.freeze(['codes.pr.bound', 'codes.pr.followup', 'codes.pr.ended']);
+const PR_TOPIC_SET = new Set(PR_TOPICS);
const MAX_BACKOFF_MS = 5 * 60_000;
const MAX_BACKOFF_MS = 5 * 60_000;
// The envelope contract (events.event-envelope@1): the envelope Codes sends has to match it, or OpenVibe.Events
// The envelope contract (events.event-envelope@1): the envelope Codes sends has to match it, or OpenVibe.Events
// refuses the event with a 422 and the row is retried forever.
// refuses the event with a 422 and the row is retried forever.
@@ -160,7 +165,23 @@ function createOutbox({ store, log = console }) {
@@ -160,7 +165,23 @@ function createOutbox({ store, log = console }) {
return eventId;
return eventId;
}
}
- return { emitTask, emitAgentStep, payloadOf, topicFor, groupOf };
+ /**
+ * A pr-follow event in the same outbox (the publisher needs no topic filter: it sends every unpublished
+ * row). subject is the ticket owner's usr_…; the payload names the ticket, the repo, the PR and the fix.
+ */
+ async function emitPr(topic, subject, payload) {
+ if (!PR_TOPIC_SET.has(topic) || !subject || !payload || typeof payload !== 'object') return null;
+ const eventId = ids.newId('event', store.now());
+ const at = store.iso();
+ await db.exec(
+ `INSERT INTO codes_outbox (event_id, topic, subject, payload, created_at, next_attempt_at)
+ VALUES ($1, $2, $3, $4::jsonb, $5::timestamptz, $5::timestamptz)`,
+ [eventId, topic, subject, JSON.stringify(payload), at],
+ );
+ return eventId;
+ }
+
+ return { emitTask, emitAgentStep, emitPr, payloadOf, topicFor, groupOf, PR_TOPICS };
}
}
function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = console, intervalMs = (config.events && config.events.intervalMs) || 5000 }) {
function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = console, intervalMs = (config.events && config.events.intervalMs) || 5000 }) {
@@ -333,4 +354,4 @@ function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = co
@@ -333,4 +354,4 @@ function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = co
};
};
}
}
-module.exports = { createOutbox, createPublisher, TOPICS, TOPIC_SET, AGENT_TOPICS, agentStepOf, topicFor, groupOf, payloadOf, repoSlug, MAX_BACKOFF_MS, DEAD_AFTER_REFUSALS, DEAD_AFTER_ATTEMPTS };
+module.exports = { createOutbox, createPublisher, TOPICS, TOPIC_SET, PR_TOPICS, AGENT_TOPICS, agentStepOf, topicFor, groupOf, payloadOf, repoSlug, MAX_BACKOFF_MS, DEAD_AFTER_REFUSALS, DEAD_AFTER_ATTEMPTS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,285 @@
@@ -0,0 +1,285 @@
+'use strict';
+/**
+ * A ticket follows its pull request (AL5): a binding is stored when a task's result pushes a PR; a failed
+ * check, a conflict or a review comment on that PR wakes the same worker with the failing evidence (at most
+ * 3 automatic fixes per PR); a person's "stop" ends it; a bot's comment on an ended ticket does nothing.
+ *
+ * The github.* envelopes below are the shape OpenVibe.Events stores for the recorded GitHub payloads in its
+ * own test/fixtures/github (same fake repo and PR: OpenVibers/OpenVibe.Codes#999), delivered to /internal/events
+ * signed the way Events signs (openvibe-sdk/events, the CODES_EVENTS_SECRET fixture).
+ */
+const assert = require('assert');
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+const { ids } = require('openvibe-contracts');
+const { parseArgs } = require('../harness/job');
+const { parsePrUrl, MAX_FIXES_PER_PR, RED_CONCLUSIONS } = require('../server/domain/pr-bindings');
+const { boot, check, done } = require('./helpers/boot');
+
+const SECRET = `fixture_${'pr-follow'.repeat(4)}`;
+const NOW = Date.now();
+const REPO = 'https://github.com/OpenVibers/OpenVibe.Codes';
+const PR_URL = 'https://github.com/OpenVibers/OpenVibe.Codes/pull/999';
+const BRANCH = 'al5-scratch';
+const PROJECT = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';
+
+const terminalJob = () => ({
+ state: 'succeeded',
+ result: { ok: true, summary: 'did the thing', push: { pushed: true, branch: BRANCH, pull_request: PR_URL } },
+ timings: { started_at: new Date(NOW).toISOString(), finished_at: new Date(NOW).toISOString() },
+ error: null,
+});
+
+const ghEvent = (eventType, subjectId, payload) => ({
+ event_id: ids.newId('event'), event_type: eventType, version: 1, source: 'github',
+ actor: { type: 'service', id: 'github' }, timestamp: new Date(NOW).toISOString(),
+ subject: { type: 'pr', id: subjectId }, payload,
+});
+const SUBJECT = 'openvibers/openvibe.codes#999';
+// The real ingress fans a check out one envelope per PR: the number lives in the subject, not the payload.
+const redCheck = () => ghEvent('github.check.completed', SUBJECT, {
+ repo: 'openvibers/openvibe.codes', kind: 'check_run', conclusion: 'failure',
+ check_name: 'codes-ci / tests', head_sha: 'abc123',
+ html_url: 'https://github.com/OpenVibers/OpenVibe.Codes/actions/runs/111/job/222',
+ details_url: 'https://github.com/OpenVibers/OpenVibe.Codes/actions/runs/111',
+ output: {
+ title: '1 failing test', summary: 'test/pr-bindings.test.js failed on the branch',
+ text: 'not ok 3 - a red check wakes the worker once\n AssertionError: expected 0 to equal 1\n# fail 1\n',
+ },
+});
+const wrongPr = () => ghEvent('github.check.completed', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 1000, kind: 'check_run', conclusion: 'failure', check_name: 'x', head_sha: 'abc123', html_url: 'https://example.com' });
+const greenCheck = () => ghEvent('github.check.completed', SUBJECT, { repo: 'openvibers/openvibe.codes', kind: 'check_run', conclusion: 'success', check_name: 'codes-ci / tests', head_sha: 'abc123', html_url: 'https://github.com/OpenVibers/OpenVibe.Codes/actions/runs/112' });
+const conflict = () => ghEvent('github.pr.updated', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 999, action: 'synchronize', merged: false, mergeable_state: 'dirty', head_sha: 'abc123', base: 'main', html_url: PR_URL });
+const personComment = (body) => ghEvent('github.pr.commented', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 999, author: 'owner-person', author_type: 'User', state: 'commented', body, html_url: `${PR_URL}#issuecomment-1` });
+const botComment = () => ghEvent('github.pr.commented', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 999, author: 'github-actions[bot]', author_type: 'Bot', state: 'commented', body: 'All checks have passed.', html_url: `${PR_URL}#issuecomment-2` });
+const merged = () => ghEvent('github.pr.updated', SUBJECT, { repo: 'openvibers/openvibe.codes', pr_number: 999, action: 'closed', merged: true, mergeable_state: 'clean', head_sha: 'abc123', base: 'main', html_url: PR_URL });
+
+(async () => {
+ const submissions = [];
+ let nextJob = 1;
+ const run = {
+ enabled: true,
+ async submit({ subject, project, args }) {
+ submissions.push({ subject, project, args });
+ return { id: `job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J${nextJob++}`, state: 'queued' };
+ },
+ async read() { throw new Error('not used'); },
+ async cancel() { return { state: 'cancelled' }; },
+ async streamUrl() { throw new Error('not used'); },
+ };
+ const t = await boot({ run, env: { CODES_EVENTS_SECRET: SECRET, CODES_PR_FOLLOW_ENABLED: '1' } });
+ const db = t.ctx.store.db;
+ const ada = t.network.addUser('ada');
+ const args = parseArgs(JSON.stringify({ repo: REPO, prompt: 'Fix the typo on the page', permission: 'edit', push: { pull_request: true } }));
+
+ const deliver = async (event) => {
+ const body = JSON.stringify({ event, seq: 1 });
+ return await t.get('/internal/events', { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });
+ };
+ const bindingOf = (taskId) => db.maybe('SELECT * FROM pr_bindings WHERE task_id = $1', [taskId]);
+ const outbox = () => db.many('SELECT topic, subject, payload FROM codes_outbox ORDER BY created_at ASC');
+
+ try {
+ await check('a finished task that pushed a PR binds it (role fix, found on create)', async () => {
+ const task = await t.ctx.tasks.start({ subject: ada.subject, project: PROJECT, args });
+ const ended = await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [task.id]), { state: 'succeeded', job: terminalJob(), timings: terminalJob().timings });
+ assert.strictEqual(ended.state, 'succeeded');
+ const b = await bindingOf(task.id);
+ assert.ok(b, 'no binding stored');
+ assert.strictEqual(b.repo, 'openvibers/openvibe.codes');
+ assert.strictEqual(b.pr_number, 999);
+ assert.strictEqual(b.role, 'fix');
+ assert.strictEqual(b.how_found, 'created');
+ assert.strictEqual(b.branch, BRANCH);
+ assert.strictEqual(b.state, 'active');
+ assert.strictEqual(b.fix_count, 0);
+ const topics = (await outbox()).map((r) => r.topic);
+ assert.ok(topics.includes('codes.pr.bound'), `no codes.pr.bound in ${topics.join(',')}`);
+ });
+
+ let parent;
+ await check('a red check wakes the worker once, with the log tail and the same session, and no merge', async () => {
+ parent = await db.maybe('SELECT * FROM run_tasks ORDER BY created_at DESC LIMIT 1');
+ // The worker's last attempt recorded its harness session; the follow-up continues that conversation.
+ await db.exec("UPDATE codes_attempts SET session = 'sess_pr_follow_1', agent = 'claude-code' WHERE task_id = $1", [parent.id]);
+ const before = submissions.length;
+ const r = await deliver(redCheck());
+ assert.strictEqual(r.status, 200, r.text);
+ assert.strictEqual(r.json().outcome, 'woke', r.text);
+ assert.strictEqual(submissions.length, before + 1, 'the red check must start exactly one follow-up');
+ const job = submissions[submissions.length - 1];
+ assert.strictEqual(job.args.repo, REPO);
+ assert.strictEqual(job.args.ref, BRANCH);
+ assert.strictEqual(job.args.push.branch, BRANCH);
+ assert.strictEqual(job.args.push.pull_request, false);
+ assert.strictEqual(job.args.push.update, true);
+ assert.strictEqual(job.args.agent, 'claude-code');
+ assert.strictEqual(job.args.resume, 'sess_pr_follow_1');
+ assert.match(job.args.prompt, /Automatic follow-up 1 of 3/);
+ assert.match(job.args.prompt, /concluded "failure"/);
+ assert.match(job.args.prompt, /not ok 3 - a red check wakes the worker once/);
+ assert.match(job.args.prompt, /\/actions\/runs\/111/);
+ assert.match(job.args.prompt, /never merge/);
+ assert.ok(!/merge( |ing|d) /.test(job.args.prompt.replace('never merge', '')), 'the follow-up must not be told to merge');
+ const b = await db.maybe('SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);
+ assert.strictEqual(b.fix_count, 1);
+ assert.notStrictEqual(b.task_id, parent.id, 'the binding follows the newest task');
+ const child = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [b.task_id]);
+ assert.ok(child && child.state === 'queued');
+ const topics = (await outbox()).map((r) => r.topic);
+ assert.ok(topics.includes('codes.pr.followup'), `no codes.pr.followup in ${topics.join(',')}`);
+ });
+
+ await check('the fourth red check does not wake (at most 3 automatic fixes per PR)', async () => {
+ // Fixes 2 and 3: end each follow-up first so the fix_running guard lets the next one through.
+ for (const n of [2, 3]) {
+ const b = await db.maybe('SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);
+ await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [b.task_id]), { state: 'succeeded', job: terminalJob(), timings: terminalJob().timings });
+ const before = submissions.length;
+ const r = await deliver(redCheck());
+ assert.strictEqual(r.json().outcome, 'woke', r.text);
+ assert.strictEqual(submissions.length, before + 1);
+ const last = submissions[submissions.length - 1];
+ assert.match(last.args.prompt, new RegExp(`Automatic follow-up ${n} of 3`));
+ }
+ const b = await db.maybe('SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);
+ await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [b.task_id]), { state: 'succeeded', job: terminalJob(), timings: terminalJob().timings });
+ const before = submissions.length;
+ const r = await deliver(redCheck());
+ assert.strictEqual(r.json().outcome, 'ignored:max_fixes', r.text);
+ assert.strictEqual(submissions.length, before, 'the fourth red must not start a follow-up');
+
+ // But a person's comment is a new instruction: it may start a manual fix past the automatic cap.
+ const manual = await deliver(personComment('please fix the last failure by hand'));
+ assert.strictEqual(manual.json().outcome, 'woke', manual.text);
+ assert.strictEqual(submissions.length, before + 1);
+ assert.match(submissions[submissions.length - 1].args.prompt, /Follow-up requested by a person/);
+ const after = await db.maybe('SELECT fix_count, task_id FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);
+ assert.strictEqual(after.fix_count, 3, 'a manual fix does not count against the automatic cap');
+ });
+
+ await check('a green check changes nothing; a payload naming another PR is not ours', async () => {
+ const before = submissions.length;
+ const r = await deliver(greenCheck());
+ assert.strictEqual(r.json().outcome, 'ignored:green', r.text);
+ assert.strictEqual(submissions.length, before);
+ const w = await deliver(wrongPr());
+ assert.strictEqual(w.json().outcome, 'ignored:payload', w.text);
+ assert.strictEqual(submissions.length, before);
+ });
+
+ await check('a conflict wakes the fix binding with the evidence', async () => {
+ // A fresh ticket on another PR, so the fix cap of #999 does not apply.
+ const other = await t.ctx.tasks.start({ subject: ada.subject, project: PROJECT, args: parseArgs(JSON.stringify({ repo: REPO, prompt: 'Another typo', permission: 'edit', push: { pull_request: true } })) });
+ await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [other.id]), {
+ state: 'succeeded',
+ job: { ...terminalJob(), result: { ...terminalJob().result, push: { pushed: true, branch: 'al5-other', pull_request: 'https://github.com/OpenVibers/OpenVibe.Codes/pull/1000' } } },
+ timings: terminalJob().timings,
+ });
+ const dirty = ghEvent('github.pr.updated', 'openvibers/openvibe.codes#1000', { repo: 'openvibers/openvibe.codes', pr_number: 1000, action: 'synchronize', merged: false, mergeable_state: 'dirty', head_sha: 'def456', base: 'main', html_url: 'https://github.com/OpenVibers/OpenVibe.Codes/pull/1000' });
+ const before = submissions.length;
+ const r = await deliver(dirty);
+ assert.strictEqual(r.json().outcome, 'woke', r.text);
+ assert.strictEqual(submissions.length, before + 1);
+ assert.match(submissions[submissions.length - 1].args.prompt, /mergeable_state "dirty"/);
+ });
+
+ await check('a merged PR ends the binding', async () => {
+ const r = await deliver(merged());
+ assert.strictEqual(r.json().outcome, 'ended:merged', r.text);
+ const b = await db.maybe('SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 ORDER BY created_at ASC', ['openvibers/openvibe.codes', 999]);
+ assert.strictEqual(b.state, 'ended');
+ assert.strictEqual(b.ended_reason, 'merged');
+ const topics = (await outbox()).map((r) => r.topic);
+ assert.ok(topics.includes('codes.pr.ended'), `no codes.pr.ended in ${topics.join(',')}`);
+ });
+
+ await check('a bot comment on an ended ticket does nothing', async () => {
+ const before = submissions.length;
+ const r = await deliver(botComment());
+ assert.strictEqual(r.json().outcome, 'ignored:ended', r.text);
+ assert.strictEqual(submissions.length, before);
+ });
+
+ await check('a person\'s "stop" ends the binding; a person\'s comment re-opens and wakes it', async () => {
+ // Re-openable ticket on PR #1000 (one fix used, cap not reached).
+ const stop1000 = personComment('stop — for now, I will finish this myself');
+ stop1000.subject = { type: 'pr', id: 'openvibers/openvibe.codes#1000' };
+ stop1000.payload = { ...stop1000.payload, pr_number: 1000 };
+ const wake1000 = personComment('actually the test still fails, please fix it');
+ wake1000.subject = { type: 'pr', id: 'openvibers/openvibe.codes#1000' };
+ wake1000.payload = { ...wake1000.payload, pr_number: 1000 };
+ // End #1000's follow-up with a plain result (no PR: it binds nothing new) so the next wake may start.
+ {
+ const b = await db.maybe("SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 AND state = 'active'", ['openvibers/openvibe.codes', 1000]);
+ await t.ctx.tasks.applyEvent(await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [b.task_id]), {
+ state: 'succeeded',
+ job: { state: 'succeeded', result: { ok: true, summary: 'fixed the conflict' }, timings: terminalJob().timings, error: null },
+ timings: terminalJob().timings,
+ });
+ }
+ let r = await deliver(stop1000);
+ assert.strictEqual(r.json().outcome, 'stopped', r.text);
+ let b = await db.maybe("SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 AND state = 'active'", ['openvibers/openvibe.codes', 1000]);
+ assert.strictEqual(b, null, 'stop must end the active binding');
+ const before = submissions.length;
+ r = await deliver(wake1000);
+ assert.strictEqual(r.json().outcome, 'reopened', r.text);
+ assert.strictEqual(submissions.length, before + 1, 'a person can wake an ended ticket');
+ b = await db.maybe("SELECT * FROM pr_bindings WHERE repo = $1 AND pr_number = $2 AND state = 'active'", ['openvibers/openvibe.codes', 1000]);
+ assert.ok(b, 'the comment re-opens the binding');
+ assert.match(submissions[submissions.length - 1].args.prompt, /actually the test still fails/);
+ });
+
+ await check('with the switch off, a red check is recorded and no follow-up starts', async () => {
+ const t2 = await boot({ run, env: { CODES_EVENTS_SECRET: SECRET } });
+ try {
+ const db2 = t2.ctx.store.db;
+ const ada2 = t2.network.addUser('bob');
+ const task = await t2.ctx.tasks.start({ subject: ada2.subject, project: PROJECT, args });
+ await t2.ctx.tasks.applyEvent(await db2.maybe('SELECT * FROM run_tasks WHERE id = $1', [task.id]), { state: 'succeeded', job: terminalJob(), timings: terminalJob().timings });
+ assert.ok(await db2.maybe('SELECT id FROM pr_bindings WHERE task_id = $1', [task.id]), 'binding is stored even with the switch off');
+ const before = submissions.length;
+ const body = JSON.stringify({ event: redCheck(), seq: 1 });
+ const res = await t2.get('/internal/events', { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });
+ assert.strictEqual(res.json().outcome, 'ignored:follow_disabled', res.text);
+ assert.strictEqual(submissions.length, before);
+ } finally { await t2.close(); }
+ });
+
+ await check('parsePrUrl and the update guards', async () => {
+ assert.deepStrictEqual(parsePrUrl(PR_URL), { repo: 'openvibers/openvibe.codes', number: 999 });
+ assert.strictEqual(parsePrUrl('https://github.com/o/n/issues/1'), null);
+ assert.strictEqual(parsePrUrl('not a url'), null);
+ assert.deepStrictEqual([...RED_CONCLUSIONS].sort(), ['action_required', 'failure', 'timed_out']);
+ assert.strictEqual(MAX_FIXES_PER_PR, 3);
+ const { STOP_RE } = require('../server/domain/pr-bindings');
+ assert.ok(STOP_RE.test('stop'));
+ assert.ok(STOP_RE.test('Stop.'));
+ assert.ok(STOP_RE.test('stop — I will take this one from here'));
+ assert.ok(!STOP_RE.test('stop the flaky retry loop, it is noisy'));
+ assert.throws(() => parseArgs(JSON.stringify({ repo: REPO, prompt: 'x', push: { branch: 'b', ref: undefined, pull_request: true, update: true } })), /never opens a new pull request/);
+ assert.throws(() => parseArgs(JSON.stringify({ repo: REPO, ref: 'main', prompt: 'x', push: { branch: 'other', update: true } })), /branch === ref/);
+ const ok = parseArgs(JSON.stringify({ repo: REPO, ref: 'b', prompt: 'x', push: { branch: 'b', update: true } }));
+ assert.strictEqual(ok.push.update, true);
+ assert.strictEqual(ok.push.pull_request, false);
+ });
+
+ await check('the worker never merges its own PR', async () => {
+ const { pushTarget } = require('../harness/job');
+ // An update push goes only to the branch it cloned; anything else is refused.
+ assert.deepStrictEqual(pushTarget('b', 'b', { update: true, exists: true }), { branch: 'b' });
+ assert.ok(pushTarget('b', 'b', { update: true, exists: false }).refused, 'a follow-up refuses a vanished branch');
+ assert.deepStrictEqual(pushTarget('b', 'main', { exists: false }), { branch: 'b' }, 'a new branch still pushes as before');
+ assert.ok(pushTarget('b', 'b', { exists: false }).refused, 'without update the old refusal stands');
+ // The harness has no merge step: no git merge, no merge API call anywhere in the job path.
+ const fs = require('fs');
+ const jobSrc = fs.readFileSync(require('path').join(__dirname, '..', 'harness', 'job.js'), 'utf8');
+ assert.ok(!/git\(\s*\['merge'/.test(jobSrc), 'no git merge in harness/job.js');
+ assert.ok(!/\/merges?\b/.test(jobSrc.replace(/mergeable_state/g, '')), 'no merge API in harness/job.js');
+ });
+
+ } finally {
+ await t.close();
+ }
+ done();
+})();