Blame · Line-by-line history

`ai-flywheel.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.

ai-flywheel.tsBlame286 lines · 1 contributor

Normal view Raw

40e7738Claude	1	/**
	2	* AI flywheel — telemetry + live feed for every AI invocation.
	3	*
	4	* Two responsibilities:
	5	* 1. Persist a row into `ai_activity` for every AI call (model, latency,
	6	* success, summary, optional repo/user/PR anchors).
	7	* 2. Publish a small JSON event onto two SSE topics so the live dashboard
	8	* and any per-repo "AI in action" panels can render the work in motion:
	9	*
	10	* ai:global → every AI event in the system
	11	* ai:repo:<repositoryId> → events anchored to that repo
	12	*
	13	* NEVER throws into the request path. Telemetry failures must not break AI
	14	* features — every DB call and every publish is wrapped. Callers can use
	15	* either:
	16	*
	17	* await recordAi({...meta}, async () => callAnthropic(...))
	18	*
	19	* or, for fire-and-forget logging without a wrapped call:
	20	*
	21	* logAiEvent({...meta, latencyMs, success: true})
	22	*/
	23
	24	import { db } from "../db";
	25	import { aiActivity, type NewAiActivity } from "../db/schema";
	26	import { desc, eq, sql } from "drizzle-orm";
	27	import { publish } from "./sse";
	28
	29	export type AiActionType =
	30	\| "review"
	31	\| "repair"
	32	\| "completion"
	33	\| "incident"
	34	\| "triage"
	35	\| "explain"
	36	\| "test"
	37	\| "changelog"
	38	\| "chat"
	39	\| "spec"
	40	\| "commit-message"
	41	\| "pr-summary"
	42	\| "issue-triage"
	43	\| "merge-resolve"
	44	\| "security-scan"
	45	\| "dep-update"
	46	\| "semantic-index"
	47	\| "other";
	48
	49	export interface AiEventMeta {
	50	actionType: AiActionType;
	51	model: string;
	52	summary: string;
	53	repositoryId?: string \| null;
	54	userId?: string \| null;
	55	pullRequestId?: string \| null;
	56	issueId?: string \| null;
	57	commitSha?: string \| null;
	58	inputTokens?: number \| null;
	59	outputTokens?: number \| null;
	60	metadata?: Record<string, unknown> \| null;
	61	}
	62
	63	export interface AiEvent extends AiEventMeta {
	64	id: string;
	65	latencyMs: number;
	66	success: boolean;
	67	error?: string \| null;
	68	createdAt: string;
	69	}
	70
	71	/**
	72	* Wrap an AI call so its outcome is persisted + published. The callback may
	73	* return any value — the wrapper passes it through. On throw, telemetry is
	74	* still recorded with success=false and the error is rethrown so callers can
	75	* keep their existing error-handling behaviour.
	76	*/
	77	export async function recordAi<T>(
	78	meta: AiEventMeta,
	79	fn: () => Promise<T>
	80	): Promise<T> {
	81	const t0 = Date.now();
	82	try {
	83	const value = await fn();
	84	void persist({ ...meta, latencyMs: Date.now() - t0, success: true });
	85	return value;
	86	} catch (err) {
	87	const message = err instanceof Error ? err.message : String(err);
	88	void persist({
	89	...meta,
	90	latencyMs: Date.now() - t0,
	91	success: false,
	92	error: redact(message),
	93	});
	94	throw err;
	95	}
	96	}
	97
	98	/**
	99	* Fire-and-forget logger for callers that already have the result in hand
	100	* (e.g. cached completions where there was no real LLM call).
	101	*/
	102	export function logAiEvent(
	103	meta: AiEventMeta & {
	104	latencyMs: number;
	105	success?: boolean;
	106	error?: string \| null;
	107	}
	108	): void {
	109	void persist({
	110	...meta,
	111	success: meta.success ?? true,
	112	error: meta.error ?? null,
	113	});
	114	}
	115
	116	interface PersistArgs extends AiEventMeta {
	117	latencyMs: number;
	118	success: boolean;
	119	error?: string \| null;
	120	}
	121
	122	async function persist(args: PersistArgs): Promise<void> {
	123	let row: { id: string; createdAt: Date } \| null = null;
	124	try {
	125	const insert: NewAiActivity = {
	126	actionType: args.actionType,
	127	model: args.model,
	128	summary: clamp(args.summary, 500),
	129	repositoryId: args.repositoryId ?? null,
	130	userId: args.userId ?? null,
	131	pullRequestId: args.pullRequestId ?? null,
	132	issueId: args.issueId ?? null,
	133	commitSha: args.commitSha ?? null,
	134	inputTokens: args.inputTokens ?? null,
	135	outputTokens: args.outputTokens ?? null,
	136	latencyMs: args.latencyMs,
	137	success: args.success,
	138	error: args.error ? clamp(args.error, 1000) : null,
	139	metadata: args.metadata ?? null,
	140	};
	141	const [inserted] = await db
	142	.insert(aiActivity)
	143	.values(insert)
	144	.returning({ id: aiActivity.id, createdAt: aiActivity.createdAt });
	145	if (inserted) row = inserted;
	146	} catch (err) {
	147	// DB writes are best-effort. We still publish so the live UI sees the
	148	// event even when we cannot persist (e.g. fresh deploy without migration).
	149	safeWarn("[ai-flywheel] persist failed:", err);
	150	}
	151
	152	const event: AiEvent = {
	153	id: row?.id ?? cryptoId(),
	154	actionType: args.actionType,
	155	model: args.model,
	156	summary: args.summary,
	157	repositoryId: args.repositoryId ?? null,
	158	userId: args.userId ?? null,
	159	pullRequestId: args.pullRequestId ?? null,
	160	issueId: args.issueId ?? null,
	161	commitSha: args.commitSha ?? null,
	162	inputTokens: args.inputTokens ?? null,
	163	outputTokens: args.outputTokens ?? null,
	164	metadata: args.metadata ?? null,
	165	latencyMs: args.latencyMs,
	166	success: args.success,
	167	error: args.error ?? null,
	168	createdAt: (row?.createdAt ?? new Date()).toISOString(),
	169	};
	170
	171	try {
	172	publish("ai:global", { event: "ai", data: event });
	173	if (event.repositoryId) {
	174	publish(`ai:repo:${event.repositoryId}`, { event: "ai", data: event });
	175	}
	176	} catch (err) {
	177	safeWarn("[ai-flywheel] publish failed:", err);
	178	}
	179	}
	180
	181	export interface ListRecentOpts {
	182	limit?: number;
	183	repositoryId?: string;
	184	}
	185
	186	export async function listRecentAiEvents(
	187	opts: ListRecentOpts = {}
	188	): Promise<AiEvent[]> {
	189	const limit = clampInt(opts.limit ?? 50, 1, 200);
	190	try {
	191	const query = db
	192	.select()
	193	.from(aiActivity)
	194	.orderBy(desc(aiActivity.createdAt))
	195	.limit(limit);
	196	const rows = opts.repositoryId
	197	? await query.where(eq(aiActivity.repositoryId, opts.repositoryId))
	198	: await query;
	199	return rows.map(rowToEvent);
	200	} catch (err) {
	201	safeWarn("[ai-flywheel] list failed:", err);
	202	return [];
	203	}
	204	}
	205
	206	export interface RollupRow {
	207	actionType: string;
	208	total: number;
	209	successes: number;
	210	failures: number;
	211	avgLatencyMs: number;
	212	}
	213
	214	export async function rollupByAction(sinceHours = 24): Promise<RollupRow[]> {
	215	try {
	216	const rows = await db
	217	.select({
	218	actionType: aiActivity.actionType,
	219	total: sql<number>`count(*)::int`,
	220	successes: sql<number>`sum(case when ${aiActivity.success} then 1 else 0 end)::int`,
	221	failures: sql<number>`sum(case when ${aiActivity.success} then 0 else 1 end)::int`,
	222	avgLatencyMs: sql<number>`coalesce(avg(${aiActivity.latencyMs}), 0)::int`,
	223	})
	224	.from(aiActivity)
	225	.where(sql`${aiActivity.createdAt} > now() - (${sinceHours} \|\| ' hours')::interval`)
	226	.groupBy(aiActivity.actionType);
	227	return rows;
	228	} catch (err) {
	229	safeWarn("[ai-flywheel] rollup failed:", err);
	230	return [];
	231	}
	232	}
	233
	234	function rowToEvent(row: typeof aiActivity.$inferSelect): AiEvent {
	235	return {
	236	id: row.id,
	237	actionType: row.actionType as AiActionType,
	238	model: row.model,
	239	summary: row.summary,
	240	repositoryId: row.repositoryId,
	241	userId: row.userId,
	242	pullRequestId: row.pullRequestId,
	243	issueId: row.issueId,
	244	commitSha: row.commitSha,
	245	inputTokens: row.inputTokens,
	246	outputTokens: row.outputTokens,
	247	metadata: (row.metadata as Record<string, unknown>) ?? null,
	248	latencyMs: row.latencyMs,
	249	success: row.success,
	250	error: row.error,
	251	createdAt: row.createdAt.toISOString(),
	252	};
	253	}
	254
	255	function clamp(s: string, n: number): string {
	256	return s.length > n ? s.slice(0, n - 1) + "…" : s;
	257	}
	258	function clampInt(n: number, lo: number, hi: number): number {
	259	return Math.max(lo, Math.min(hi, Math.floor(n)));
	260	}
	261	function redact(msg: string): string {
	262	// Strip obvious bearer tokens / API keys before persisting. Patterns:
	263	// sk-... Anthropic / OpenAI keys
	264	// glc_... gluecron PAT
	265	// glct_... gluecron OAuth access token
	266	// ghi_... marketplace install token
	267	// Bearer... any explicit bearer header value
	268	return msg.replace(
	269	/(?:sk-[A-Za-z0-9_-]{16,}\|gl(?:c\|ct)_[A-Za-z0-9_-]{12,}\|ghi_[A-Za-z0-9_-]{12,}\|Bearer\s+[A-Za-z0-9_.\-]+)/g,
	270	"[REDACTED]"
	271	);
	272	}
	273	function cryptoId(): string {
	274	const b = new Uint8Array(16);
	275	crypto.getRandomValues(b);
	276	return Array.from(b, (x) => x.toString(16).padStart(2, "0")).join("");
	277	}
	278	function safeWarn(prefix: string, err: unknown): void {
	279	try {
	280	console.warn(prefix, err);
	281	} catch {
	282	/* ignore */
	283	}
	284	}
	285
	286	export const __test = { redact, clamp, clampInt };