Blame · Line-by-line history

`autopilot.ts`

Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.

autopilot.tsBlame1158 lines · 3 contributors

Normal view Raw

2b821b7Claude	1	/**
	2	* Autopilot — self-sufficiency loop.
	3	*
	4	* Runs existing platform-maintenance tasks (mirror sync, merge queue progress,
	5	* weekly digests, advisory rescans) on an interval so the host runs itself
	6	* without an external cron. All sub-tasks are injected so tests can stub them
	7	* without touching the DB; the default task set wires real helpers from the
	8	* locked libs. Nothing here throws — every sub-task and the outer tick are
	9	* try/caught so a single failure never blocks the others.
	10	*/
	11
2b9055eClaude	12	import { and, eq, gte, sql } from "drizzle-orm";
2b821b7Claude	13	import { db } from "../db";
2b9055eClaude	14	import {
	15	mergeQueueEntries,
	16	prComments,
	17	pullRequests,
	18	repoDependencies,
	19	repositories,
	20	users,
	21	} from "../db/schema";
2b821b7Claude	22	import { syncAllDue } from "./mirrors";
	23	import { peekHead } from "./merge-queue";
	24	import { sendDigestsToAll } from "./email-digest";
	25	import { scanRepositoryForAlerts } from "./advisories";
a76d984Claude	26	import { releaseExpiredWaitTimers } from "./environments";
665c8bfClaude	27	import { runScheduledWorkflowsTick } from "./scheduled-workflows";
2b9055eClaude	28	import {
	29	evaluateAutoMerge,
	30	recordAutoMergeAttempt,
	31	type AutoMergeContext,
	32	type AutoMergeDecision,
	33	} from "./auto-merge";
	34	import { matchProtection } from "./branch-protection";
	35	import { performMerge, type PerformMergeResult } from "./pr-merge";
	36	import { audit } from "./notify";
	37	import { runAiBuildTaskOnce } from "./ai-build-tasks";
46d6165Claude	38	import {
	39	sendSleepModeDigestForUser,
	40	SLEEP_MODE_USER_CAP_PER_TICK,
	41	SLEEP_MODE_COOLDOWN_HOURS,
	42	} from "./sleep-mode";
534f04aClaude	43	import {
	44	runStalePrSweepOnce,
	45	runStaleIssueSweepOnce,
	46	} from "./stale-sweep";
	47	import { computePrRiskForPullRequest } from "./pr-risk";
	48	import { prRiskScores } from "../db/schema";
c63b860Claude	49	import { purgeScheduledAccounts } from "./account-deletion";
cd4f63bTest User	50	import { purgeExpiredPlaygroundAccounts } from "./playground";
b1be050CC LABS App	51	import {
	52	runSyntheticChecks,
	53	persistChecks,
	54	latestStatusByCheck,
	55	type SyntheticCheckResult,
	56	} from "./synthetic-monitor";
2b821b7Claude	57
	58	export interface AutopilotTaskResult {
	59	name: string;
	60	ok: boolean;
	61	durationMs: number;
	62	error?: string;
	63	}
	64
	65	export interface AutopilotTickResult {
	66	startedAt: string;
	67	finishedAt: string;
	68	tasks: AutopilotTaskResult[];
	69	}
	70
	71	export interface AutopilotTask {
	72	name: string;
	73	run: () => Promise<void>;
	74	}
	75
	76	export interface StartAutopilotOpts {
	77	intervalMs?: number;
	78	now?: () => number;
	79	tasks?: AutopilotTask[];
	80	}
	81
	82	export interface RunTickOpts {
	83	tasks?: AutopilotTask[];
	84	now?: () => number;
	85	}
	86
	87	const DEFAULT_INTERVAL_MS = 5 * 60 * 1000;
	88	const ADVISORY_RESCAN_BATCH = 5;
2b9055eClaude	89	/** K3 — recency window for auto-merge candidate selection. */
	90	const AUTO_MERGE_LOOKBACK_HOURS = 24;
	91	/** K3 — hard cap on PRs evaluated per tick (runaway protection). */
	92	const AUTO_MERGE_MAX_PER_TICK = 50;
	93	/** K3 — stable marker for the auto-merge audit comment. */
	94	const AUTO_MERGE_COMMENT_MARKER = "<!-- gluecron:auto-merge:v1 -->";
534f04aClaude	95	/** M3 — hard cap on PRs scored per tick (runaway protection). */
	96	const PR_RISK_RESCORE_MAX_PER_TICK = 20;
	97	/** M3 — recency window for the pr-risk-rescore sweep. */
	98	const PR_RISK_RESCORE_LOOKBACK_HOURS = 1;
