Pre-launch — Gluecron is in final validation. Public signups and git hosting for non-owner users open after launch review.
ccantynz/Gluecron.com
CodeIssuesDiscussionsWikiPull RequestsProjectsCommitsActionsReleasesContributorsPulse● GatesSecuritySettingsDeploymentsPipelineInsightsAgents✨ Explain✨ Ask AI✨ Workspace✨ Spec✨ Tests▓ Debt Map✨ NL Search🏛 Archaeology
Blame · Line-by-line history

agent-runtime.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.

Gluecron.com/src/lib/agent-runtime.ts
agent-runtime.tsBlame613 lines · 1 contributor
Normal viewRaw
a6d8fd5Claude1/**
2 * Block K1 — Autonomous agent runtime + sandbox.
3 *
4 * The substrate every other Block K agent (triage, fix, review_response,
5 * deploy_watcher, heal_bot) runs on top of. A single `agent_runs` row records
6 * one invocation: its kind + trigger, lifecycle status, a size-capped
7 * append-only log, cost accounting, and an optional error message.
8 *
9 * File layout mirrors commit-statuses.ts:
10 * 1. Pure helpers (types, truncation, state machine) — fully unit-testable
11 * 2. Sandboxing primitive (runSandboxed) — Bun.spawn wrapper
12 * 3. DB helpers — wrap every call in try/catch
13 *
14 * Every DB helper returns null/false on failure and console.errors; this
15 * matches the "never crash the caller" philosophy of workflow-runner.ts and
16 * post-receive.ts.
17 */
18
19import { sql } from "drizzle-orm";
20import { db } from "../db";
21
22// ---------------------------------------------------------------------------
23// Pure helpers
24// ---------------------------------------------------------------------------
25
26export type AgentRunStatus =
27 | "queued"
28 | "running"
29 | "succeeded"
30 | "failed"
31 | "killed"
32 | "timeout";
33
34export type AgentRunTrigger =
35 | "issue.opened"
36 | "pr.opened"
37 | "pr.review_comment"
38 | "deploy.failed"
39 | "manual"
40 | "scheduled";
41
42export type AgentKind =
43 | "triage"
44 | "fix"
45 | "review_response"
46 | "deploy_watcher"
47 | "heal_bot"
48 | "custom";
49
50const TERMINAL_STATUSES: ReadonlySet<AgentRunStatus> = new Set([
51 "succeeded",
52 "failed",
53 "killed",
54 "timeout",
55]);
56
57/** True iff `s` is a final, immutable status. */
58export function isTerminalStatus(s: AgentRunStatus): boolean {
59 return TERMINAL_STATUSES.has(s);
60}
61
62/**
63 * Allowed transitions:
64 * queued → running | killed (killed lets an operator cancel before start)
65 * running → succeeded | failed | timeout | killed
66 * terminal → nothing
67 *
68 * Self-transitions are disallowed so a double-write doesn't silently "succeed".
69 */
70export function canTransition(
71 from: AgentRunStatus,
72 to: AgentRunStatus
73): boolean {
74 if (from === to) return false;
75 if (isTerminalStatus(from)) return false;
76 if (from === "queued") {
77 return to === "running" || to === "killed";
78 }
79 if (from === "running") {
80 return (
81 to === "succeeded" ||
82 to === "failed" ||
83 to === "timeout" ||
84 to === "killed"
85 );
86 }
87 return false;
88}
89
90const LOG_TRUNCATED_SENTINEL = "[log truncated]\n";
91
92/**
93 * Append `addition` to `existing` and cap the total at `maxBytes`.
94 *
95 * When the combined length exceeds the cap we keep only the last `maxBytes`
96 * characters and prepend the `[log truncated]\n` sentinel — but only once.
97 * If `existing` already starts with the sentinel we don't double it.
98 *
99 * Measurement is character-based (JS string .length) rather than UTF-8 byte
100 * length. For our log content (ASCII-dominant stdout/stderr) the two are
101 * equivalent to within a few percent; correctness of the cap in the
102 * pathological multibyte case is not worth the Buffer round-trip cost.
103 */
104export function truncateLog(
105 existing: string,
106 addition: string,
107 maxBytes: number = 256 * 1024
108): string {
109 const combined = (existing || "") + (addition || "");
110 if (combined.length <= maxBytes) return combined;
111
112 // Strip any prior sentinel so we don't double it after a re-truncation.
113 const stripped = combined.startsWith(LOG_TRUNCATED_SENTINEL)
114 ? combined.slice(LOG_TRUNCATED_SENTINEL.length)
115 : combined;
116
117 // Take the last maxBytes chars and prepend a single sentinel. Reserve space
118 // for the sentinel so the final string length is <= maxBytes + sentinel.
119 const tail = stripped.slice(-maxBytes);
120 return LOG_TRUNCATED_SENTINEL + tail;
121}
122
123/** Cap an error message / stack trace for DB storage. */
124export function truncateError(
125 msg: string,
126 maxChars: number = 4096
127): string {
128 if (!msg) return "";
129 if (msg.length <= maxChars) return msg;
130 return msg.slice(0, maxChars) + "\n[... truncated ...]";
131}
132
133// ---------------------------------------------------------------------------
134// Sandboxing
135// ---------------------------------------------------------------------------
136
137export interface SandboxOptions {
138 cwd: string;
139 env?: Record<string, string>;
140 timeoutMs?: number;
141 stdoutCapBytes?: number;
142 stderrCapBytes?: number;
143}
144
145export interface SandboxResult {
146 exitCode: number | null;
147 stdout: string;
148 stderr: string;
149 timedOut: boolean;
150}
151
152const DEFAULT_SANDBOX_TIMEOUT_MS = 10 * 60 * 1000; // 10 minutes
153const DEFAULT_SANDBOX_STREAM_CAP = 64 * 1024; // 64 KB per stream
154const SANDBOX_KILL_GRACE_MS = 5_000;
155
156function capStream(value: string, limit: number): string {
157 if (value.length <= limit) return value;
158 return value.slice(0, limit) + "\n[... truncated ...]";
159}
160
161/**
162 * Spawn `cmd args` under a minimal env with a hard timeout.
163 *
164 * Semantics:
165 * - env defaults to `{ PATH, HOME }` from the parent process if the caller
166 * passes nothing. This is deliberately narrower than `process.env` so
167 * secrets (DATABASE_URL, ANTHROPIC_API_KEY, etc) aren't accidentally
168 * leaked to agent-authored commands.
169 * - On timeout we send SIGTERM, then SIGKILL 5s later.
170 * - stdout/stderr are capped to 64 KB each by default; overflow is
171 * replaced with a `\n[... truncated ...]` sentinel.
172 * - Never throws — a spawn failure is reported as exitCode=null with the
173 * error message on stderr.
174 */
175export async function runSandboxed(
176 cmd: string,
177 args: string[],
178 opts: SandboxOptions
179): Promise<SandboxResult> {
180 const timeoutMs = opts.timeoutMs ?? DEFAULT_SANDBOX_TIMEOUT_MS;
181 const stdoutCap = opts.stdoutCapBytes ?? DEFAULT_SANDBOX_STREAM_CAP;
182 const stderrCap = opts.stderrCapBytes ?? DEFAULT_SANDBOX_STREAM_CAP;
183
184 const env = opts.env
185 ? opts.env
186 : {
187 PATH: process.env.PATH ?? "/usr/bin:/bin",
188 HOME: process.env.HOME ?? "/tmp",
189 };
190
191 let proc: ReturnType<typeof Bun.spawn> | null = null;
192 let timedOut = false;
193 let killTimer: ReturnType<typeof setTimeout> | null = null;
194 let escalateTimer: ReturnType<typeof setTimeout> | null = null;
195
196 try {
197 proc = Bun.spawn([cmd, ...args], {
198 cwd: opts.cwd,
199 env,
200 stdout: "pipe",
201 stderr: "pipe",
202 });
203
204 killTimer = setTimeout(() => {
205 timedOut = true;
206 try {
207 proc?.kill("SIGTERM");
208 } catch {
209 /* ignore */
210 }
211 escalateTimer = setTimeout(() => {
212 try {
213 proc?.kill("SIGKILL");
214 } catch {
215 /* ignore */
216 }
217 }, SANDBOX_KILL_GRACE_MS);
218 }, timeoutMs);
219
220 const stdoutPromise = proc.stdout
221 ? new Response(proc.stdout as ReadableStream).text()
222 : Promise.resolve("");
223 const stderrPromise = proc.stderr
224 ? new Response(proc.stderr as ReadableStream).text()
225 : Promise.resolve("");
226
227 const [stdoutRaw, stderrRaw] = await Promise.all([
228 stdoutPromise.catch(() => ""),
229 stderrPromise.catch(() => ""),
230 ]);
231 const exitCode = await proc.exited;
232
233 if (killTimer) clearTimeout(killTimer);
234 if (escalateTimer) clearTimeout(escalateTimer);
235
236 return {
237 exitCode,
238 stdout: capStream(stdoutRaw, stdoutCap),
239 stderr: capStream(
240 timedOut
241 ? `${stderrRaw}\n[sandbox killed after ${timeoutMs}ms timeout]`
242 : stderrRaw,
243 stderrCap
244 ),
245 timedOut,
246 };
247 } catch (err) {
248 if (killTimer) clearTimeout(killTimer);
249 if (escalateTimer) clearTimeout(escalateTimer);
250 return {
251 exitCode: null,
252 stdout: "",
253 stderr: capStream(
254 `[sandbox] spawn failed: ${(err as Error).message}`,
255 stderrCap
256 ),
257 timedOut,
258 };
259 }
260}
261
262// ---------------------------------------------------------------------------
263// DB helpers
264// ---------------------------------------------------------------------------
265
266/**
267 * Shape of a row in `agent_runs` — mirrors drizzle/0034_agent_runs.sql.
268 * Defined here (rather than imported from db/schema) because the Drizzle
269 * table is added by the main thread; this file stays self-contained so
270 * agents can build against it before the schema ships.
271 */
272export interface AgentRun {
273 id: string;
274 repositoryId: string;
275 kind: AgentKind | string;
276 trigger: AgentRunTrigger | string;
277 triggerRef: string | null;
278 status: AgentRunStatus;
279 summary: string | null;
280 log: string;
281 costInputTokens: number;
282 costOutputTokens: number;
283 costCents: number;
284 startedAt: Date | null;
285 finishedAt: Date | null;
286 createdAt: Date;
287 errorMessage: string | null;
288}
289
290/**
291 * Map a raw Postgres row (snake_case) to our TS shape. Neon returns values
292 * pre-parsed (timestamps as Date, ints as number) so we only rename.
293 */
294function rowToAgentRun(row: Record<string, unknown>): AgentRun {
295 return {
296 id: String(row.id),
297 repositoryId: String(row.repository_id),
298 kind: String(row.kind),
299 trigger: String(row.trigger),
300 triggerRef: (row.trigger_ref as string | null) ?? null,
301 status: String(row.status) as AgentRunStatus,
302 summary: (row.summary as string | null) ?? null,
303 log: String(row.log ?? ""),
304 costInputTokens: Number(row.cost_input_tokens ?? 0),
305 costOutputTokens: Number(row.cost_output_tokens ?? 0),
306 costCents: Number(row.cost_cents ?? 0),
307 startedAt: (row.started_at as Date | null) ?? null,
308 finishedAt: (row.finished_at as Date | null) ?? null,
309 createdAt: (row.created_at as Date) ?? new Date(0),
310 errorMessage: (row.error_message as string | null) ?? null,
311 };
312}
313
314export interface StartAgentRunInput {
315 repositoryId: string;
316 kind: AgentKind;
317 trigger: AgentRunTrigger;
318 triggerRef?: string | null;
319}
320
321/** Insert a queued run. Returns the row or null on DB failure. */
322export async function startAgentRun(
323 input: StartAgentRunInput
324): Promise<AgentRun | null> {
325 try {
326 const rows = (await db.execute(sql`
327 INSERT INTO agent_runs (repository_id, kind, trigger, trigger_ref, status)
328 VALUES (
329 ${input.repositoryId},
330 ${input.kind},
331 ${input.trigger},
332 ${input.triggerRef ?? null},
333 'queued'
334 )
335 RETURNING *
336 `)) as unknown as Array<Record<string, unknown>>;
337 const row = Array.isArray(rows) ? rows[0] : undefined;
338 return row ? rowToAgentRun(row) : null;
339 } catch (err) {
340 console.error("[agent-runtime] startAgentRun:", err);
341 return null;
342 }
343}
344
345export async function getAgentRun(id: string): Promise<AgentRun | null> {
346 try {
347 const rows = (await db.execute(sql`
348 SELECT * FROM agent_runs WHERE id = ${id} LIMIT 1
349 `)) as unknown as Array<Record<string, unknown>>;
350 const row = Array.isArray(rows) ? rows[0] : undefined;
351 return row ? rowToAgentRun(row) : null;
352 } catch (err) {
353 console.error("[agent-runtime] getAgentRun:", err);
354 return null;
355 }
356}
357
358export interface ListAgentRunsOptions {
359 limit?: number;
360 status?: AgentRunStatus;
361 kind?: AgentKind;
362}
363
364export async function listAgentRunsForRepo(
365 repositoryId: string,
366 opts: ListAgentRunsOptions = {}
367): Promise<AgentRun[]> {
368 const limit = Math.max(1, Math.min(opts.limit ?? 50, 500));
369 try {
370 // Filters are composed conditionally — sql`` interpolation handles
371 // parameterisation safely.
372 const statusFilter = opts.status
373 ? sql`AND status = ${opts.status}`
374 : sql``;
375 const kindFilter = opts.kind ? sql`AND kind = ${opts.kind}` : sql``;
376 const rows = (await db.execute(sql`
377 SELECT * FROM agent_runs
378 WHERE repository_id = ${repositoryId}
379 ${statusFilter}
380 ${kindFilter}
381 ORDER BY created_at DESC
382 LIMIT ${limit}
383 `)) as unknown as Array<Record<string, unknown>>;
384 if (!Array.isArray(rows)) return [];
385 return rows.map(rowToAgentRun);
386 } catch (err) {
387 console.error("[agent-runtime] listAgentRunsForRepo:", err);
388 return [];
389 }
390}
391
392async function setStatus(
393 runId: string,
394 status: AgentRunStatus,
395 extra: {
396 summary?: string | null;
397 errorMessage?: string | null;
398 setStartedAt?: boolean;
399 setFinishedAt?: boolean;
400 } = {}
401): Promise<boolean> {
402 try {
403 const summaryClause =
404 extra.summary !== undefined
405 ? sql`, summary = ${extra.summary}`
406 : sql``;
407 const errorClause =
408 extra.errorMessage !== undefined
409 ? sql`, error_message = ${extra.errorMessage}`
410 : sql``;
411 const startedClause = extra.setStartedAt ? sql`, started_at = now()` : sql``;
412 const finishedClause = extra.setFinishedAt
413 ? sql`, finished_at = now()`
414 : sql``;
415 await db.execute(sql`
416 UPDATE agent_runs
417 SET status = ${status}
418 ${summaryClause}
419 ${errorClause}
420 ${startedClause}
421 ${finishedClause}
422 WHERE id = ${runId}
423 `);
424 return true;
425 } catch (err) {
426 console.error("[agent-runtime] setStatus:", err);
427 return false;
428 }
429}
430
431export interface AgentExecutorContext {
432 appendLog: (line: string) => Promise<void>;
433 recordCost: (
434 inputTokens: number,
435 outputTokens: number,
436 cents: number
437 ) => Promise<void>;
438}
439
440export interface AgentExecutorResult {
441 ok: boolean;
442 summary: string;
443}
444
445export interface ExecuteAgentRunOptions {
446 timeoutMs?: number;
447}
448
449const DEFAULT_EXECUTOR_TIMEOUT_MS = 10 * 60 * 1000;
450
451/**
452 * Drive a queued run through its lifecycle. Never throws.
453 *
454 * 1. Transition queued → running (abort if transition fails).
455 * 2. Invoke `executor` with a context that lets it append logs + record cost.
456 * 3. If executor resolves: transition running → succeeded|failed + store summary.
457 * 4. If timeout fires first: transition running → timeout.
458 * 5. If executor throws: transition running → failed with truncated stack.
459 *
460 * Note: this wrapper does NOT hold a Bun.Subprocess handle — the executor
461 * owns any child process. On timeout we simply stop awaiting and flip DB
462 * state; the executor's runSandboxed call should be passed a timeout slightly
463 * lower than ours so its own SIGKILL happens first.
464 */
465export async function executeAgentRun<T extends AgentExecutorResult>(
466 runId: string,
467 executor: (ctx: AgentExecutorContext) => Promise<T>,
468 opts: ExecuteAgentRunOptions = {}
469): Promise<void> {
470 const timeoutMs = opts.timeoutMs ?? DEFAULT_EXECUTOR_TIMEOUT_MS;
471
472 const transitioned = await setStatus(runId, "running", {
473 setStartedAt: true,
474 });
475 if (!transitioned) {
476 console.error(
477 `[agent-runtime] executeAgentRun: could not transition ${runId} to running`
478 );
479 return;
480 }
481
482 const ctx: AgentExecutorContext = {
483 appendLog: async (line: string) => {
484 await appendAgentLog(runId, line);
485 },
486 recordCost: async (input, output, cents) => {
487 await recordAgentCost(runId, input, output, cents);
488 },
489 };
490
491 let timeoutHandle: ReturnType<typeof setTimeout> | null = null;
492 let didTimeout = false;
493 const timeoutPromise = new Promise<"__timeout__">((resolve) => {
494 timeoutHandle = setTimeout(() => {
495 didTimeout = true;
496 resolve("__timeout__");
497 }, timeoutMs);
498 });
499
500 try {
501 const race = await Promise.race([executor(ctx), timeoutPromise]);
502 if (timeoutHandle) clearTimeout(timeoutHandle);
503
504 if (didTimeout || race === "__timeout__") {
505 await setStatus(runId, "timeout", {
506 setFinishedAt: true,
507 errorMessage: truncateError(
508 `Agent run exceeded timeout of ${timeoutMs}ms`
509 ),
510 });
511 return;
512 }
513
514 const result = race as T;
515 await setStatus(runId, result.ok ? "succeeded" : "failed", {
516 setFinishedAt: true,
517 summary: (result.summary ?? "").slice(0, 2000),
518 });
519 } catch (err) {
520 if (timeoutHandle) clearTimeout(timeoutHandle);
521 const stack =
522 err instanceof Error
523 ? err.stack || err.message
524 : String(err);
525 await setStatus(runId, "failed", {
526 setFinishedAt: true,
527 errorMessage: truncateError(stack),
528 });
529 }
530}
531
532/**
533 * Best-effort kill: flip DB status from queued/running → killed. Does NOT
534 * terminate an in-flight subprocess — that subprocess is owned by whichever
535 * executor invoked runSandboxed. Cross-process SIGKILL would require tracking
536 * a Bun.Subprocess handle in memory, which doesn't survive restarts and is
537 * out of scope for v1. Operators wanting a real kill should restart the
538 * worker; the executor's own runSandboxed timeout will then bound runtime.
539 */
540export async function killAgentRun(runId: string): Promise<boolean> {
541 try {
542 await db.execute(sql`
543 UPDATE agent_runs
544 SET status = 'killed', finished_at = now()
545 WHERE id = ${runId}
546 AND status IN ('queued', 'running')
547 `);
548 return true;
549 } catch (err) {
550 console.error("[agent-runtime] killAgentRun:", err);
551 return false;
552 }
553}
554
555/**
556 * Append `line` to the run's log, capped at 256 KB. Re-reads the current
557 * log then writes back — good enough for v1 since a run has a single
558 * executor and therefore no concurrent writers.
559 */
560export async function appendAgentLog(
561 runId: string,
562 line: string
563): Promise<boolean> {
564 try {
565 const rows = (await db.execute(sql`
566 SELECT log FROM agent_runs WHERE id = ${runId} LIMIT 1
567 `)) as unknown as Array<Record<string, unknown>>;
568 const row = Array.isArray(rows) ? rows[0] : undefined;
569 if (!row) return false;
570 const existing = String(row.log ?? "");
571 const next = truncateLog(existing, line.endsWith("\n") ? line : line + "\n");
572 await db.execute(sql`
573 UPDATE agent_runs SET log = ${next} WHERE id = ${runId}
574 `);
575 return true;
576 } catch (err) {
577 console.error("[agent-runtime] appendAgentLog:", err);
578 return false;
579 }
580}
581
582/** Atomically add to the cost counters. */
583export async function recordAgentCost(
584 runId: string,
585 inputTokens: number,
586 outputTokens: number,
587 cents: number
588): Promise<boolean> {
589 try {
590 const dIn = Math.max(0, Math.floor(inputTokens || 0));
591 const dOut = Math.max(0, Math.floor(outputTokens || 0));
592 const dCents = Math.max(0, Math.floor(cents || 0));
593 await db.execute(sql`
594 UPDATE agent_runs
595 SET cost_input_tokens = cost_input_tokens + ${dIn},
596 cost_output_tokens = cost_output_tokens + ${dOut},
597 cost_cents = cost_cents + ${dCents}
598 WHERE id = ${runId}
599 `);
600 return true;
601 } catch (err) {
602 console.error("[agent-runtime] recordAgentCost:", err);
603 return false;
604 }
605}
606
607export const __internal = {
608 LOG_TRUNCATED_SENTINEL,
609 DEFAULT_SANDBOX_TIMEOUT_MS,
610 DEFAULT_SANDBOX_STREAM_CAP,
611 SANDBOX_KILL_GRACE_MS,
612 DEFAULT_EXECUTOR_TIMEOUT_MS,
613};