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