2b821b7Claude	99
	100	/**
	101	* Default task set. Each task is a thin wrapper around an existing locked
	102	* helper — no gate/merge logic is duplicated here.
	103	*/
	104	export function defaultTasks(): AutopilotTask[] {
	105	return [
	106	{
	107	name: "mirror-sync",
	108	run: async () => {
	109	await syncAllDue();
	110	},
	111	},
	112	{
	113	name: "merge-queue",
	114	run: async () => {
	115	await processMergeQueues();
	116	},
	117	},
	118	{
	119	name: "weekly-digest",
	120	run: async () => {
	121	await sendDigestsToAll();
	122	},
	123	},
	124	{
	125	name: "advisory-rescan",
	126	run: async () => {
	127	await rescanAdvisoriesBatch(ADVISORY_RESCAN_BATCH);
	128	},
	129	},
a76d984Claude	130	{
	131	name: "wait-timer-release",
	132	run: async () => {
	133	await releaseExpiredWaitTimers();
	134	},
	135	},
665c8bfClaude	136	{
	137	name: "scheduled-workflows",
	138	run: async () => {
	139	await runScheduledWorkflowsTick();
	140	},
	141	},
2b9055eClaude	142	{
	143	name: "auto-merge-sweep",
	144	run: async () => {
	145	await runAutoMergeSweep();
	146	},
	147	},
	148	{
	149	name: "ai-build-from-issues",
	150	run: async () => {
	151	const summary = await runAiBuildTaskOnce();
	152	console.log(
	153	`[autopilot] ai-build: queued=${summary.queued} skipped=${summary.skipped}`
	154	);
	155	},
	156	},
46d6165Claude	157	{
	158	name: "sleep-mode-digest",
	159	run: async () => {
	160	const summary = await runSleepModeDigestTaskOnce();
	161	console.log(
	162	`[autopilot] sleep-mode-digest: sent=${summary.sent} skipped=${summary.skipped}`
	163	);
	164	},
	165	},
534f04aClaude	166	{
	167	name: "stale-pr-sweep",
	168	run: async () => {
	169	// Two-stage gate: poke at 7d stale, close at 14d after poke
	170	// (when the repo opts in via `auto_close_stale_prs`).
	171	// Wrapped in try/catch so a finder crash never wedges the tick.
	172	try {
	173	const summary = await runStalePrSweepOnce();
	174	console.log(
	175	`[autopilot] stale-pr-sweep: poked=${summary.poked} closed=${summary.closed}`
	176	);
	177	} catch (err) {
	178	console.error("[autopilot] stale-pr-sweep: threw:", err);
	179	}
	180	},
	181	},
	182	{
	183	name: "stale-issue-sweep",
	184	run: async () => {
	185	// Mirror of stale-pr-sweep with the issue thresholds (30d/60d).
	186	try {
	187	const summary = await runStaleIssueSweepOnce();
	188	console.log(
	189	`[autopilot] stale-issue-sweep: poked=${summary.poked} closed=${summary.closed}`
	190	);
	191	} catch (err) {
	192	console.error("[autopilot] stale-issue-sweep: threw:", err);
	193	}
	194	},
	195	},
	196	{
	197	name: "pr-risk-rescore",
	198	run: async () => {
	199	const summary = await runPrRiskRescoreTaskOnce();
	200	console.log(
	201	`[autopilot] pr-risk-rescore: scored=${summary.scored} skipped=${summary.skipped}`
	202	);
	203	},
	204	},
c63b860Claude	205
	206	{
	207	// Block P5 — Hard-delete users whose 30-day grace period expired.
	208	name: "account-purge",
	209	run: async () => {
	210	try {
	211	const summary = await purgeScheduledAccounts({ cap: 50 });
	212	console.log(
	213	`[autopilot] account-purge: purged=${summary.purged} errors=${summary.errors}`
	214	);
	215	} catch (err) {
	216	console.error("[autopilot] account-purge: threw:", err);
	217	}
	218	},
	219	},
cd4f63bTest User	220	{
	221	// Block Q3 — Hard-delete anonymous playground accounts past their
	222	// 24h TTL. CASCADE handles repos, sessions, issues. Per-user
	223	// try/catch in the lib so one FK violation can't stall the queue.
	224	name: "playground-purge",
	225	run: async () => {
	226	try {
	227	const summary = await purgeExpiredPlaygroundAccounts({ cap: 50 });
	228	console.log(
	229	`[autopilot] playground-purge: purged=${summary.purged} errors=${summary.errors}`
	230	);
	231	} catch (err) {
	232	console.error("[autopilot] playground-purge: threw:", err);
	233	}
	234	},
	235	},
b1be050CC LABS App	236	{
	237	// BLOCK S4 — Synthetic monitor.
	238	//
	239	// Runs the URL-only smoke suite (see src/lib/synthetic-monitor.ts),
	240	// records the outcome into `synthetic_checks`, and on a
	241	// green->red transition fires a webhook to MONITOR_ALERT_WEBHOOK_URL
	242	// (when configured) so the owner finds out instantly that the live
	243	// site is broken. Wrapped in try/catch — the monitor must never
	244	// wedge the tick.
	245	name: "synthetic-monitor",
	246	run: async () => {
	247	try {
	248	const summary = await runSyntheticMonitorTaskOnce();
	249	console.log(
	250	`[autopilot] synthetic-monitor: green=${summary.green} red=${summary.red} transitions=${summary.transitions}`
	251	);
	252	} catch (err) {
	253	console.error("[autopilot] synthetic-monitor: threw:", err);
	254	}
	255	},
	256	},
2b821b7Claude	257	];
	258	}
	259
