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.tsBlame258 lines · 1 contributor

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
	12	import { sql } from "drizzle-orm";
	13	import { db } from "../db";
	14	import { mergeQueueEntries, repoDependencies } from "../db/schema";
	15	import { syncAllDue } from "./mirrors";
	16	import { peekHead } from "./merge-queue";
	17	import { sendDigestsToAll } from "./email-digest";
	18	import { scanRepositoryForAlerts } from "./advisories";
a76d984Claude	19	import { releaseExpiredWaitTimers } from "./environments";
2b821b7Claude	20
	21	export interface AutopilotTaskResult {
	22	name: string;
	23	ok: boolean;
	24	durationMs: number;
	25	error?: string;
	26	}
	27
	28	export interface AutopilotTickResult {
	29	startedAt: string;
	30	finishedAt: string;
	31	tasks: AutopilotTaskResult[];
	32	}
	33
	34	export interface AutopilotTask {
	35	name: string;
	36	run: () => Promise<void>;
	37	}
	38
	39	export interface StartAutopilotOpts {
	40	intervalMs?: number;
	41	now?: () => number;
	42	tasks?: AutopilotTask[];
	43	}
	44
	45	export interface RunTickOpts {
	46	tasks?: AutopilotTask[];
	47	now?: () => number;
	48	}
	49
	50	const DEFAULT_INTERVAL_MS = 5 * 60 * 1000;
	51	const ADVISORY_RESCAN_BATCH = 5;
	52
	53	/**
	54	* Default task set. Each task is a thin wrapper around an existing locked
	55	* helper — no gate/merge logic is duplicated here.
	56	*/
	57	export function defaultTasks(): AutopilotTask[] {
	58	return [
	59	{
	60	name: "mirror-sync",
	61	run: async () => {
	62	await syncAllDue();
	63	},
	64	},
	65	{
	66	name: "merge-queue",
	67	run: async () => {
	68	await processMergeQueues();
	69	},
	70	},
	71	{
	72	name: "weekly-digest",
	73	run: async () => {
	74	await sendDigestsToAll();
	75	},
	76	},
	77	{
	78	name: "advisory-rescan",
	79	run: async () => {
	80	await rescanAdvisoriesBatch(ADVISORY_RESCAN_BATCH);
	81	},
	82	},
a76d984Claude	83	{
	84	name: "wait-timer-release",
	85	run: async () => {
	86	await releaseExpiredWaitTimers();
	87	},
	88	},
2b821b7Claude	89	];
	90	}
	91
	92	/**
	93	* Visits each distinct (repo, base_branch) that has queued rows and logs a
	94	* stub depth line. The actual gate-running + merge happens in the pulls
	95	* route; this tick is just a heartbeat so we can wire per-queue progress
	96	* through without duplicating merge logic.
	97	*/
	98	async function processMergeQueues(): Promise<void> {
	99	let distinct: Array<{ repositoryId: string; baseBranch: string }> = [];
	100	try {
	101	const rows = await db
	102	.selectDistinct({
	103	repositoryId: mergeQueueEntries.repositoryId,
	104	baseBranch: mergeQueueEntries.baseBranch,
	105	})
	106	.from(mergeQueueEntries)
	107	.where(sql`${mergeQueueEntries.state} IN ('queued','running')`);
	108	distinct = rows;
	109	} catch (err) {
	110	console.error("[autopilot] merge-queue: distinct query failed:", err);
	111	return;
	112	}
	113	for (const d of distinct) {
	114	try {
	115	const head = await peekHead(d.repositoryId, d.baseBranch);
	116	if (head) {
	117	console.log(
	118	`[autopilot] merge queue depth head=${head.id.slice(0, 8)} repo=${d.repositoryId.slice(0, 8)} base=${d.baseBranch}`
	119	);
	120	}
	121	} catch (err) {
	122	console.error(
	123	`[autopilot] merge-queue: peek failed for repo=${d.repositoryId}:`,
	124	err
	125	);
	126	}
	127	}
	128	}
	129
	130	/**
	131	* Pick a small batch of repos that actually have dep rows and re-run
	132	* advisory scan against them. Cheap — one SELECT DISTINCT with LIMIT.
	133	*/
	134	async function rescanAdvisoriesBatch(limit: number): Promise<void> {
	135	let repoIds: string[] = [];
	136	try {
	137	const rows = await db
	138	.selectDistinct({ repositoryId: repoDependencies.repositoryId })
	139	.from(repoDependencies)
	140	.limit(limit);
	141	repoIds = rows.map((r) => r.repositoryId);
	142	} catch (err) {
	143	console.error("[autopilot] advisory-rescan: query failed:", err);
	144	return;
	145	}
	146	for (const id of repoIds) {
	147	try {
	148	await scanRepositoryForAlerts(id);
	149	} catch (err) {
	150	console.error(
	151	`[autopilot] advisory-rescan: scan failed for repo=${id}:`,
	152	err
	153	);
	154	}
	155	}
	156	}
	157
	158	/** Resolve the tick interval from env → opts → default. */
	159	function resolveIntervalMs(optsMs?: number): number {
	160	if (typeof optsMs === "number" && optsMs > 0) return optsMs;
	161	const raw = process.env.AUTOPILOT_INTERVAL_MS;
	162	if (raw) {
	163	const parsed = Number(raw);
	164	if (Number.isFinite(parsed) && parsed > 0) return parsed;
	165	}
	166	return DEFAULT_INTERVAL_MS;
	167	}
	168
	169	/**
	170	* Start the recurring autopilot loop. No-op when AUTOPILOT_DISABLED=1.
	171	* The first tick fires after `intervalMs`, not immediately, to keep boot
	172	* fast. Returns a `stop()` that clears the interval.
	173	*/
	174	export function startAutopilot(opts?: StartAutopilotOpts): { stop: () => void } {
	175	if (process.env.AUTOPILOT_DISABLED === "1") {
	176	return { stop: () => {} };
	177	}
	178	const intervalMs = resolveIntervalMs(opts?.intervalMs);
	179	const tasks = opts?.tasks ?? defaultTasks();
	180	let running = false;
	181	const handle = setInterval(() => {
	182	if (running) return;
	183	running = true;
	184	void runAutopilotTick({ tasks, now: opts?.now })
	185	.catch(() => {
	186	// runAutopilotTick already never throws, but belt-and-braces.
	187	})
	188	.finally(() => {
	189	running = false;
	190	});
	191	}, intervalMs);
	192	return {
	193	stop: () => clearInterval(handle),
	194	};
	195	}
	196
