Blame · Line-by-line history

`claude-web-session.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.

claude-web-session.tsBlame404 lines · 1 contributor

Normal view Raw

90c7531Claude	1	/**
	2	* Claude-on-the-web runtime (Block CW).
	3	*
	4	* Drives one turn of an interactive Claude Code session running on the
	5	* gluecron web server. Each turn:
	6	* 1. Resolves (and lazily clones) the session's working directory.
	7	* 2. Spawns `claude` as a subprocess pointed at that workdir, with
	8	* --resume <session-uuid> for follow-up turns so prior context is
	9	* preserved without us re-sending the transcript.
	10	* 3. Streams stdout chunks back via an AsyncIterable the route layer
	11	* hands to an SSE response.
	12	* 4. Persists the assistant turn + updates the Claude session UUID +
	13	* the last_active_at watermark.
	14	*
	15	* v1 design rules:
	16	* - Admin-only — gated by the route, not this lib.
	17	* - Single shared compute: every session shares the web server's CPU
	18	* and disk. We cap a single turn at MAX_TURN_MS so a runaway prompt
	19	* can't pin the box forever.
	20	* - No container: workdir is a plain directory under CLAUDE_WEB_WORKDIR
	21	* (default /var/lib/gluecron/claude-web). Caller is responsible for
	22	* not pointing this at a tenant-shared filesystem.
	23	* - Anthropic creds: we DO NOT pass ANTHROPIC_API_KEY through the
	24	* environment if the operator has logged in `claude` interactively
	25	* (the CLI manages its own creds in ~/.claude). If the env var is
	26	* set, we pass it through unchanged.
	27	*
	28	* Test seam: `__setSpawnForTests` lets unit tests intercept the spawn
	29	* without actually running the Claude CLI.
	30	*/
	31
	32	import { mkdir, stat } from "fs/promises";
	33	import { join } from "path";
	34	import { db } from "../db";
	35	import {
	36	claudeWebMessages,
	37	claudeWebSessions,
	38	type ClaudeWebSession,
	39	} from "../db/schema";
	40	import { and, eq } from "drizzle-orm";
	41	import { getRepoPath } from "../git/repository";
	42
	43	const MAX_TURN_MS = 5 * 60_000;
	44
	45	export function claudeWebRoot(): string {
	46	return (
	47	process.env.CLAUDE_WEB_WORKDIR \|\|
	48	"/var/lib/gluecron/claude-web"
	49	).replace(/\/$/, "");
	50	}
	51
	52	export function claudeBinary(): string {
	53	return process.env.CLAUDE_BIN \|\| "claude";
	54	}
	55
	56	// ---------------------------------------------------------------------------
	57	// Spawn seam
	58	// ---------------------------------------------------------------------------
	59
	60	export interface SpawnHandle {
	61	/** Async iterable of utf-8 stdout chunks. */
	62	stdout: AsyncIterable<string>;
	63	/** Resolves to the final exit code + collected stderr after stdout ends. */
	64	done: Promise<{ exitCode: number; stderr: string }>;
	65	/** Kills the underlying subprocess. */
	66	kill: () => void;
	67	}
	68
	69	export type SpawnFn = (
	70	cmd: string[],
	71	opts: { cwd: string; env: Record<string, string> }
	72	) => SpawnHandle;
	73
	74	const defaultSpawn: SpawnFn = (cmd, opts) => {
	75	const proc = Bun.spawn(cmd, {
	76	cwd: opts.cwd,
	77	env: opts.env,
	78	stdin: "ignore",
	79	stdout: "pipe",
	80	stderr: "pipe",
	81	});
	82	const stdout = (async function* () {
	83	const reader = (proc.stdout as ReadableStream<Uint8Array>).getReader();
	84	const dec = new TextDecoder();
	85	try {
	86	while (true) {
	87	const { done, value } = await reader.read();
	88	if (done) break;
	89	if (value && value.length) yield dec.decode(value, { stream: true });
	90	}
	91	const tail = dec.decode();
	92	if (tail) yield tail;
	93	} finally {
	94	reader.releaseLock();
	95	}
	96	})();
	97	const done = (async () => {
	98	const [exitCode, stderr] = await Promise.all([
	99	proc.exited,
	100	new Response(proc.stderr).text(),
	101	]);
	102	return { exitCode, stderr };
	103	})();
	104	return { stdout, done, kill: () => proc.kill() };
	105	};
	106
	107	let _spawn: SpawnFn = defaultSpawn;
	108	export function __setSpawnForTests(fn: SpawnFn \| null): void {
	109	_spawn = fn ?? defaultSpawn;
	110	}
	111
	112	// ---------------------------------------------------------------------------
	113	// Workdir
	114	// ---------------------------------------------------------------------------
	115
	116	export function sessionWorkdir(sessionId: string): string {
	117	return join(claudeWebRoot(), sessionId);
	118	}
	119
	120	/**
	121	* Make sure the session's working dir exists as a fresh clone of the
	122	* repo's bare store at the given branch. Idempotent — if the dir already
	123	* has a `.git` folder, we leave it alone and the operator's existing
	124	* working state is preserved across turns.
	125	*
	126	* Uses `git clone --branch <branch> --depth 1 <bare> <workdir>` so the
	127	* initial materialisation is fast even on huge repos. Operators can
	128	* `git fetch --unshallow` from inside the session if they need history.
	129	*/
	130	export async function ensureWorkdir(
	131	session: ClaudeWebSession,
	132	ownerName: string,
	133	repoName: string,
	134	spawn: SpawnFn = _spawn
	135	): Promise<{ ok: true } \| { ok: false; error: string }> {
	136	const root = claudeWebRoot();
	137	try {
	138	await mkdir(root, { recursive: true });
	139	} catch (err) {
	140	return {
	141	ok: false,
	142	error: `mkdir ${root}: ${err instanceof Error ? err.message : err}`,
	143	};
	144	}
	145	const workdir = session.workdirPath;
	146	// If .git already exists in the workdir, the clone is done.
	147	try {
	148	const s = await stat(join(workdir, ".git"));
	149	if (s.isDirectory()) return { ok: true };
	150	} catch {
	151	/* fall through to clone */
	152	}
	153
	154	const bare = getRepoPath(ownerName, repoName);
	155	const handle = spawn(
	156	["git", "clone", "--branch", session.branch, "--depth", "1", bare, workdir],
	157	{ cwd: root, env: scrubbedEnv() }
	158	);
	159	// Drain stdout (clone is quiet, but we still need to consume).
	160	for await (const _ of handle.stdout) {
	161	/* ignore */
	162	}
	163	const { exitCode, stderr } = await handle.done;
	164	if (exitCode !== 0) {
	165	return { ok: false, error: stderr.trim() \|\| `git clone exit ${exitCode}` };
	166	}
	167	return { ok: true };
	168	}
	169
	170	// ---------------------------------------------------------------------------
	171	// Turn driver
	172	// ---------------------------------------------------------------------------
	173
	174	export interface TurnInput {
	175	session: ClaudeWebSession;
	176	ownerName: string;
	177	repoName: string;
	178	prompt: string;
	179	}
	180
	181	export interface TurnEvent {
	182	/** Streaming text chunk from Claude's stdout. */
	183	chunk?: string;
	184	/** Terminal event — call once, after stdout ends. */
	185	done?: {
	186	exitCode: number;
	187	durationMs: number;
	188	/** Claude CLI session UUID extracted from the run (when present). */
	189	claudeSessionId?: string;
	190	stderr: string;
	191	};
	192	}
	193
	194	/**
	195	* Run a single conversational turn against Claude. Yields stdout chunks
	196	* as they arrive, then a single `done` event with metadata. The caller
	197	* (typically the SSE route) is responsible for serialising events onto
	198	* the wire and persisting the final assistant body.
	199	*
	200	* Honours `MAX_TURN_MS` — if the subprocess hasn't finished by then we
	201	* call kill() and emit done with exitCode=124 (matching `timeout`).
	202	*/
	203	export async function* runTurn(
	204	input: TurnInput,
	205	spawn: SpawnFn = _spawn
	206	): AsyncGenerator<TurnEvent, void, void> {
	207	const start = Date.now();
	208	const cmd = [
	209	claudeBinary(),
	210	"--print",
	211	"--output-format",
	212	"stream-json",
	213	...(input.session.claudeSessionId
	214	? ["--resume", input.session.claudeSessionId]
	215	: []),
	216	input.prompt,
	217	];
	218
	219	const handle = spawn(cmd, {
	220	cwd: input.session.workdirPath,
	221	env: passthroughEnv(),
	222	});
	223
	224	const timer = setTimeout(() => {
	225	try {
	226	handle.kill();
	227	} catch {
	228	/* ignore */
	229	}
	230	}, MAX_TURN_MS);
	231
	232	let sessionId: string \| undefined;
	233	try {
	234	for await (const chunk of handle.stdout) {
	235	// Best-effort parse for the Claude session UUID inside stream-json
	236	// events. The CLI emits a `"session_id":"<uuid>"` key on its init
	237	// event; we just grep the chunk text so format drift doesn't break.
	238	if (!sessionId) {
	239	const m = chunk.match(/"session_id"\s:\s"([0-9a-f-]{32,36})"/i);
	240	if (m) sessionId = m[1];
	241	}
	242	yield { chunk };
	243	}
	244	} finally {
	245	clearTimeout(timer);
	246	}
	247
	248	const { exitCode, stderr } = await handle.done;
	249	yield {
	250	done: {
	251	exitCode,
	252	durationMs: Date.now() - start,
	253	claudeSessionId: sessionId,
	254	stderr,
	255	},
	256	};
	257	}
	258
	259	/**
	260	* Env scrubbing for child subprocesses. Drops Postgres / SMTP / SSH
	261	* creds the Claude binary doesn't need. Keeps ANTHROPIC_API_KEY, PATH,
	262	* HOME, and the CLAUDE_* family.
	263	*/
	264	function passthroughEnv(): Record<string, string> {
	265	const out: Record<string, string> = {};
	266	const passlist = new Set([
	267	"PATH",
	268	"HOME",
	269	"USER",
	270	"LANG",
	271	"LC_ALL",
	272	"TERM",
	273	"ANTHROPIC_API_KEY",
	274	"ANTHROPIC_BASE_URL",
	275	"ANTHROPIC_AUTH_TOKEN",
	276	]);
	277	for (const [k, v] of Object.entries(process.env)) {
	278	if (typeof v !== "string") continue;
	279	if (passlist.has(k) \|\| k.startsWith("CLAUDE_")) out[k] = v;
	280	}
	281	return out;
	282	}
	283
	284	function scrubbedEnv(): Record<string, string> {
	285	// For `git clone` we want even fewer — just PATH + HOME so git can
	286	// find its config and ssh wrappers.
	287	const out: Record<string, string> = {};
	288	for (const k of ["PATH", "HOME", "USER", "LANG"]) {
	289	if (typeof process.env[k] === "string") out[k] = process.env[k] as string;
	290	}
	291	// Disable any interactive prompt — clone should be non-interactive.
	292	out.GIT_TERMINAL_PROMPT = "0";
	293	return out;
	294	}
	295
	296	// ---------------------------------------------------------------------------
	297	// Transcript persistence
	298	// ---------------------------------------------------------------------------
	299
	300	export async function listMessages(sessionId: string) {
	301	return db
	302	.select()
	303	.from(claudeWebMessages)
	304	.where(eq(claudeWebMessages.sessionId, sessionId))
	305	.orderBy(claudeWebMessages.createdAt);
	306	}
	307
	308	export async function appendMessage(input: {
	309	sessionId: string;
	310	role: "user" \| "assistant" \| "system";
	311	body: string;
	312	exitCode?: number;
	313	durationMs?: number;
	314	}): Promise<void> {
	315	await db.insert(claudeWebMessages).values({
	316	sessionId: input.sessionId,
	317	role: input.role,
	318	body: input.body,
	319	exitCode: input.exitCode ?? null,
	320	durationMs: input.durationMs ?? null,
	321	});
	322	}
	323
	324	export async function touchSession(input: {
	325	sessionId: string;
	326	claudeSessionId?: string;
	327	status?: string;
	328	}): Promise<void> {
	329	const set: Record<string, unknown> = { lastActiveAt: new Date() };
	330	if (input.claudeSessionId) set.claudeSessionId = input.claudeSessionId;
	331	if (input.status) set.status = input.status;
	332	await db
	333	.update(claudeWebSessions)
	334	.set(set)
	335	.where(eq(claudeWebSessions.id, input.sessionId));
	336	}
	337
	338	export async function createSession(input: {
	339	repositoryId: string;
	340	ownerUserId: string;
	341	title?: string;
	342	branch?: string;
	343	}): Promise<ClaudeWebSession> {
	344	// Insert first so we have the id for the workdir path.
	345	const [row] = await db
	346	.insert(claudeWebSessions)
	347	.values({
	348	repositoryId: input.repositoryId,
	349	ownerUserId: input.ownerUserId,
	350	title: input.title ?? "New session",
	351	branch: input.branch ?? "main",
	352	workdirPath: "pending",
	353	status: "cold",
	354	})
	355	.returning();
	356	if (!row) throw new Error("createSession: insert returned no row");
	357	const workdir = sessionWorkdir(row.id);
	358	const [updated] = await db
	359	.update(claudeWebSessions)
	360	.set({ workdirPath: workdir })
	361	.where(eq(claudeWebSessions.id, row.id))
	362	.returning();
	363	return updated ?? { ...row, workdirPath: workdir };
	364	}
	365
	366	export async function getSession(
	367	id: string,
	368	ownerUserId?: string
	369	): Promise<ClaudeWebSession \| null> {
	370	const conds = ownerUserId
	371	? and(
	372	eq(claudeWebSessions.id, id),
	373	eq(claudeWebSessions.ownerUserId, ownerUserId)
	374	)
	375	: eq(claudeWebSessions.id, id);
	376	const [row] = await db
	377	.select()
	378	.from(claudeWebSessions)
	379	.where(conds)
	380	.limit(1);
	381	return row ?? null;
	382	}
	383
	384	export async function listSessionsForRepo(
	385	repositoryId: string,
	386	limit = 50
	387	): Promise<ClaudeWebSession[]> {
	388	return db
	389	.select()
	390	.from(claudeWebSessions)
	391	.where(eq(claudeWebSessions.repositoryId, repositoryId))
	392	.limit(limit);
	393	}
	394
	395	export async function deleteSession(id: string): Promise<void> {
	396	await db.delete(claudeWebSessions).where(eq(claudeWebSessions.id, id));
	397	}
	398
	399	/** Test-only access to internals. */
	400	export const __test = {
	401	passthroughEnv,
	402	scrubbedEnv,
	403	MAX_TURN_MS,
	404	};