b1be050CC LABS App	260	// ---------------------------------------------------------------------------
	261	// BLOCK S4 — synthetic-monitor task
	262	// ---------------------------------------------------------------------------
	263
	264	export interface SyntheticMonitorTaskDeps {
	265	/** Override the suite runner (DI for tests). */
	266	runChecks?: () => Promise<SyntheticCheckResult[]>;
	267	/** Override the persistence step (DI for tests). */
	268	persist?: (results: SyntheticCheckResult[]) => Promise<void>;
	269	/** Override the previous-state loader (DI for tests). */
	270	loadPrevious?: () => Promise<Record<string, SyntheticCheckResult>>;
	271	/** Override the webhook poster (DI for tests). */
	272	postAlert?: (url: string, payload: unknown) => Promise<void>;
	273	/** Override the alert-webhook URL lookup (defaults to env). */
	274	alertUrl?: () => string;
	275	}
	276
	277	export interface SyntheticMonitorTaskSummary {
	278	green: number;
	279	red: number;
	280	yellow: number;
	281	transitions: number;
	282	}
	283
	284	async function defaultPostAlert(
	285	url: string,
	286	payload: unknown
	287	): Promise<void> {
	288	try {
	289	await fetch(url, {
	290	method: "POST",
	291	headers: { "Content-Type": "application/json" },
	292	body: JSON.stringify(payload),
	293	});
	294	} catch (err) {
	295	console.error("[autopilot] synthetic-monitor: alert webhook failed:", err);
	296	}
	297	}
	298
	299	/**
	300	* One iteration of the synthetic-monitor task. Runs the checks, persists
	301	* them, compares against the prior state, and fires a webhook on each
	302	* green->red transition (red->red repeats stay quiet so we don't spam
	303	* the channel). Never throws.
	304	*/
	305	export async function runSyntheticMonitorTaskOnce(
	306	deps: SyntheticMonitorTaskDeps = {}
	307	): Promise<SyntheticMonitorTaskSummary> {
	308	const runChecks = deps.runChecks ?? (() => runSyntheticChecks());
	309	const persist = deps.persist ?? persistChecks;
	310	const loadPrevious =
	311	deps.loadPrevious ??
	312	(async () => {
	313	const latest = await latestStatusByCheck();
	314	// Strip the `checkedAt` from the result shape so the diff loop
	315	// compares the canonical SyntheticCheckResult fields only.
	316	const out: Record<string, SyntheticCheckResult> = {};
	317	for (const [k, v] of Object.entries(latest)) {
	318	const { checkedAt: _unused, ...rest } = v;
	319	void _unused;
	320	out[k] = rest;
	321	}
	322	return out;
	323	});
	324	const postAlert = deps.postAlert ?? defaultPostAlert;
	325	const alertUrl =
	326	deps.alertUrl ?? (() => process.env.MONITOR_ALERT_WEBHOOK_URL \|\| "");
	327
	328	let previous: Record<string, SyntheticCheckResult> = {};
	329	try {
	330	previous = await loadPrevious();
	331	} catch (err) {
	332	console.error(
	333	"[autopilot] synthetic-monitor: loadPrevious threw:",
	334	err
	335	);
	336	previous = {};
	337	}
	338
	339	const results = await runChecks();
	340	await persist(results);
	341
	342	let green = 0;
	343	let red = 0;
	344	let yellow = 0;
	345	let transitions = 0;
	346	const url = alertUrl();
	347
	348	for (const r of results) {
	349	if (r.status === "green") green += 1;
	350	else if (r.status === "red") red += 1;
	351	else yellow += 1;
	352
	353	const prior = previous[r.name];
	354	// green->red transition: prior was green (or absent and current is red
	355	// after a green is also a transition — but absent-before is treated as
	356	// green to avoid spamming on a fresh DB). We only alert on the
	357	// green->red edge so red->red doesn't re-fire.
	358	const priorWasGreen = !prior \|\| prior.status === "green";
	359	if (priorWasGreen && r.status === "red") {
	360	transitions += 1;
	361	if (url) {
	362	await postAlert(url, {
	363	check: r.name,
	364	status: r.status,
	365	statusCode: r.statusCode ?? null,
	366	durationMs: r.durationMs,
	367	error: r.error ?? null,
	368	checkedAt: new Date().toISOString(),
	369	});
	370	}
	371	}
	372	}
	373
	374	return { green, red, yellow, transitions };
	375	}
	376