8e9f1d9Claude	197	/** Last tick snapshot for observability. Module-level, swap-on-complete. */
	198	let lastTick: AutopilotTickResult \| null = null;
	199	let tickCount = 0;
	200
	201	/** Return the most recent completed tick, or null if autopilot hasn't run yet. */
	202	export function getLastTick(): AutopilotTickResult \| null {
	203	return lastTick;
	204	}
	205
	206	/** Return the total number of completed ticks in this process. */
	207	export function getTickCount(): number {
	208	return tickCount;
	209	}
	210
2b821b7Claude	211	/**
	212	* Run one tick: invokes every sub-task with its own try/catch, records a
	213	* per-task result, and emits a single summary line. Never throws.
	214	*/
	215	export async function runAutopilotTick(
	216	opts?: RunTickOpts
	217	): Promise<AutopilotTickResult> {
	218	const now = opts?.now ?? Date.now;
	219	const tasks = opts?.tasks ?? defaultTasks();
	220	const startedAt = new Date(now()).toISOString();
	221	const results: AutopilotTaskResult[] = [];
	222	for (const t of tasks) {
	223	const t0 = now();
	224	try {
	225	await t.run();
	226	results.push({ name: t.name, ok: true, durationMs: now() - t0 });
	227	} catch (err) {
	228	const message =
	229	err instanceof Error ? err.message : String(err ?? "unknown error");
	230	console.error(`[autopilot] ${t.name}: ${message}`);
	231	results.push({
	232	name: t.name,
	233	ok: false,
	234	durationMs: now() - t0,
	235	error: message,
	236	});
	237	}
	238	}
	239	const finishedAt = new Date(now()).toISOString();
	240	const totalMs = results.reduce((a, r) => a + r.durationMs, 0);
	241	const okCount = results.filter((r) => r.ok).length;
	242	console.log(
	243	`[autopilot] tick ok tasks=${okCount}/${results.length} ms=${totalMs}`
	244	);
8e9f1d9Claude	245	const result: AutopilotTickResult = { startedAt, finishedAt, tasks: results };
	246	lastTick = result;
	247	tickCount += 1;
	248	return result;
2b821b7Claude	249	}
	250
	251	/** Exposed for unit tests. */
	252	export const __test = {
	253	resolveIntervalMs,
	254	processMergeQueues,
	255	rescanAdvisoriesBatch,
	256	DEFAULT_INTERVAL_MS,
	257	ADVISORY_RESCAN_BATCH,
	258	};