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