fix(flue-review): hydrate from the GitHub tarball instead of JS git

infra/flue-review/.flue/lib/untar.ts

@@ -0,0 +1,162 @@
+// Streaming untar for GitHub source tarballs (post-gunzip). The parser is the
+// trust boundary between an externally-supplied archive and the review
+// workspace: entry paths and symlink targets are validated so no write can
+// land outside the destination directory.
+
+/** The subset of the Workspace surface the untar needs; narrow for testing. */
+export interface UntarTarget {
+	mkdir(path: string, options?: { recursive?: boolean }): Promise<void>;
+	symlink(target: string, linkPath: string): Promise<void>;
+	writeFileBytes(path: string, content: Uint8Array): Promise<void>;
+}
+
+function stripRoot(name: string): string | undefined {
+	const slash = name.indexOf("/");
+	if (slash === -1) return undefined;
+	const rest = name.slice(slash + 1);
+	return rest.length > 0 ? rest : undefined;
+}
+
+function parentOf(path: string): string {
+	return path.slice(0, path.lastIndexOf("/"));
+}
+
+function normalizeSegments(path: string): string[] | undefined {
+	const out: string[] = [];
+	for (const part of path.split("/")) {
+		if (part === "" || part === ".") continue;
+		if (part === "..") {
+			if (out.length === 0) return undefined;
+			out.pop();
+			continue;
+		}
+		out.push(part);
+	}
+	return out;
+}
+
+/**
+ * Untar a ustar stream into `destDir`, stripping the archive's single
+ * top-level directory. Handles regular files, directories, symlinks, GNU
+ * longname ('L') and pax ('x') path overrides. One entry's content is
+ * buffered at a time. Rejects any entry path or symlink target that would
+ * resolve outside `destDir`.
+ */
+export async function untarInto(
+	target: UntarTarget,
+	stream: ReadableStream<Uint8Array>,
+	destDir: string,
+): Promise<{ files: number; bytes: number }> {
+	const decoder = new TextDecoder();
+	let buffer = new Uint8Array(0);
+	let files = 0;
+	let bytes = 0;
+	let pendingLongName: string | undefined;
+	let pendingPaxPath: string | undefined;
+	const dirsMade = new Set<string>();
+
+	const append = (chunk: Uint8Array) => {
+		const next = new Uint8Array(buffer.length + chunk.length);
+		next.set(buffer, 0);
+		next.set(chunk, buffer.length);
+		buffer = next;
+	};
+	const readCString = (view: Uint8Array): string => {
+		const end = view.indexOf(0);
+		return decoder.decode(end === -1 ? view : view.subarray(0, end));
+	};
+	const ensureDir = async (path: string) => {
+		if (dirsMade.has(path)) return;
+		await target.mkdir(path, { recursive: true });
+		dirsMade.add(path);
+	};
+
+	const reader = stream.getReader();
+	let done = false;
+	const need = async (n: number): Promise<boolean> => {
+		while (buffer.length < n && !done) {
+			const r = await reader.read();
+			if (r.done) done = true;
+			else append(r.value);
+		}
+		return buffer.length >= n;
+	};
+
+	while (await need(512)) {
+		const header = buffer.subarray(0, 512);
+		buffer = buffer.subarray(512);
+		// Two consecutive zero blocks terminate the archive.
+		if (header.every((b) => b === 0)) break;
+
+		const rawName = readCString(header.subarray(0, 100));
+		const prefix = readCString(header.subarray(345, 500));
+		const size = parseInt(readCString(header.subarray(124, 136)).trim() || "0", 8);
+		// Mode bytes (100-108) are ignored: the workspace has no chmod and the
+		// reviewer never executes files.
+		const type = String.fromCharCode(header[156] ?? 48);
+		const linkTarget = readCString(header.subarray(157, 257));
+
+		const padded = Math.ceil(size / 512) * 512;
+		if (!(await need(padded)) && size > 0) {
+			throw new Error(`tar truncated: needed ${padded} bytes for entry ${rawName}`);
+		}
+		const content = buffer.subarray(0, size);
+		buffer = buffer.subarray(Math.min(padded, buffer.length));
+
+		if (type === "L") {
+			pendingLongName = readCString(content);
+			continue;
+		}
+		if (type === "x" || type === "g") {
+			// pax records: "<len> key=value\n"
+			const text = decoder.decode(content);
+			for (const line of text.split("\n")) {
+				const eq = line.indexOf("=");
+				if (eq > 0 && line.slice(line.indexOf(" ") + 1, eq) === "path") {
+					pendingPaxPath = line.slice(eq + 1);
+				}
+			}
+			continue;
+		}
+
+		const fullName =
+			pendingPaxPath ?? pendingLongName ?? (prefix ? `${prefix}/${rawName}` : rawName);
+		pendingLongName = undefined;
+		pendingPaxPath = undefined;
+
+		if (fullName.startsWith("/")) {
+			throw new Error(`tar entry path is absolute: ${fullName}`);
+		}
+		const relative = stripRoot(fullName);
+		if (!relative) continue;
+		const segments = normalizeSegments(relative);
+		if (segments === undefined || segments.length === 0) {
+			throw new Error(`tar entry path escapes the destination: ${fullName}`);
+		}
+		const dest = `${destDir}/${segments.join("/")}`;
+
+		if (type === "5") {
+			await ensureDir(dest);
+		} else if (type === "2") {
+			if (linkTarget.startsWith("/")) {
+				throw new Error(`tar symlink target is absolute: ${fullName} -> ${linkTarget}`);
+			}
+			// Resolve the target against the link's directory; it must stay
+			// inside destDir even before the workspace resolves anything.
+			const resolved = normalizeSegments(`${segments.slice(0, -1).join("/")}/${linkTarget}`);
+			if (resolved === undefined) {
+				throw new Error(`tar symlink target escapes the destination: ${fullName} -> ${linkTarget}`);
+			}
+			await ensureDir(parentOf(dest));
+			await target.symlink(linkTarget, dest);
+		} else if (type === "0" || type === "\0" || type === "7") {
+			await ensureDir(parentOf(dest));
+			// Copy out of the rolling buffer: content is a subarray view.
+			await target.writeFileBytes(dest, new Uint8Array(content));
+			files += 1;
+			bytes += size;
+		}
+		// Hardlinks and other exotic types don't occur in GitHub tarballs; skip.
+	}
+	return { files, bytes };
+}

