packages/forkhorse-core/src/receipts.ts

1import { claimForAgent, markEntrant } from "./race-entrants";
2import { usageSources } from "./usage";
3import { RpcTarget } from "cloudflare:workers";
4import { Snapshot } from "../../forkhorse-runner/src/snapshot";
5import { loadChecks, MissingChecksError } from "../../forkhorse-runner/src/checks";
6import { receiptJson, requirePassingReceipt, sha256 } from "../../forkhorse-runner/src/runner";
7import type { AuthorizedReview, ReviewAuthority, RunRequest, StoredReceipt } from "../../forkhorse-runner/src/types";
8import { HttpError } from "./http";
9import type { Env, Project } from "./types";
10import type { ClaimReleaseAudit, Fork, Packet, ReceiptSummary, Review } from "./workflow-types";
11
12export const settings = (project: Project) => project.settings ?? { requireReceipt: false };
13export const publicReview = ({ receiptGeneration: _generation, ...review }: Review) => ({
14  ...review, receipt: review.receipt ?? null, returned: review.returned ?? null,
15  usageAtReview: usageSources(review.usageAtReview),
16});
17export type ExecuteReceipt = (
18  request: RunRequest, authority: ReceiptAuthority, generation: string,
19) => Promise<StoredReceipt>;
20const same = (review: Review | undefined, run: AuthorizedReview) =>
21  review && !review.stale && (!review.promoted || run.trigger === "review-created") && review.id === run.reviewId &&
22  review.sha === run.commit && review.repoName === run.forkRepoName &&
23  review.receiptGeneration === run.generation && review.receipt?.trigger === run.trigger;
24
25/** Immutable content-addressed records. Chunks stay well below the 128 KiB value limit,
26 * including worst-case UTF-8 expansion. The manifest is committed with all chunks. */
27export async function storeReceipt(storage: DurableObjectStorage, record: StoredReceipt): Promise<void> {
28  if (!/^[a-f0-9]{64}$/.test(record.digest) || await sha256(receiptJson(record.receipt)) !== record.digest)
29    throw new Error("Invalid receipt digest");
30  const serialized = JSON.stringify(record);
31  if (serialized.length > 8 * 1024 * 1024) throw new Error("Receipt too large");
32  const chunks = Array.from({ length: Math.ceil(serialized.length / 16384) }, (_, i) =>
33    serialized.slice(i * 16384, (i + 1) * 16384));
34  await storage.transaction(async (txn) => {
35    const key = `receipt:${record.digest}`;
36    if (await txn.get(key)) return;
37    for (let i = 0; i < chunks.length; i++) await txn.put(`${key}:${i}`, chunks[i]!);
38    await txn.put(key, { chunks: chunks.length });
39  });
40}
41export async function readReceipt(storage: DurableObjectStorage, digest: string | null): Promise<StoredReceipt | null> {
42  if (!digest || !/^[a-f0-9]{64}$/.test(digest)) return null;
43  const key = `receipt:${digest}`;
44  const manifest = await storage.get<{ chunks: number }>(key);
45  if (!manifest) return null;
46  if (!Number.isInteger(manifest.chunks) || manifest.chunks < 1 || manifest.chunks > 512)
47    throw new Error("Invalid receipt manifest");
48  const chunks: string[] = [];
49  for (let i = 0; i < manifest.chunks; i++) {
50    const chunk = await storage.get<string>(`${key}:${i}`);
51    if (chunk === undefined) throw new Error("Incomplete receipt");
52    chunks.push(chunk);
53  }
54  const record = JSON.parse(chunks.join("")) as StoredReceipt;
55  if (record.digest !== digest || await sha256(receiptJson(record.receipt)) !== digest)
56    throw new Error("Invalid stored receipt");
57  return record;
58}
59
60/** A capability for ONE reserved execution, sent only to the same-script runner.
61 * These callbacks deliberately do not acquire Supervisor's workflow mutex: promote
62 * owns that mutex while awaiting the runner. Attachment uses a storage transaction. */
63export class ReceiptAuthority extends RpcTarget implements ReviewAuthority {
64  #storage: DurableObjectStorage;
65  #run: AuthorizedReview;
66  constructor(storage: DurableObjectStorage, run: AuthorizedReview) {
67    super(); this.#storage = storage; this.#run = structuredClone(run);
68  }
69  async authorize(request: RunRequest): Promise<AuthorizedReview> {
70    if (request.reviewId !== this.#run.reviewId || request.trigger !== this.#run.trigger ||
71        !same(await this.#storage.get<Review>(`review:${request.reviewId}`), this.#run))
72      throw new Error("Stale receipt execution");
73    return structuredClone(this.#run);
74  }
75  async storeReceipt(record: StoredReceipt): Promise<void> {
76    const r = record.receipt, run = this.#run;
77    if (r.projectId !== run.projectId || r.reviewId !== run.reviewId || r.commit !== run.commit ||
78        r.generation !== run.generation || r.trigger !== run.trigger || r.forkRepoName !== run.forkRepoName)
79      throw new Error("Receipt outside capability");
80    await storeReceipt(this.#storage, record);
81  }
82  async setReviewReceiptDigest(run: AuthorizedReview, digest: string): Promise<boolean> {
83    if (Object.keys(this.#run).some(key => run[key as keyof AuthorizedReview] !== this.#run[key as keyof AuthorizedReview])) return false;
84    const record = await readReceipt(this.#storage, digest);
85    if (!record || record.receipt.generation !== run.generation || record.receipt.commit !== run.commit ||
86        record.receipt.reviewId !== run.reviewId || record.receipt.projectId !== run.projectId ||
87        record.receipt.forkRepoName !== run.forkRepoName || record.receipt.trigger !== run.trigger) return false;
88    return this.#storage.transaction(async (txn) => {
89      const review = await txn.get<Review>(`review:${run.reviewId}`);
90      if (!same(review, run) || !review || review.receiptDigest !== null) return false;
91      const claim = await claimForAgent(txn, review.packet, review.agent);
92      // A promoted advisory run may finish after its claim was atomically released,
93      // including after a new claim is made. Only its exact promotion audit restores
94      // that historical binding; same() still fences stale commits and generations.
95      const release = review.promoted && run.trigger === "review-created"
96        ? await txn.get<ClaimReleaseAudit>(`claimRelease:${review.id}`) : undefined;
97      const promotedBinding = release?.releasedBy === "promote" && release.review === review.id &&
98        release.packet === review.packet && release.agent === review.agent && release.repoName === review.repoName;
99      if (!promotedBinding && (!claim || claim.repoName !== review.repoName || claim.agent !== review.agent)) return false;
100      const checks = record.receipt.checks;
101      const status: ReceiptSummary["status"] = !checks.length ? "infrastructure-error"
102        : checks.some(c => c.outcome === "infrastructure-error") ? "infrastructure-error"
103        : checks.some(c => c.outcome === "timed-out") ? "timed-out"
104        : checks.some(c => c.outcome === "failed") ? "failed" : "passed";
105      const finishedAt = new Date().toISOString();
106      review.receiptDigest = digest;
107      review.receipt = {
108        ...review.receipt!, status, digest, finishedAt,
109        durationMs: Math.max(0, Date.parse(finishedAt) - Date.parse(review.receipt!.startedAt)),
110        checks: checks.map(c => ({ name: c.check.name, tier: c.check.tier, runtime: c.runtime, message: c.message, outcome: c.outcome, durationMs: c.durationMs })),
111      };
112      await txn.put(`review:${review.id}`, review);
113      if (status === "passed") await markEntrant(txn, review, "receiptPassedAt", finishedAt);
114      return true;
115    });
116  }
117}
118
119/** Invoked only after project/caller/claim/fork/head authorization in Workflow. */
120export class Receipts {
121  constructor(private storage: DurableObjectStorage, private env: Env,
122    private execute: ExecuteReceipt, private background: (job: Promise<void>) => void) {}
123
124  async reserve(project: Project, review: Review, fork: Fork, trigger: RunRequest["trigger"]): Promise<AuthorizedReview> {
125    if (fork.repoName === project.repoName || fork.repoName !== review.repoName || fork.agent !== review.agent ||
126        await this.storage.get(`repo:${fork.repoName}`) !== review.agent)
127      throw new Error("Receipt requires an authorized fork");
128    const run: AuthorizedReview = {
129      reviewId: review.id, trigger, projectId: project.id, forkRepoName: fork.repoName,
130      forkRemote: fork.remote, commit: review.sha, generation: crypto.randomUUID(),
131    };
132    review.receiptGeneration = run.generation;
133    review.receiptDigest = null;
134    review.receipt = {
135      status: "running", trigger, digest: null, startedAt: new Date().toISOString(),
136      finishedAt: null, durationMs: null, checks: [],
137    };
138    if ((await this.storage.get<Packet>(`packet:${review.packet}`))?.mode === "race") {
139      await this.storage.transaction(async txn => {
140        await txn.put(`review:${review.id}`, review);
141        if (trigger === "review-created")
142          await markEntrant(txn, review, "reviewRequestedAt", review.createdAt);
143      });
144    } else await this.storage.put(`review:${review.id}`, review);
145    return run;
146  }
147  start(run: AuthorizedReview, advisory = false): void {
148    this.background(this.run(run, advisory).then(() => {}));
149  }
150  async run(run: AuthorizedReview, advisory = false): Promise<StoredReceipt | null> {
151    try {
152      const authority = new ReceiptAuthority(this.storage, run);
153      await authority.authorize(run);
154      using repo = await this.env.ARTIFACTS.get(run.forkRepoName);
155      const checks = await loadChecks(repo, run.commit);
156      await this.storage.transaction(async txn => {
157        const review = await txn.get<Review>(`review:${run.reviewId}`);
158        if (!same(review, run) || !review) throw new Error("Stale receipt execution");
159        review.receipt!.checks = checks.map(c => ({ name: c.name, tier: c.tier, outcome: "running", durationMs: null }));
160        await txn.put(`review:${review.id}`, review);
161      });
162      return await this.execute({ reviewId: run.reviewId, trigger: run.trigger }, authority, run.generation);
163    } catch (error) {
164      // Missing checks/images, startup and cleanup errors have no publishable receipt.
165      // Never expose provider exceptions, tokens, or commands in the state projection.
166      await this.storage.transaction(async txn => {
167        const review = await txn.get<Review>(`review:${run.reviewId}`);
168        if (!same(review, run) || !review || review.receiptDigest) return;
169        if (advisory && error instanceof MissingChecksError) {
170          review.receipt = null;
171          review.receiptGeneration = undefined;
172          await txn.put(`review:${review.id}`, review);
173          return;
174        }
175        const finishedAt = new Date().toISOString();
176        review.receipt = { ...review.receipt!, status: "infrastructure-error", finishedAt,
177          durationMs: Math.max(0, Date.parse(finishedAt) - Date.parse(review.receipt!.startedAt)),
178          checks: review.receipt!.checks.map(c => ({ ...c, outcome: "infrastructure-error" })),
179        };
180        await txn.put(`review:${review.id}`, review);
181      });
182      return null;
183    }
184  }
185  async requirePassing(project: Project, review: Review): Promise<void> {
186    try {
187      if (!settings(project).requireReceipt) return;
188      using repo = await this.env.ARTIFACTS.get(review.repoName);
189      const checks = await loadChecks(repo, review.sha);
190      const moduleDigests: Record<string, string> = {};
191      for (const check of checks) if (check.tier === "quick") {
192        const snapshot = new Snapshot(repo, review.sha);
193        try {
194          const source = await snapshot.read(check.module);
195          if (source === null) throw new Error("Quick module missing");
196          moduleDigests[check.module] = `module:sha256:${await sha256(source)}`;
197        } finally { snapshot.close(); }
198      }
199      await requirePassingReceipt(await readReceipt(this.storage, review.receiptDigest), {
200        projectId: project.id, reviewId: review.id, forkRepoName: review.repoName, commit: review.sha,
201        generation: review.receiptGeneration ?? "", digest: review.receiptDigest,
202        checks, moduleDigests, requireReceipt: true,
203      });
204    } catch { throw new HttpError(409, "A current passing check receipt is required."); }
205  }
206}
207
208/** A terminated execution is never silently resumed or presented as passing. */
209export async function recoverInterruptedReceipts(storage: DurableObjectStorage): Promise<void> {
210  await storage.transaction(async txn => {
211    for (const review of (await txn.list<Review>({ prefix: "review:" })).values()) {
212      if (review.receipt?.status !== "running") continue;
213      const finishedAt = new Date().toISOString();
214      review.receiptGeneration = undefined;
215      review.receiptDigest = null;
216      review.receipt = { ...review.receipt, status: "infrastructure-error", digest: null, finishedAt,
217        durationMs: Math.max(0, Date.parse(finishedAt) - Date.parse(review.receipt.startedAt)),
218        checks: review.receipt.checks.map(check => ({ ...check, outcome: "infrastructure-error" })),
219      };
220      await txn.put(`review:${review.id}`, review);
221    }
222  });
223}
224