46d6165Claude	377	// ---------------------------------------------------------------------------
	378	// L1 — sleep-mode-digest
	379	// ---------------------------------------------------------------------------
	380
	381	export interface SleepModeDigestCandidate {
	382	userId: string;
	383	digestHourUtc: number;
	384	lastDigestSentAt: Date \| null;
	385	}
	386
	387	export interface SleepModeDigestTaskDeps {
	388	/** Override the candidate finder. */
	389	findCandidates?: (cap: number) => Promise<SleepModeDigestCandidate[]>;
	390	/** Override the send-one-user helper (DI for tests). */
	391	sendOne?: (userId: string) => Promise<{ ok: boolean; reason?: string }>;
	392	/** Override the wall clock (DI for tests). */
	393	now?: () => Date;
	394	/** Override the per-tick cap. */
	395	cap?: number;
	396	/** Override the cooldown hours. */
	397	cooldownHours?: number;
	398	}
	399
	400	export interface SleepModeDigestTaskSummary {
	401	sent: number;
	402	skipped: number;
	403	}
	404
	405	/**
	406	* Default candidate-finder. Returns enabled users whose
	407	* `lastDigestSentAt` is older than the cooldown OR null. The hour-match
	408	* filter is applied in JS by `runSleepModeDigestTaskOnce` so it stays
	409	* timezone-independent of any SQL `extract(hour ...)` behaviour.
	410	*/
	411	async function defaultFindSleepModeCandidates(
	412	cap: number
	413	): Promise<SleepModeDigestCandidate[]> {
	414	try {
	415	const rows = await db
	416	.select({
	417	userId: users.id,
	418	digestHourUtc: users.sleepModeDigestHourUtc,
	419	lastDigestSentAt: users.lastDigestSentAt,
	420	})
	421	.from(users)
	422	.where(eq(users.sleepModeEnabled, true))
	423	.limit(cap);
	424	return rows.map((r) => ({
	425	userId: r.userId,
	426	digestHourUtc: r.digestHourUtc,
	427	lastDigestSentAt: r.lastDigestSentAt,
	428	}));
	429	} catch (err) {
	430	console.error("[autopilot] sleep-mode-digest: candidate query failed:", err);
	431	return [];
	432	}
	433	}
	434
	435	/**
	436	* One iteration of the sleep-mode-digest task. Never throws.
	437	*
	438	* Per-user filters (applied in JS so we can DI a clock):
	439	* 1. `lastDigestSentAt` is null OR older than cooldown (23h).
	440	* 2. `now.getUTCHours() === digestHourUtc` — fires once at the user's
	441	* configured local UTC hour.
	442	*
	443	* Caps at `SLEEP_MODE_USER_CAP_PER_TICK` (100) users per tick.
	444	*/
	445	export async function runSleepModeDigestTaskOnce(
	446	deps: SleepModeDigestTaskDeps = {}
	447	): Promise<SleepModeDigestTaskSummary> {
	448	const findCandidates =
	449	deps.findCandidates ?? defaultFindSleepModeCandidates;
	450	const sendOne = deps.sendOne ?? sendSleepModeDigestForUser;
	451	const now = deps.now ?? (() => new Date());
	452	const cap = deps.cap ?? SLEEP_MODE_USER_CAP_PER_TICK;
	453	const cooldownHours = deps.cooldownHours ?? SLEEP_MODE_COOLDOWN_HOURS;
	454
	455	let candidates: SleepModeDigestCandidate[] = [];
	456	try {
	457	candidates = await findCandidates(cap);
	458	} catch (err) {
	459	console.error("[autopilot] sleep-mode-digest: findCandidates threw:", err);
	460	return { sent: 0, skipped: 0 };
	461	}
	462
	463	const nowDate = now();
	464	const currentHour = nowDate.getUTCHours();
	465	const cooldownMs = cooldownHours * 60 * 60 * 1000;
	466
	467	let sent = 0;
	468	let skipped = 0;
	469
	470	for (const cand of candidates) {
	471	try {
	472	// Hour-match: must equal the user's configured UTC delivery hour.
	473	if (cand.digestHourUtc !== currentHour) {
	474	skipped += 1;
	475	continue;
	476	}
	477	// Cooldown: skip if we sent within the last cooldown window.
	478	if (
	479	cand.lastDigestSentAt &&
	480	nowDate.getTime() - new Date(cand.lastDigestSentAt).getTime() <
	481	cooldownMs
	482	) {
	483	skipped += 1;
	484	continue;
	485	}
	486	const result = await sendOne(cand.userId);
	487	if (result.ok) sent += 1;
	488	else skipped += 1;
	489	} catch (err) {
	490	skipped += 1;
	491	console.error(
	492	`[autopilot] sleep-mode-digest: per-user failure for user=${cand.userId}:`,
	493	err
	494	);
	495	}
	496	}
	497
	498	return { sent, skipped };
	499	}
	500
534f04aClaude	501	// ---------------------------------------------------------------------------
	502	// M3 — pr-risk-rescore
	503	// ---------------------------------------------------------------------------
	504
	505	export interface PrRiskRescoreCandidate {
	506	pullRequestId: string;
	507	headBranch: string;
	508	updatedAt: Date;
	509	}
	510
	511	export interface PrRiskRescoreTaskDeps {
	512	/** Override candidate finder for tests. */
	513	findCandidates?: (
	514	lookbackHours: number,
	515	cap: number
	516	) => Promise<PrRiskRescoreCandidate[]>;
	517	/** Override score computation for tests. */
	518	scoreOne?: (prId: string) => Promise<{ ok: boolean }>;
	519	/** Override per-tick cap. */
	520	cap?: number;
	521	/** Override lookback. */
	522	lookbackHours?: number;
	523	}
	524
	525	export interface PrRiskRescoreTaskSummary {
	526	scored: number;
	527	skipped: number;
	528	}
	529
	530	/**
	531	* Default candidate-finder. Returns open, non-draft PRs from non-archived
	532	* repos whose `updated_at` falls inside the lookback window. The "scored
	533	* at all" filter is applied as a second pass via `defaultFilterNeedsScoring`.
	534	*/
	535	async function defaultFindPrRiskCandidates(
	536	lookbackHours: number,
	537	cap: number
	538	): Promise<PrRiskRescoreCandidate[]> {
	539	const cutoff = new Date(Date.now() - lookbackHours * 60 * 60 * 1000);
	540	try {
	541	const rows = await db
	542	.select({
	543	pullRequestId: pullRequests.id,
	544	headBranch: pullRequests.headBranch,
	545	updatedAt: pullRequests.updatedAt,
	546	})
	547	.from(pullRequests)
	548	.innerJoin(
	549	repositories,
	550	eq(repositories.id, pullRequests.repositoryId)
	551	)
	552	.where(
	553	and(
	554	eq(pullRequests.state, "open"),
	555	eq(pullRequests.isDraft, false),
	556	eq(repositories.isArchived, false),
	557	gte(pullRequests.updatedAt, cutoff)
	558	)
	559	)
	560	.orderBy(sql`${pullRequests.updatedAt} DESC`)
	561	.limit(cap);
	562	return rows.map((r) => ({
	563	pullRequestId: r.pullRequestId,
	564	headBranch: r.headBranch,
	565	updatedAt: r.updatedAt,
	566	}));
	567	} catch (err) {
	568	console.error("[autopilot] pr-risk-rescore: candidate query failed:", err);
	569	return [];
	570	}
	571	}
	572
	573	/**
	574	* Drop candidates that already have ANY cached score row. The unique
	575	* constraint on (pull_request_id, commit_sha) handles the "score-the-
	576	* same-SHA-twice" case at persist time; this filter just keeps the work
	577	* list small enough to fit under the per-tick cap when many PRs are
	578	* being pushed concurrently.
	579	*/
	580	async function defaultFilterNeedsScoring(
	581	candidates: PrRiskRescoreCandidate[]
	582	): Promise<PrRiskRescoreCandidate[]> {
	583	if (candidates.length === 0) return [];
	584	try {
	585	const rows = await db
	586	.select({ pullRequestId: prRiskScores.pullRequestId })
	587	.from(prRiskScores);
	588	const scoredIds = new Set(rows.map((r) => r.pullRequestId));
	589	return candidates.filter((c) => !scoredIds.has(c.pullRequestId));
	590	} catch (err) {
	591	console.error("[autopilot] pr-risk-rescore: filter query failed:", err);
	592	// Fail-open: better to score everything than silently skip.
	593	return candidates;
	594	}
	595	}
	596
	597	/**
	598	* One iteration of the pr-risk-rescore task. Never throws. Compute risk
	599	* for up to `cap` recently-touched open PRs that have no cached score
	600	* yet, so reviewers usually see a populated card on first visit.
	601	*/
	602	export async function runPrRiskRescoreTaskOnce(
	603	deps: PrRiskRescoreTaskDeps = {}
	604	): Promise<PrRiskRescoreTaskSummary> {
	605	const findCandidates = deps.findCandidates ?? defaultFindPrRiskCandidates;
	606	const scoreOne =
	607	deps.scoreOne ??
	608	(async (prId: string) => {
	609	try {
	610	const result = await computePrRiskForPullRequest(prId);
	611	return { ok: result !== null };
	612	} catch {
	613	return { ok: false };
	614	}
	615	});
	616	const cap = deps.cap ?? PR_RISK_RESCORE_MAX_PER_TICK;
	617	const lookbackHours =
	618	deps.lookbackHours ?? PR_RISK_RESCORE_LOOKBACK_HOURS;
	619
	620	let candidates: PrRiskRescoreCandidate[] = [];
	621	try {
	622	candidates = await findCandidates(lookbackHours, cap);
	623	} catch (err) {
	624	console.error("[autopilot] pr-risk-rescore: findCandidates threw:", err);
	625	return { scored: 0, skipped: 0 };
	626	}
	627
	628	// Only score PRs missing a cached row. Skip filter when the caller
	629	// injected a custom finder (tests pass already-filtered lists).
	630	const needsScoring =
	631	deps.findCandidates === undefined
	632	? await defaultFilterNeedsScoring(candidates)
	633	: candidates;
	634
	635	let scored = 0;
	636	let skipped = 0;
	637	for (const cand of needsScoring.slice(0, cap)) {
	638	try {
	639	const result = await scoreOne(cand.pullRequestId);
	640	if (result.ok) scored += 1;
	641	else skipped += 1;
	642	} catch (err) {
	643	skipped += 1;
	644	console.error(
	645	`[autopilot] pr-risk-rescore: per-PR failure for pr=${cand.pullRequestId}:`,
	646	err
	647	);
	648	}
	649	}
	650	if (needsScoring.length > cap) {
	651	skipped += needsScoring.length - cap;
	652	}
	653
	654	return { scored, skipped };
	655	}
	656
