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