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