packages/forkhorse-runner/src/runner.ts

1import { loadChecks } from "./checks";
2import { runQuickCheck } from "./quick";
3import { remoteUrl } from "./git";
4import type { AuthorizedReview, Check, FullCheck, CheckReceipt, ContainerPort, ForkRepository, Receipt, RunnerPorts, RunRequest, StoredReceipt } from "./types";
5
6export const LOG_BYTES_PER_STREAM = 32_768;
7export const CHECK_TIMEOUT_MS = 120_000;
8export const TOKEN_TTL_SECONDS = 180;
9const WORKSPACE = "/workspace/repo";
10const encoder = new TextEncoder();
11
12export async function sha256(text: string): Promise<string> {
13  const hash = await crypto.subtle.digest("SHA-256", encoder.encode(text));
14  return [...new Uint8Array(hash)].map((byte) => byte.toString(16).padStart(2, "0")).join("");
15}
16
17/** Reconstruct a fixed-order representation, independent of RPC object key ordering. */
18export function receiptJson(receipt: Receipt): string {
19  return JSON.stringify({
20    schema: receipt.schema, evidence: receipt.evidence, projectId: receipt.projectId,
21    reviewId: receipt.reviewId, forkRepoName: receipt.forkRepoName, commit: receipt.commit,
22    generation: receipt.generation, trigger: receipt.trigger,
23    checks: receipt.checks.map((r) => ({
24      commit: r.commit,
25      check: r.check.tier === "quick"
26        ? { name: r.check.name, tier: r.check.tier, module: r.check.module }
27        : { name: r.check.name, tier: r.check.tier, image: r.check.image, run: r.check.run },
28      // Omit new fields for old receipts so their immutable digests remain valid.
29      ...(r.runtime !== undefined && { runtime: r.runtime }),
30      ...(r.message !== undefined && { message: r.message }),
31      imageDigest: r.imageDigest, exitStatus: r.exitStatus, outcome: r.outcome,
32      startedAt: r.startedAt, finishedAt: r.finishedAt, durationMs: r.durationMs,
33      logs: { stdout: r.logs.stdout, stderr: r.logs.stderr, truncated: r.logs.truncated },
34      logDigest: r.logDigest,
35    })),
36  });
37}
38
39async function readBounded(stream: ReadableStream | null): Promise<{ text: string; truncated: boolean }> {
40  if (!stream) return { text: "", truncated: false };
41  const reader = stream.getReader();
42  const bytes = new Uint8Array(LOG_BYTES_PER_STREAM);
43  let used = 0;
44  let truncated = false;
45  try {
46    while (true) {
47      const { done, value } = await reader.read();
48      if (done) break;
49      if (!(value instanceof Uint8Array)) throw new Error("Invalid log stream");
50      const keep = Math.min(value.byteLength, bytes.length - used);
51      bytes.set(value.subarray(0, keep), used);
52      used += keep;
53      truncated ||= keep < value.byteLength;
54      // Drain without retaining the tail so a full pipe cannot deadlock the process.
55    }
56  } finally {
57    reader.releaseLock();
58  }
59  return { text: new TextDecoder().decode(bytes.subarray(0, used)), truncated };
60}
61
62async function command(container: ContainerPort, argv: string[], signal: AbortSignal,
63  options: ContainerExecOptions = {}): Promise<{ exitCode: number; logs: CheckReceipt["logs"] }> {
64  signal.throwIfAborted();
65  const process = await container.exec(argv, { ...options, signal });
66  signal.throwIfAborted();
67  const [stdout, stderr, exitCode] = await Promise.all([
68    readBounded(process.stdout), readBounded(process.stderr), process.exitCode,
69  ]);
70  signal.throwIfAborted();
71  if (!Number.isInteger(exitCode) || exitCode < 0 || exitCode > 255) throw new Error("Invalid exit status");
72  return { exitCode, logs: { stdout: stdout.text, stderr: stderr.text, truncated: stdout.truncated || stderr.truncated } };
73}
74
75async function runCheck(ports: RunnerPorts, repo: ForkRepository, review: AuthorizedReview,
76  check: FullCheck, image: string): Promise<CheckReceipt> {
77  const container = ports.container;
78  if (!container) throw new Error("Container binding required");
79  const started = Date.now();
80  const controller = new AbortController();
81  let token: ArtifactsCreateTokenResult | undefined;
82  let timer: ReturnType<typeof setTimeout> | undefined;
83  let result: Awaited<ReturnType<typeof command>> | undefined;
84  let outcome: CheckReceipt["outcome"] = "infrastructure-error";
85  try {
86    // Mint before starting the container/deadline, so an in-flight mint cannot outlive cleanup.
87    // Authorization already succeeded; slow control-plane I/O cannot leave a billable container.
88    const minted = await repo.createToken("read", TOKEN_TTL_SECONDS);
89    token = minted;
90    if (minted.scope !== "read" || !minted.plaintext || /[\r\n\0]/.test(minted.plaintext)) throw new Error("Invalid read token");
91    const deadline = new Promise<never>((_resolve, reject) => {
92      timer = setTimeout(() => {
93        controller.abort();
94        reject(new Error("Check deadline exceeded"));
95      }, CHECK_TIMEOUT_MS);
96    });
97    const execute = async () => {
98      await container.interceptOutboundHttps(remoteUrl(review.forkRemote).hostname, ports.gitGateway(review.forkRemote));
99      controller.signal.throwIfAborted();
100      container.start({ image, enableInternet: false, entrypoint: ["sleep", "infinity"] });
101      await container.setInactivityTimeout(CHECK_TIMEOUT_MS);
102      controller.signal.throwIfAborted();
103      const env = {
104        GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: "/dev/null", GIT_TERMINAL_PROMPT: "0",
105        GIT_SSL_CAINFO: "/etc/cloudflare/certs/cloudflare-containers-ca.crt",
106        GIT_CONFIG_COUNT: "3", GIT_CONFIG_KEY_0: "http.extraHeader",
107        GIT_CONFIG_VALUE_0: `Authorization: Bearer ${minted.plaintext}`,
108        GIT_CONFIG_KEY_1: "http.followRedirects", GIT_CONFIG_VALUE_1: "false",
109        GIT_CONFIG_KEY_2: "http.userAgent", GIT_CONFIG_VALUE_2: "git/2.47.0",
110      };
111      const prepare = async (argv: string[], cwd?: string) => {
112        const output = await command(container, argv, controller.signal, { env, cwd });
113        if (output.exitCode !== 0) throw new Error("Checkout failed");
114        return output;
115      };
116      await prepare(["git", "init", WORKSPACE]);
117      await prepare(["git", "fetch", "--no-tags", "--depth=1", "--", review.forkRemote, review.commit], WORKSPACE);
118      await prepare(["git", "-c", "core.hooksPath=/dev/null", "checkout", "--detach", review.commit], WORKSPACE);
119      const head = await prepare(["git", "rev-parse", "HEAD"], WORKSPACE);
120      if (head.logs.truncated || head.logs.stdout.trim() !== review.commit) throw new Error("Checkout mismatch");
121      // Credentials were per-command env, never git config or a clone URL. Revoke before untrusted code.
122      if (!await repo.revokeToken(minted.id)) throw new Error("Token revocation unconfirmed");
123      token = undefined;
124      controller.signal.throwIfAborted();
125      return command(container, ["timeout", "--kill-after=5", "110", "sh", "-c", check.run],
126        controller.signal, { cwd: WORKSPACE });
127    };
128    result = await Promise.race([execute(), deadline]);
129    outcome = result.exitCode === 0 ? "passed" : "failed";
130  } catch {
131    // Never persist provider/clone exception text: it can contain credentials or command env.
132    outcome = controller.signal.aborted ? "timed-out" : "infrastructure-error";
133  } finally {
134    if (timer !== undefined) clearTimeout(timer);
135    controller.abort();
136  }
137  // Always attempt both cleanup actions. Cleanup failure prevents publication of any receipt.
138  const cleanup = await Promise.allSettled([
139    container.destroy(),
140    token ? repo.revokeToken(token.id).then((revoked) => {
141      if (!revoked) throw new Error("Token revocation unconfirmed");
142    }) : Promise.resolve(),
143  ]);
144  if (cleanup.some((entry) => entry.status === "rejected")) throw new Error("Runner cleanup failed");
145  const finished = Date.now();
146  const logs = result?.logs ?? { stdout: "", stderr: "", truncated: false };
147  return {
148    commit: review.commit, check, runtime: "container", imageDigest: image.slice(image.lastIndexOf("sha256:")),
149    exitStatus: result?.exitCode ?? null, outcome,
150    startedAt: new Date(started).toISOString(), finishedAt: new Date(finished).toISOString(),
151    durationMs: Math.max(0, finished - started), logs, logDigest: await sha256(JSON.stringify(logs)),
152  };
153}
154
155/** Shared by the Durable Object adapter and the explicitly mocked offline acceptance suite. */
156export async function runReviewChecks(request: RunRequest, ports: RunnerPorts): Promise<StoredReceipt> {
157  if (!request.reviewId || !["review-created", "before-promote"].includes(request.trigger)) throw new Error("Invalid run request");
158  const review = await ports.authority.authorize(request);
159  if (review.reviewId !== request.reviewId || review.trigger !== request.trigger ||
160      !/^[a-f0-9]{40}$/.test(review.commit) || !review.projectId || !review.forkRepoName || !review.generation) {
161    throw new Error("Invalid authorization result");
162  }
163  remoteUrl(review.forkRemote);
164  if (ports.container?.running) throw new Error("Runner requires a fresh container");
165  using repo = await ports.getFork(review.forkRepoName);
166  const checks = await loadChecks(repo, review.commit);
167  const receipts: CheckReceipt[] = [];
168  // Execution order is quick-first even if the manifest starts with a full check.
169  const ordered = [...checks.filter(c => c.tier === "quick"), ...checks.filter(c => c.tier === "full")];
170  let quickFailed = false;
171  let fullImages: Map<string, string> | undefined;
172  for (const check of ordered) {
173    if (check.tier === "quick") {
174      const receipt = await runQuickCheck(ports.quick, repo, review, check);
175      receipts.push(receipt);
176      quickFailed ||= receipt.outcome !== "passed";
177    } else if (quickFailed) {
178      const now = new Date().toISOString();
179      const logs = { stdout: "", stderr: "", truncated: false };
180      receipts.push({ commit: review.commit, check, runtime: "container", imageDigest: "", exitStatus: null,
181        outcome: "failed", startedAt: now, finishedAt: now, durationMs: null,
182        message: "Skipped: a quick check failed.", logs, logDigest: await sha256(JSON.stringify(logs)) });
183    } else {
184      if (!fullImages) {
185        fullImages = new Map();
186        // Preserve full-tier preflight: validate every image before any container starts.
187        for (const full of checks) if (full.tier === "full") {
188          const images = ports.container?.images;
189          const image = images && Object.hasOwn(images, full.image) ? images[full.image] : undefined;
190          if (!image || !/@sha256:[a-f0-9]{64}$/.test(image)) throw new Error("Check image must resolve to a configured digest");
191          fullImages.set(full.name, image);
192        }
193      }
194      const image = fullImages.get(check.name)!;
195      receipts.push(await runCheck(ports, repo, review, check, image));
196    }
197  }
198  const hasQuick = checks.some(c => c.tier === "quick");
199  const hasFull = checks.some(c => c.tier === "full");
200  const evidence: Receipt["evidence"] = hasQuick && ports.quick?.evidence !== "dynamic-worker" ? "mocked-worker-loader"
201    : hasFull && ports.evidence !== "container" ? ports.evidence
202    : hasQuick ? hasFull ? "mixed" : "dynamic-worker" : ports.evidence;
203  const receipt: Receipt = {
204    schema: 1, evidence, projectId: review.projectId, reviewId: review.reviewId,
205    forkRepoName: review.forkRepoName, commit: review.commit, generation: review.generation,
206    trigger: review.trigger, checks: receipts,
207  };
208  const record = { digest: await sha256(receiptJson(receipt)), receipt };
209  await ports.authority.storeReceipt(record);
210  if (!await ports.authority.setReviewReceiptDigest(review, record.digest)) throw new Error("Review changed during checks");
211  return record;
212}
213
214/** Approval gate helper. FH-005 must call with current server-owned review/check expectations. */
215export async function requirePassingReceipt(record: StoredReceipt | null, expected: {
216  projectId: string; reviewId: string; forkRepoName: string; commit: string; generation: string;
217  checks: Check[]; digest: string | null; requireReceipt?: boolean;
218  /** Digests independently read by core from the exact reviewed commit. */
219  moduleDigests?: Record<string, string>;
220}): Promise<void> {
221  if (expected.requireReceipt === false) return;
222  if (!record || record.receipt.schema !== 1 || !["container", "dynamic-worker", "mixed"].includes(record.receipt.evidence) ||
223      record.digest !== expected.digest || await sha256(receiptJson(record.receipt)) !== record.digest ||
224      ["projectId", "reviewId", "forkRepoName", "commit", "generation"].some((key) =>
225        record.receipt[key as keyof Receipt] !== expected[key as keyof typeof expected]) ||
226      !expected.checks.length || record.receipt.checks.length !== expected.checks.length ||
227      record.receipt.checks.some((r, i) => {
228        const checks = [...expected.checks.filter(c => c.tier === "quick"), ...expected.checks.filter(c => c.tier === "full")];
229        const check = checks[i];
230        if (r.commit !== expected.commit || r.outcome !== "passed" || r.exitStatus !== 0 ||
231            !check || r.check.name !== check.name || r.check.tier !== check.tier) return true;
232        if (check.tier === "quick") return r.check.tier !== "quick" || r.check.module !== check.module ||
233          r.runtime !== "dynamic-worker" || !["dynamic-worker", "mixed"].includes(record.receipt.evidence) ||
234          !/^module:sha256:[a-f0-9]{64}$/.test(r.imageDigest) || r.imageDigest !== expected.moduleDigests?.[check.module];
235        return r.check.tier !== "full" || r.check.image !== check.image || r.check.run !== check.run ||
236          (r.runtime !== undefined && r.runtime !== "container") || !["container", "mixed"].includes(record.receipt.evidence);
237      })) {
238    throw new Error("A current passing check receipt is required");
239  }
240}
241