2b9055eClaude	657	// ---------------------------------------------------------------------------
	658	// K3 — auto-merge-sweep
	659	// ---------------------------------------------------------------------------
	660
	661	interface SweepCandidate {
	662	prId: string;
	663	prNumber: number;
	664	prTitle: string;
	665	prBody: string \| null;
	666	baseBranch: string;
	667	headBranch: string;
	668	isDraft: boolean;
	669	repositoryId: string;
	670	authorUserId: string;
	671	ownerUsername: string \| null;
	672	repoName: string;
	673	state: string;
	674	}
	675
	676	export interface AutoMergeSweepDeps {
	677	/** Inject candidate-finder for tests. */
	678	findCandidates?: (lookbackHours: number, limit: number) => Promise<SweepCandidate[]>;
	679	/** Inject evaluator for tests. */
	680	evaluate?: (ctx: AutoMergeContext) => Promise<AutoMergeDecision>;
	681	/** Inject the merge executor for tests. */
	682	merge?: (cand: SweepCandidate) => Promise<PerformMergeResult>;
	683	/** Inject the audit-recording side-effect for tests. */
	684	recordAttempt?: (
	685	repoId: string,
	686	prId: string,
	687	decision: AutoMergeDecision
	688	) => Promise<void>;
	689	/** Inject the audit/comment side-effects for the merged path (tests). */
	690	onMerged?: (
	691	cand: SweepCandidate,
	692	result: PerformMergeResult
	693	) => Promise<void>;
	694	/** Inject the audit side-effect for the merge-failed path (tests). */
	695	onMergeFailed?: (cand: SweepCandidate, error: string) => Promise<void>;
	696	/** Inject the AI-key short-circuit signal for tests. */
	697	shouldShortCircuitAi?: (cand: SweepCandidate) => Promise<boolean>;
	698	}
	699
	700	export interface AutoMergeSweepSummary {
	701	evaluated: number;
	702	merged: number;
	703	blocked: number;
	704	}
	705
	706	/**
	707	* Default candidate-finder. Selects open, non-draft PRs from non-archived
	708	* repos whose `updated_at` is within the lookback window. Joins repo +
	709	* owner so the merge executor doesn't need extra round trips. Cap is
	710	* enforced at the SQL layer.
	711	*/
	712	async function defaultFindAutoMergeCandidates(
	713	lookbackHours: number,
	714	limit: number
	715	): Promise<SweepCandidate[]> {
	716	const cutoff = new Date(Date.now() - lookbackHours * 60 * 60 * 1000);
	717	try {
	718	const rows = await db
	719	.select({
	720	prId: pullRequests.id,
	721	prNumber: pullRequests.number,
	722	prTitle: pullRequests.title,
	723	prBody: pullRequests.body,
	724	baseBranch: pullRequests.baseBranch,
	725	headBranch: pullRequests.headBranch,
	726	isDraft: pullRequests.isDraft,
	727	repositoryId: pullRequests.repositoryId,
	728	authorUserId: pullRequests.authorId,
	729	ownerUsername: users.username,
	730	repoName: repositories.name,
	731	state: pullRequests.state,
	732	})
	733	.from(pullRequests)
	734	.innerJoin(
	735	repositories,
	736	eq(repositories.id, pullRequests.repositoryId)
	737	)
	738	.leftJoin(users, eq(users.id, repositories.ownerId))
	739	.where(
	740	and(
	741	eq(pullRequests.state, "open"),
	742	eq(pullRequests.isDraft, false),
	743	eq(repositories.isArchived, false),
	744	gte(pullRequests.updatedAt, cutoff)
	745	)
	746	)
	747	.limit(limit);
	748	return rows.map((r) => ({
	749	prId: r.prId,
	750	prNumber: r.prNumber,
	751	prTitle: r.prTitle,
	752	prBody: r.prBody,
	753	baseBranch: r.baseBranch,
	754	headBranch: r.headBranch,
	755	isDraft: r.isDraft,
	756	repositoryId: r.repositoryId,
	757	authorUserId: r.authorUserId,
	758	ownerUsername: r.ownerUsername ?? null,
	759	repoName: r.repoName,
	760	state: r.state,
	761	}));
	762	} catch (err) {
	763	console.error("[autopilot] auto-merge: candidate query failed:", err);
	764	return [];
	765	}
	766	}
	767
	768	/**
	769	* Determine whether the matched branch_protection rule on this PR
	770	* requires AI approval but no `ANTHROPIC_API_KEY` is configured. In that
	771	* case the AI-approval check would inevitably fail downstream, so we
	772	* short-circuit to a "blocked" decision without invoking `evaluateAutoMerge`
	773	* — keeps the log readable and prevents misleading "AI review unavailable"
	774	* lines in the audit trail.
	775	*/
	776	async function defaultShouldShortCircuitAi(
	777	cand: SweepCandidate
	778	): Promise<boolean> {
	779	if (process.env.ANTHROPIC_API_KEY) return false;
	780	try {
	781	const rule = await matchProtection(cand.repositoryId, cand.baseBranch);
	782	return !!(rule && rule.requireAiApproval);
	783	} catch {
	784	return false;
	785	}
	786	}
	787
	788	/**
	789	* Default success-path: post an `auto_merge.merged` audit row + a stable
	790	* marker comment on the PR so a partial-merge retry doesn't double-post.
	791	* Both are best-effort; failures are logged not thrown.
	792	*/
	793	async function defaultOnMerged(
	794	cand: SweepCandidate,
	795	result: PerformMergeResult
	796	): Promise<void> {
	797	try {
	798	await audit({
	799	repositoryId: cand.repositoryId,
	800	action: "auto_merge.merged",
	801	targetType: "pull_request",
	802	targetId: cand.prId,
	803	metadata: {
	804	prNumber: cand.prNumber,
	805	baseBranch: cand.baseBranch,
	806	headBranch: cand.headBranch,
	807	closedIssueNumbers: result.closedIssueNumbers,
	808	resolvedFiles: result.resolvedFiles,
	809	},
	810	});
	811	} catch (err) {
	812	console.error("[autopilot] auto-merge: merged audit failed:", err);
	813	}
	814	try {
	815	await db.insert(prComments).values({
	816	pullRequestId: cand.prId,
	817	authorId: cand.authorUserId,
	818	isAiReview: true,
	819	body: `${AUTO_MERGE_COMMENT_MARKER}\nAuto-merged by Gluecron autopilot — branch protection conditions satisfied.`,
	820	});
	821	} catch (err) {
	822	console.error("[autopilot] auto-merge: comment insert failed:", err);
	823	}
	824	}
	825
	826	/** Default failure-path: only an audit row; no comment (we may retry). */
	827	async function defaultOnMergeFailed(
	828	cand: SweepCandidate,
	829	error: string
	830	): Promise<void> {
	831	try {
	832	await audit({
	833	repositoryId: cand.repositoryId,
	834	action: "auto_merge.merge_failed",
	835	targetType: "pull_request",
	836	targetId: cand.prId,
	837	metadata: {
	838	prNumber: cand.prNumber,
	839	baseBranch: cand.baseBranch,
	840	headBranch: cand.headBranch,
	841	error,
	842	},
	843	});
	844	} catch (err) {
	845	console.error("[autopilot] auto-merge: merge_failed audit failed:", err);
	846	}
	847	}
	848
	849	/**
	850	* Execute one sweep over recently-updated open PRs. For each, evaluate
	851	* with K2's `evaluateAutoMerge`; on `merge: true`, call `performMerge` and
	852	* record the merged/merge-failed audit row + comment. Always record the
	853	* `auto_merge.evaluated` audit row via `recordAutoMergeAttempt`.
	854	*
	855	* Returns a counts summary that the autopilot prints as the tick log line.
	856	* Never throws.
	857	*/
	858	export async function runAutoMergeSweep(
	859	deps: AutoMergeSweepDeps = {}
	860	): Promise<AutoMergeSweepSummary> {
	861	const findCandidates = deps.findCandidates ?? defaultFindAutoMergeCandidates;
	862	const evaluate =
	863	deps.evaluate ?? ((ctx) => evaluateAutoMerge(ctx, {}));
	864	const merge =
	865	deps.merge ??
	866	(async (cand) => {
	867	if (!cand.ownerUsername) {
	868	return {
	869	ok: false,
	870	error: "owner username unresolved",
	871	closedIssueNumbers: [],
	872	resolvedFiles: [],
	873	};
	874	}
	875	return performMerge({
	876	pr: {
	877	id: cand.prId,
	878	number: cand.prNumber,
	879	title: cand.prTitle,
	880	body: cand.prBody,
	881	baseBranch: cand.baseBranch,
	882	headBranch: cand.headBranch,
	883	repositoryId: cand.repositoryId,
	884	authorId: cand.authorUserId,
	885	state: cand.state as "open",
	886	isDraft: cand.isDraft,
	887	},
	888	ownerName: cand.ownerUsername,
	889	repoName: cand.repoName,
	890	actorUserId: cand.authorUserId,
	891	});
	892	});
	893	const recordAttempt = deps.recordAttempt ?? recordAutoMergeAttempt;
	894	const onMerged = deps.onMerged ?? defaultOnMerged;
	895	const onMergeFailed = deps.onMergeFailed ?? defaultOnMergeFailed;
	896	const shouldShortCircuitAi =
	897	deps.shouldShortCircuitAi ?? defaultShouldShortCircuitAi;
	898
	899	let candidates: SweepCandidate[] = [];
	900	try {
	901	candidates = await findCandidates(
	902	AUTO_MERGE_LOOKBACK_HOURS,
	903	AUTO_MERGE_MAX_PER_TICK
	904	);
	905	} catch (err) {
	906	console.error("[autopilot] auto-merge: findCandidates threw:", err);
	907	return { evaluated: 0, merged: 0, blocked: 0 };
	908	}
	909
	910	let evaluated = 0;
	911	let merged = 0;
	912	let blocked = 0;
	913
	914	for (const cand of candidates) {
	915	try {
	916	evaluated += 1;
	917
	918	// AI-key short-circuit: if the rule requires AI approval and we have
	919	// no key, treat as blocked without calling the evaluator (which would
	920	// log a misleading "AI review unavailable").
	921	let decision: AutoMergeDecision;
	922	if (await shouldShortCircuitAi(cand)) {
	923	decision = {
	924	merge: false,
	925	reason:
	926	"Branch protection requires AI approval but ANTHROPIC_API_KEY is unset.",
	927	blocking: [
	928	"ANTHROPIC_API_KEY missing; AI approval cannot be sourced.",
	929	],
	930	};
	931	} else {
	932	decision = await evaluate({
	933	pullRequestId: cand.prId,
	934	repositoryId: cand.repositoryId,
	935	baseBranch: cand.baseBranch,
	936	isDraft: cand.isDraft,
	937	authorUserId: cand.authorUserId,
	938	});
	939	}
	940
	941	// Always record the evaluation, regardless of outcome.
	942	try {
	943	await recordAttempt(cand.repositoryId, cand.prId, decision);
	944	} catch (err) {
	945	console.error(
	946	`[autopilot] auto-merge: recordAttempt failed for pr=${cand.prId}:`,
	947	err
	948	);
	949	}
	950
	951	if (!decision.merge) {
	952	blocked += 1;
	953	continue;
	954	}
	955
	956	// Perform the actual merge.
	957	const result = await merge(cand);
	958	if (result.ok) {
	959	merged += 1;
	960	await onMerged(cand, result);
	961	} else {
	962	blocked += 1;
	963	await onMergeFailed(cand, result.error \|\| "unknown merge error");
	964	}
	965	} catch (err) {
	966	blocked += 1;
	967	console.error(
	968	`[autopilot] auto-merge: per-PR failure for pr=${cand.prId}:`,
	969	err
	970	);
	971	}
	972	}
	973
	974	console.log(
	975	`[autopilot] auto-merge: evaluated=${evaluated} merged=${merged} blocked=${blocked}`
	976	);
	977
	978	return { evaluated, merged, blocked };
	979	}
	980