infra/flue-review/.flue/sandboxes/cloudflare-shell.ts

@@ -220,6 +220,52 @@ function buildCodeToolDescription(): string {
 	].join("\n");
 }
 
+// R2 operations log start/end so a stalled call is visible as an unmatched
+// start line. Wrapped once and cached per underlying bucket: Workspace
+// fingerprints construction options per storage, so every construction site
+// must receive the identical instance.
+const instrumentedBuckets = new WeakMap<R2Bucket, R2Bucket>();
+let r2OpSeq = 0;
+function instrumentR2(bucket: R2Bucket): R2Bucket {
+	const cached = instrumentedBuckets.get(bucket);
+	if (cached) return cached;
+	const wrap = (method: string) =>
+		async function (...args: unknown[]) {
+			const id = ++r2OpSeq;
+			const key = typeof args[0] === "string" ? args[0] : "?";
+			const t0 = Date.now();
+			console.log(JSON.stringify({ message: "r2 op start", id, method, key }));
+			try {
+				// oxlint-disable-next-line typescript/no-unsafe-type-assertion
+				const result = await (bucket as unknown as Record<string, Function>)[method](...args);
+				console.log(JSON.stringify({ message: "r2 op end", id, method, key, ms: Date.now() - t0 }));
+				return result;
+			} catch (error) {
+				console.error(
+					JSON.stringify({
+						message: "r2 op failed",
+						id,
+						method,
+						key,
+						ms: Date.now() - t0,
+						error: error instanceof Error ? error.message : String(error),
+					}),
+				);
+				throw error;
+			}
+		};
+	const proxied = new Proxy(bucket, {
+		get(target, prop, receiver) {
+			if (typeof prop === "string" && ["get", "put", "delete", "head", "list"].includes(prop)) {
+				return wrap(prop);
+			}
+			return Reflect.get(target, prop, receiver);
+		},
+	});
+	instrumentedBuckets.set(bucket, proxied);
+	return proxied;
+}
+
 export function getDefaultWorkspace(r2?: R2Bucket, name?: string): Workspace {
 	const { storage } = getCloudflareContext();
 	return new Workspace({
@@ -228,6 +274,6 @@ export function getDefaultWorkspace(r2?: R2Bucket, name?: string): Workspace {
 		// `name` keys R2 objects and namespaces this workspace; required when an
 		// R2 bucket is provided (large files spill under r2://<name>/...).
 		...(name ? { name } : {}),
-		...(r2 ? { r2 } : {}),
+		...(r2 ? { r2: instrumentR2(r2) } : {}),
 	});
 }

infra/flue-review/.flue/workflows/review.ts

@@ -2,7 +2,8 @@
 //
 // Reviews one pull request and returns structured findings plus a verdict. No
 // firecracker container: the PR is hydrated into a durable cf-shell Workspace
-// (DO SQLite + R2 for large files) via JS git, and the agent inspects it with a
+// (DO SQLite + R2 for large files) from the GitHub tarball of the PR head, and
+// the agent inspects it with a
 // Worker-Loader-backed `code` tool. It does NOT post to GitHub: the workflow's
 // trusted Action code posts with a write-scoped installation token, so no
 // secret is ever reachable by the model.
@@ -16,8 +17,6 @@
 // agent initializer and the clone performed in the Action target the exact same
 // DO SQLite + R2 namespace.
 
-import { WorkspaceFileSystem } from "@cloudflare/shell";
-import { createGit } from "@cloudflare/shell/git";
 import {
 	defineAgent,
 	defineWorkflow,
@@ -45,6 +44,7 @@ import {
 	type ReviewStage,
 	type ReviewTerminal,
 } from "../lib/review-watchdog.js";
+import { untarInto } from "../lib/untar.js";
 import { getDefaultWorkspace, getShellSandbox } from "../sandboxes/cloudflare-shell.js";
 import review from "../skills/review/SKILL.md" with { type: "skill" };
 
@@ -158,35 +158,52 @@ function buildPrContext(payload: ReviewPayload, priorReview?: string): string {
 	return lines.join("\n");
 }
 
-// Hydrate the PR into the durable Workspace via JS git (shallow clone of base,
-// then fetch + checkout the PR head -- refs/pull/N/head covers fork PRs). Large
-// objects (the git packfile) spill to R2 under the workspace name. Idempotent:
-// a HYDRATED marker skips re-cloning on workflow re-entry.
+// Hydration stage logs bracket each phase so a hang shows as a start line
+// with no matching end line. R2 operations are instrumented in the shared
+// getDefaultWorkspace.
+function hydrateStep(payload: ReviewPayload, step: string, startedAt: number): void {
+	console.log(
+		JSON.stringify({
+			message: "hydrate step",
+			step,
+			ms: Date.now() - startedAt,
+			attemptId: payload.attemptId,
+			prNumber: payload.prNumber,
+		}),
+	);
+}
+
+// Hydrate the PR into the durable Workspace from the GitHub tarball of the PR
+// head SHA (reachable in the base repo for fork PRs too). No git objects and
+// no pack indexing: pure-JS pack inflation in the DO stops completing once
+// the repo's shallow pack grows past roughly 16MB, so hydration must not
+// depend on git. gzip decompression is the runtime-native DecompressionStream.
+// Idempotent: a HYDRATED marker skips re-fetching on workflow re-entry.
 async function hydrate(env: Env, payload: ReviewPayload): Promise<void> {
+	const t0 = Date.now();
 	const workspace = getDefaultWorkspace(env.REVIEW_WORKSPACE, workspaceName());
-	if (await workspace.exists(HYDRATED)) return;
-
-	const fs = new WorkspaceFileSystem(workspace);
-	const cloneUrl = `https://github.com/${payload.owner}/${payload.repo}.git`;
-	const git = createGit(fs);
-	await git.clone({
-		url: cloneUrl,
-		dir: REPO_DIR,
-		branch: payload.baseRef,
-		singleBranch: true,
-		depth: 1,
-	});
-	const fetched = await git.fetch({
-		ref: `pull/${payload.prNumber}/head`,
-		depth: 1,
-		dir: REPO_DIR,
+	hydrateStep(payload, "workspace created", t0);
+	if (await workspace.exists(HYDRATED)) {
+		hydrateStep(payload, "already hydrated", t0);
+		return;
+	}
+	if (!payload.headSha) throw new Error("hydrate requires the PR head SHA");
+
+	const url = `https://api.github.com/repos/${payload.owner}/${payload.repo}/tarball/${payload.headSha}`;
+	const response = await fetch(url, {
+		headers: { "User-Agent": "emdash-flue-review", Accept: "application/vnd.github+json" },
 	});
-	if (!fetched.fetchHead) throw new Error("PR head fetch did not return a commit");
-	if (payload.headSha && fetched.fetchHead.toLowerCase() !== payload.headSha.toLowerCase()) {
-		throw new Error("PR head changed after the review was requested");
+	if (!response.ok || !response.body) {
+		throw new Error(`tarball fetch failed: ${response.status} ${await response.text()}`);
 	}
-	await git.checkout({ ref: fetched.fetchHead, dir: REPO_DIR, force: true });
+	hydrateStep(payload, "tarball response", t0);
+
+	const tarStream = response.body.pipeThrough(new DecompressionStream("gzip"));
+	const { files, bytes } = await untarInto(workspace, tarStream, REPO_DIR);
+	hydrateStep(payload, `untarred ${files} files ${bytes} bytes`, t0);
+
 	await workspace.writeFile(HYDRATED, new Date().toISOString());
+	hydrateStep(payload, "hydrated", t0);
 }
 
 function logReviewEvent(