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