2b821b7Claude	981	/**
	982	* Visits each distinct (repo, base_branch) that has queued rows and logs a
	983	* stub depth line. The actual gate-running + merge happens in the pulls
	984	* route; this tick is just a heartbeat so we can wire per-queue progress
	985	* through without duplicating merge logic.
	986	*/
	987	async function processMergeQueues(): Promise<void> {
	988	let distinct: Array<{ repositoryId: string; baseBranch: string }> = [];
	989	try {
	990	const rows = await db
	991	.selectDistinct({
	992	repositoryId: mergeQueueEntries.repositoryId,
	993	baseBranch: mergeQueueEntries.baseBranch,
	994	})
	995	.from(mergeQueueEntries)
	996	.where(sql`${mergeQueueEntries.state} IN ('queued','running')`);
	997	distinct = rows;
	998	} catch (err) {
	999	console.error("[autopilot] merge-queue: distinct query failed:", err);
	1000	return;
	1001	}
	1002	for (const d of distinct) {
	1003	try {
	1004	const head = await peekHead(d.repositoryId, d.baseBranch);
	1005	if (head) {
	1006	console.log(
	1007	`[autopilot] merge queue depth head=${head.id.slice(0, 8)} repo=${d.repositoryId.slice(0, 8)} base=${d.baseBranch}`
	1008	);
	1009	}
	1010	} catch (err) {
	1011	console.error(
	1012	`[autopilot] merge-queue: peek failed for repo=${d.repositoryId}:`,
	1013	err
	1014	);
	1015	}
	1016	}
	1017	}
	1018
	1019	/**
	1020	* Pick a small batch of repos that actually have dep rows and re-run
	1021	* advisory scan against them. Cheap — one SELECT DISTINCT with LIMIT.
	1022	*/
	1023	async function rescanAdvisoriesBatch(limit: number): Promise<void> {
	1024	let repoIds: string[] = [];
	1025	try {
	1026	const rows = await db
	1027	.selectDistinct({ repositoryId: repoDependencies.repositoryId })
	1028	.from(repoDependencies)
	1029	.limit(limit);
	1030	repoIds = rows.map((r) => r.repositoryId);
	1031	} catch (err) {
	1032	console.error("[autopilot] advisory-rescan: query failed:", err);
	1033	return;
	1034	}
	1035	for (const id of repoIds) {
	1036	try {
	1037	await scanRepositoryForAlerts(id);
	1038	} catch (err) {
	1039	console.error(
	1040	`[autopilot] advisory-rescan: scan failed for repo=${id}:`,
	1041	err
	1042	);
	1043	}
	1044	}
	1045	}
	1046
	1047	/** Resolve the tick interval from env → opts → default. */
	1048	function resolveIntervalMs(optsMs?: number): number {
	1049	if (typeof optsMs === "number" && optsMs > 0) return optsMs;
	1050	const raw = process.env.AUTOPILOT_INTERVAL_MS;
	1051	if (raw) {
	1052	const parsed = Number(raw);
	1053	if (Number.isFinite(parsed) && parsed > 0) return parsed;
	1054	}
	1055	return DEFAULT_INTERVAL_MS;
	1056	}
	1057
	1058	/**
	1059	* Start the recurring autopilot loop. No-op when AUTOPILOT_DISABLED=1.
	1060	* The first tick fires after `intervalMs`, not immediately, to keep boot
	1061	* fast. Returns a `stop()` that clears the interval.
	1062	*/
	1063	export function startAutopilot(opts?: StartAutopilotOpts): { stop: () => void } {
	1064	if (process.env.AUTOPILOT_DISABLED === "1") {
	1065	return { stop: () => {} };
	1066	}
	1067	const intervalMs = resolveIntervalMs(opts?.intervalMs);
	1068	const tasks = opts?.tasks ?? defaultTasks();
	1069	let running = false;
	1070	const handle = setInterval(() => {
	1071	if (running) return;
	1072	running = true;
	1073	void runAutopilotTick({ tasks, now: opts?.now })
	1074	.catch(() => {
	1075	// runAutopilotTick already never throws, but belt-and-braces.
	1076	})
	1077	.finally(() => {
	1078	running = false;
	1079	});
	1080	}, intervalMs);
	1081	return {
	1082	stop: () => clearInterval(handle),
	1083	};
	1084	}
	1085
8e9f1d9Claude	1086	/** Last tick snapshot for observability. Module-level, swap-on-complete. */
	1087	let lastTick: AutopilotTickResult \| null = null;
	1088	let tickCount = 0;
	1089
	1090	/** Return the most recent completed tick, or null if autopilot hasn't run yet. */
	1091	export function getLastTick(): AutopilotTickResult \| null {
	1092	return lastTick;
	1093	}
	1094
	1095	/** Return the total number of completed ticks in this process. */
	1096	export function getTickCount(): number {
	1097	return tickCount;
	1098	}
	1099
2b821b7Claude	1100	/**
	1101	* Run one tick: invokes every sub-task with its own try/catch, records a
	1102	* per-task result, and emits a single summary line. Never throws.
	1103	*/
	1104	export async function runAutopilotTick(
	1105	opts?: RunTickOpts
	1106	): Promise<AutopilotTickResult> {
	1107	const now = opts?.now ?? Date.now;
	1108	const tasks = opts?.tasks ?? defaultTasks();
	1109	const startedAt = new Date(now()).toISOString();
	1110	const results: AutopilotTaskResult[] = [];
	1111	for (const t of tasks) {
	1112	const t0 = now();
	1113	try {
	1114	await t.run();
	1115	results.push({ name: t.name, ok: true, durationMs: now() - t0 });
	1116	} catch (err) {
	1117	const message =
	1118	err instanceof Error ? err.message : String(err ?? "unknown error");
	1119	console.error(`[autopilot] ${t.name}: ${message}`);
	1120	results.push({
	1121	name: t.name,
	1122	ok: false,
	1123	durationMs: now() - t0,
	1124	error: message,
	1125	});
	1126	}
	1127	}
	1128	const finishedAt = new Date(now()).toISOString();
	1129	const totalMs = results.reduce((a, r) => a + r.durationMs, 0);
	1130	const okCount = results.filter((r) => r.ok).length;
	1131	console.log(
	1132	`[autopilot] tick ok tasks=${okCount}/${results.length} ms=${totalMs}`
	1133	);
8e9f1d9Claude	1134	const result: AutopilotTickResult = { startedAt, finishedAt, tasks: results };
	1135	lastTick = result;
	1136	tickCount += 1;
	1137	return result;
2b821b7Claude	1138	}
	1139
	1140	/** Exposed for unit tests. */
	1141	export const __test = {
	1142	resolveIntervalMs,
	1143	processMergeQueues,
	1144	rescanAdvisoriesBatch,
	1145	DEFAULT_INTERVAL_MS,
	1146	ADVISORY_RESCAN_BATCH,
2b9055eClaude	1147	AUTO_MERGE_LOOKBACK_HOURS,
	1148	AUTO_MERGE_MAX_PER_TICK,
	1149	AUTO_MERGE_COMMENT_MARKER,
534f04aClaude	1150	PR_RISK_RESCORE_MAX_PER_TICK,
	1151	PR_RISK_RESCORE_LOOKBACK_HOURS,
2b9055eClaude	1152	defaultFindAutoMergeCandidates,
	1153	defaultOnMerged,
	1154	defaultOnMergeFailed,
	1155	defaultShouldShortCircuitAi,
534f04aClaude	1156	defaultFindPrRiskCandidates,
	1157	defaultFilterNeedsScoring,
2b821b7Claude	1158	};