fix(cache): lane compile-cache writes per pull request and bound them at admission

apps/dispatcher/src/compile-cache-ledger-do.ts

@@ -0,0 +1,31 @@
+// FlareDispatch Dispatcher — the CompileCacheLedger Durable Object.
+//
+// One instance per repository (`idFromName(<owner/name>)`, lowercase). The
+// ledger logic is compile-cache-ledger.ts; this class runs it over the
+// object's storage, one call at a time.
+//
+// Spec: specs/adr/0007-cost-efficient-execution.md § Shared compile cache.
+
+import { DurableObject } from "cloudflare:workers";
+import { type CompileCacheLedgerApi, makeCompileCacheLedger, type ReserveRequest, type ReserveResult } from "./compile-cache-ledger";
+import type { Env } from "./env";
+
+export class CompileCacheLedger extends DurableObject<Env> implements CompileCacheLedgerApi {
+  private readonly ledger = makeCompileCacheLedger(this.ctx.storage);
+
+  async reserve(request: ReserveRequest): Promise<ReserveResult> {
+    return this.ledger.reserve(request);
+  }
+
+  async end(request: { readonly executionId: string; readonly expiresAt: number }): Promise<void> {
+    return this.ledger.end(request);
+  }
+
+  async release(request: { readonly bytes: number; readonly nowSec: number }): Promise<number> {
+    return this.ledger.release(request);
+  }
+
+  async measure(request: { readonly bytes: number; readonly nowSec: number }): Promise<number> {
+    return this.ledger.measure(request);
+  }
+}

apps/dispatcher/src/compile-cache-ledger.test.ts

@@ -0,0 +1,66 @@
+// The compile-cache ledger: the per-repository byte count every PUT reserves
+// against, and the per-execution budget and end marker that stop a token's
+// writes.
+
+import { describe, expect, it } from "vitest";
+import { type LedgerStorage, makeCompileCacheLedger } from "./compile-cache-ledger";
+
+const memoryStorage = (): LedgerStorage & { readonly map: Map<string, unknown> } => {
+  const map = new Map<string, unknown>();
+  return {
+    map,
+    get: async <T>(key: string) => map.get(key) as T | undefined,
+    put: async <T>(key: string, value: T) => {
+      map.set(key, value);
+    },
+    delete: async (key: string) => map.delete(key),
+    list: async <T>(opts: { prefix: string }) =>
+      new Map([...map].filter(([key]) => key.startsWith(opts.prefix)) as [string, T][]),
+  };
+};
+
+const NOW = 1_800_000_000;
+const bounds = { maxRepoBytes: 100, maxExecutionBytes: 60 };
+const reserve = (executionId: string, bytes: number) => ({ executionId, expiresAt: NOW + 3600, bytes, ...bounds });
+
+describe("compile-cache ledger", () => {
+  it("admits writes until the repository bound and counts every admitted byte", async () => {
+    const ledger = makeCompileCacheLedger(memoryStorage());
+    expect(await ledger.reserve(reserve("a", 50))).toEqual({ ok: true, repoBytes: 50 });
+    expect(await ledger.reserve(reserve("b", 50))).toEqual({ ok: true, repoBytes: 100 });
+    expect(await ledger.reserve(reserve("c", 1))).toEqual({ ok: false, reason: "repository", repoBytes: 100 });
+  });
+
+  it("refuses an execution's write past its own budget while other executions still write", async () => {
+    const ledger = makeCompileCacheLedger(memoryStorage());
+    expect(await ledger.reserve(reserve("a", 60))).toEqual({ ok: true, repoBytes: 60 });
+    expect(await ledger.reserve(reserve("a", 1))).toEqual({ ok: false, reason: "execution", repoBytes: 60 });
+    expect(await ledger.reserve(reserve("b", 10))).toEqual({ ok: true, repoBytes: 70 });
+  });
+
+  it("refuses every write of an execution that ended, before or after its first write", async () => {
+    const ledger = makeCompileCacheLedger(memoryStorage());
+    await ledger.reserve(reserve("a", 10));
+    await ledger.end({ executionId: "a", expiresAt: NOW + 3600 });
+    expect(await ledger.reserve(reserve("a", 1))).toEqual({ ok: false, reason: "ended", repoBytes: 10 });
+    await ledger.end({ executionId: "never-wrote", expiresAt: NOW + 3600 });
+    expect(await ledger.reserve(reserve("never-wrote", 1))).toEqual({ ok: false, reason: "ended", repoBytes: 10 });
+  });
+
+  it("subtracts released bytes, never below zero", async () => {
+    const ledger = makeCompileCacheLedger(memoryStorage());
+    await ledger.reserve(reserve("a", 40));
+    expect(await ledger.release({ bytes: 30, nowSec: NOW })).toBe(10);
+    expect(await ledger.release({ bytes: 30, nowSec: NOW })).toBe(0);
+  });
+
+  it("replaces the count with a measured size and forgets executions whose tokens expired", async () => {
+    const storage = memoryStorage();
+    const ledger = makeCompileCacheLedger(storage);
+    await ledger.reserve({ ...reserve("old", 10), expiresAt: NOW - 1 });
+    await ledger.reserve(reserve("live", 10));
+    expect(await ledger.measure({ bytes: 75, nowSec: NOW })).toBe(75);
+    expect([...storage.map.keys()].filter((key) => key.startsWith("x:"))).toEqual(["x:live"]);
+    expect(await ledger.reserve(reserve("live", 25))).toEqual({ ok: true, repoBytes: 100 });
+  });
+});

apps/dispatcher/src/compile-cache-ledger.ts

@@ -0,0 +1,103 @@
+// FlareDispatch Dispatcher — the compile-cache ledger.
+//
+// One ledger per repository (the `CompileCacheLedger` Durable Object,
+// compile-cache-ledger-do.ts, addressed by `idFromName(<owner/name>)`) holds:
+//
+//   repo        the repository's bytes: every admitted PUT adds its length,
+//               the sweep subtracts what it deletes and, after listing the
+//               whole repository, replaces the count with the measured size
+//   x:<exec>    one execution's admitted bytes, whether it ended, and when
+//               its token expires
+//
+// A PUT reserves its length here before it writes, so the repository bound
+// and the per-execution budget hold at admission rather than as of the last
+// sweep. The workflow ends an execution's record when the run finishes, which
+// refuses that token's further writes. The Durable Object runs one call at a
+// time, so two reserves never both pass on the same headroom.
+//
+// An overwritten key counts twice until the next whole-repository sweep; the
+// count errs high, never low.
+//
+// Spec: specs/adr/0007-cost-efficient-execution.md § Shared compile cache.
+
+/** The storage the ledger reads and writes: a Durable Object's `ctx.storage`. */
+export type LedgerStorage = {
+  get<T>(key: string): Promise<T | undefined>;
+  put<T>(key: string, value: T): Promise<void>;
+  delete(key: string): Promise<boolean>;
+  list<T>(options: { prefix: string }): Promise<Map<string, T>>;
+};
+
+type ExecutionRecord = { readonly bytes: number; readonly ended: boolean; readonly expiresAt: number };
+
+const REPO_KEY = "repo";
+const EXECUTION_PREFIX = "x:";
+
+export type ReserveRequest = {
+  readonly executionId: string;
+  /** The token's expiry, epoch seconds: the record is kept until then. */
+  readonly expiresAt: number;
+  readonly bytes: number;
+  readonly maxRepoBytes: number;
+  readonly maxExecutionBytes: number;
+};
+
+/** A reserve's verdict, with the repository's count after it. */
+export type ReserveResult =
+  | { readonly ok: true; readonly repoBytes: number }
+  | { readonly ok: false; readonly reason: "ended" | "execution" | "repository"; readonly repoBytes: number };
+
+/** The ledger's calls, as the route and the sweep make them. */
+export type CompileCacheLedgerApi = {
+  reserve(request: ReserveRequest): Promise<ReserveResult>;
+  end(request: { readonly executionId: string; readonly expiresAt: number }): Promise<void>;
+  release(request: { readonly bytes: number; readonly nowSec: number }): Promise<number>;
+  measure(request: { readonly bytes: number; readonly nowSec: number }): Promise<number>;
+};
+
+/** The ledger over `storage`. */
+export const makeCompileCacheLedger = (storage: LedgerStorage): CompileCacheLedgerApi => {
+  const repoBytes = async (): Promise<number> => (await storage.get<number>(REPO_KEY)) ?? 0;
+  const forgetExpired = async (nowSec: number): Promise<void> => {
+    const records = await storage.list<ExecutionRecord>({ prefix: EXECUTION_PREFIX });
+    for (const [key, record] of records) if (record.expiresAt <= nowSec) await storage.delete(key);
+  };
+  return {
+    reserve: async (request) => {
+      const key = `${EXECUTION_PREFIX}${request.executionId}`;
+      const execution = (await storage.get<ExecutionRecord>(key)) ?? {
+        bytes: 0,
+        ended: false,
+        expiresAt: request.expiresAt,
+      };
+      const repo = await repoBytes();
+      if (execution.ended) return { ok: false, reason: "ended", repoBytes: repo };
+      if (execution.bytes + request.bytes > request.maxExecutionBytes)
+        return { ok: false, reason: "execution", repoBytes: repo };
+      if (repo + request.bytes > request.maxRepoBytes) return { ok: false, reason: "repository", repoBytes: repo };
+      await storage.put(key, { ...execution, bytes: execution.bytes + request.bytes });
+      await storage.put(REPO_KEY, repo + request.bytes);
+      return { ok: true, repoBytes: repo + request.bytes };
+    },
+    end: async (request) => {
+      const key = `${EXECUTION_PREFIX}${request.executionId}`;
+      const execution = await storage.get<ExecutionRecord>(key);
+      await storage.put(key, {
+        bytes: execution?.bytes ?? 0,
+        ended: true,
+        expiresAt: Math.max(execution?.expiresAt ?? 0, request.expiresAt),
+      });
+    },
+    release: async (request) => {
+      const next = Math.max(0, (await repoBytes()) - request.bytes);
+      await storage.put(REPO_KEY, next);
+      await forgetExpired(request.nowSec);
+      return next;
+    },
+    measure: async (request) => {
+      await storage.put(REPO_KEY, request.bytes);
+      await forgetExpired(request.nowSec);
+      return request.bytes;
+    },
+  };
+};

apps/dispatcher/src/compile-cache-token.ts

@@ -1,13 +1,19 @@
 // FlareDispatch Dispatcher — compile-cache tokens.
 //
 // The compile-cache route (routes/compile-cache.ts) admits a request carrying
-// a token minted for one repository and one execution, valid until that
-// execution's ceiling:
+// a token minted for one repository, one execution and one lane, valid until
+// that execution's ceiling:
 //
-//   token = "cc1." <expiresAt> "." base64url(executionId) "." mac
-//   mac   = base64url( HMAC-SHA256( k_cc, "cc1\n" repo "\n" executionId "\n" expiresAt ) )
+//   token = "cc2." <expiresAt> "." base64url(executionId) "." lane "." mac
+//   mac   = base64url( HMAC-SHA256( k_cc, "cc2\n" repo "\n" executionId "\n" lane "\n" expiresAt ) )
 //   k_cc  = HKDF-SHA256( HMAC_SECRET, salt="", info="flare-dispatch/compile-cache/v1" )
 //
+// `lane` (primitives/compile-cache.ts `CompileCacheLane`) decides where the
+// token's writes land and which entries it reads: `base`, `pr-<n>` or
+// `read-only`. The run definition names it from the execution's trigger; a
+// dispatch caller able to name `base` holds `HMAC_SECRET`, the key this MAC
+// derives from, so the lane adds no authority the caller lacks.
+//
 // `repo` is lowercase `owner/name` and is not in the token: the route takes it
 // from the request path, so a token presented under another repository's
 // prefix fails the MAC. `expiresAt` is epoch seconds. The HKDF label keeps
@@ -22,12 +28,14 @@
 import {
   COMPILE_CACHE_ENDPOINT_KEY,
   COMPILE_CACHE_TOKEN_KEY,
+  type CompileCacheLane,
+  isCompileCacheLane,
 } from "@fractalboxdev/flare-dispatch-core/primitives";
 import { deriveSecret } from "./capability-token";
 import type { Env } from "./env";
 
 const HKDF_INFO = "flare-dispatch/compile-cache/v1";
-const VERSION = "cc1";
+const VERSION = "cc2";
 
 const encoder = new TextEncoder();
 
@@ -51,10 +59,11 @@ const fromBase64url = (text: string): string | undefined => {
 /** The execution id charset (log-token.ts `EXECUTION_ID_RE`). */
 const EXECUTION_ID = /^[A-Za-z0-9:_-]{3,128}$/;
 
-/** The token's scope: one repository, one execution, until `expiresAt` (epoch seconds). */
+/** The token's scope: one repository, one execution, one lane, until `expiresAt` (epoch seconds). */
 export type CompileCacheScope = {
   readonly repo: string;
   readonly executionId: string;
+  readonly lane: CompileCacheLane;
   readonly expiresAt: number;
 };
 
@@ -68,7 +77,9 @@ const macKey = async (ikm: string): Promise<CryptoKey> =>
   );
 
 const message = (scope: CompileCacheScope): Uint8Array =>
-  encoder.encode(`${VERSION}\n${scope.repo.toLowerCase()}\n${scope.executionId}\n${scope.expiresAt}`);
+  encoder.encode(
+    `${VERSION}\n${scope.repo.toLowerCase()}\n${scope.executionId}\n${scope.lane}\n${scope.expiresAt}`,
+  );
 
 /** The key material: the Worker's `HMAC_SECRET`; `undefined` when unset (the route then refuses). */
 export const resolveCompileCacheSecret = (env: Pick<Env, "HMAC_SECRET">): string | undefined =>
@@ -77,10 +88,11 @@ export const resolveCompileCacheSecret = (env: Pick<Env, "HMAC_SECRET">): string
 /** Mint the token for `scope`. */
 export const signCompileCacheToken = async (ikm: string, scope: CompileCacheScope): Promise<string> => {
   if (!EXECUTION_ID.test(scope.executionId)) throw new Error("compile-cache token: invalid execution id");
+  if (!isCompileCacheLane(scope.lane)) throw new Error("compile-cache token: invalid lane");
   if (!Number.isSafeInteger(scope.expiresAt) || scope.expiresAt <= 0)
     throw new Error("compile-cache token: invalid expiry");
   const mac = new Uint8Array(await crypto.subtle.sign("HMAC", await macKey(ikm), message(scope)));
-  return `${VERSION}.${scope.expiresAt}.${base64url(encoder.encode(scope.executionId))}.${base64url(mac)}`;
+  return `${VERSION}.${scope.expiresAt}.${base64url(encoder.encode(scope.executionId))}.${scope.lane}.${base64url(mac)}`;
 };
 
 /** Why a presented token was refused. */
@@ -97,14 +109,20 @@ export const verifyCompileCacheToken = async (
   nowSec: number,
 ): Promise<{ readonly ok: true; readonly scope: CompileCacheScope } | { readonly ok: false; readonly reason: CompileCacheTokenRefusal }> => {
   const parts = presented.split(".");
-  if (parts.length !== 4 || parts[0] !== VERSION || !/^[1-9][0-9]{0,15}$/.test(parts[1]!))
+  if (parts.length !== 5 || parts[0] !== VERSION || !/^[1-9][0-9]{0,15}$/.test(parts[1]!))
     return { ok: false, reason: "malformed" };
   const expiresAt = Number(parts[1]);
   const executionId = fromBase64url(parts[2]!);
-  if (executionId === undefined || !EXECUTION_ID.test(executionId) || !/^[A-Za-z0-9_-]{43}$/.test(parts[3]!))
+  const lane = parts[3]!;
+  if (
+    executionId === undefined ||
+    !EXECUTION_ID.test(executionId) ||
+    !isCompileCacheLane(lane) ||
+    !/^[A-Za-z0-9_-]{43}$/.test(parts[4]!)
+  )
     return { ok: false, reason: "malformed" };
-  const mac = Uint8Array.from(atob(parts[3]!.replace(/-/g, "+").replace(/_/g, "/")), (c) => c.charCodeAt(0));
-  const scope = { repo, executionId, expiresAt };
+  const mac = Uint8Array.from(atob(parts[4]!.replace(/-/g, "+").replace(/_/g, "/")), (c) => c.charCodeAt(0));
+  const scope = { repo, executionId, lane, expiresAt };
   const valid = await crypto.subtle.verify("HMAC", await macKey(ikm), mac, message(scope));
   if (!valid) return { ok: false, reason: "invalid" };
   if (nowSec >= expiresAt) return { ok: false, reason: "expired" };
@@ -113,8 +131,8 @@ export const verifyCompileCacheToken = async (
 
 /** What the workflow knows when it decides whether to mint. */
 export type CompileCacheMint = {
-  /** The run's own answer (`RunSpec.compileCache`). */
-  readonly wanted: boolean;
+  /** The run's own answer (`RunSpec.compileCache`): its lane, or `undefined` for no cache. */
+  readonly lane: CompileCacheLane | undefined;
   readonly repo: string;
   readonly executionId: string;
   /** The Workflow instance's creation time, epoch ms: stable across invocations. */
@@ -125,23 +143,34 @@ export type CompileCacheMint = {
   readonly secret: string | undefined;
 };
 
+/** A minted grant: the config overrides that carry it, its lane, and its token's expiry (epoch seconds). */
+export type CompileCacheGrant = {
+  readonly overrides: Record<string, string>;
+  readonly lane: CompileCacheLane;
+  readonly expiresAt: number;
+};
+
 /**
- * The config overrides that hand a run its compile-cache endpoint and token,
- * or `undefined` when the run does not want the cache or the deploy has no
+ * The grant that hands a run its compile-cache endpoint and token, or
+ * `undefined` when the run does not want the cache or the deploy has no
  * public origin or signing secret.
  */
-export const compileCacheOverrides = async (
-  mint: CompileCacheMint,
-): Promise<Record<string, string> | undefined> => {
-  if (!mint.wanted || mint.publicOrigin === undefined || mint.publicOrigin === "" || mint.secret === undefined)
+export const compileCacheOverrides = async (mint: CompileCacheMint): Promise<CompileCacheGrant | undefined> => {
+  if (mint.lane === undefined || mint.publicOrigin === undefined || mint.publicOrigin === "" || mint.secret === undefined)
     return undefined;
+  const expiresAt = Math.floor(mint.createdAtMs / 1000) + Math.ceil(mint.ceilingSec);
   const token = await signCompileCacheToken(mint.secret, {
     repo: mint.repo,
     executionId: mint.executionId,
-    expiresAt: Math.floor(mint.createdAtMs / 1000) + Math.ceil(mint.ceilingSec),
+    lane: mint.lane,
+    expiresAt,
   });
   return {
-    [COMPILE_CACHE_ENDPOINT_KEY]: `${mint.publicOrigin.replace(/\/$/, "")}/v1/compile-cache`,
-    [COMPILE_CACHE_TOKEN_KEY]: token,
+    overrides: {
+      [COMPILE_CACHE_ENDPOINT_KEY]: `${mint.publicOrigin.replace(/\/$/, "")}/v1/compile-cache`,
+      [COMPILE_CACHE_TOKEN_KEY]: token,
+    },
+    lane: mint.lane,
+    expiresAt,
   };
 };

apps/dispatcher/src/env.ts

@@ -335,11 +335,13 @@ export interface Env {
 
   /**
    * Compile-cache bounds (routes/compile-cache.ts): one entry's bytes, one
-   * repository prefix's bytes, and the entry age in days the sweep deletes.
-   * Unset or malformed → 100 MiB, 20 GiB, 14 days.
+   * repository's bytes, one execution's written bytes, and the entry age in
+   * days the sweep deletes. Unset or malformed → 100 MiB, 20 GiB, 10 GiB,
+   * 14 days.
    */
   readonly COMPILE_CACHE_MAX_OBJECT_BYTES?: string;
   readonly COMPILE_CACHE_MAX_REPO_BYTES?: string;
+  readonly COMPILE_CACHE_MAX_EXECUTION_BYTES?: string;
   readonly COMPILE_CACHE_TTL_DAYS?: string;
 
   /**
@@ -431,6 +433,14 @@ export interface Env {
    */
   readonly AGENT_BUDGET?: DurableObjectNamespace<import("./agent-budget-do").AgentBudget>;
 
+  /**
+   * One compile-cache ledger per repository (compile-cache-ledger.ts): the
+   * byte count every compile-cache PUT reserves against, and each
+   * execution's write budget and end marker. Absent → the compile cache
+   * serves reads and refuses every write.
+   */
+  readonly COMPILE_CACHE_LEDGER?: DurableObjectNamespace<import("./compile-cache-ledger-do").CompileCacheLedger>;
+
   /**
    * Optional dedicated secret for the agent model-proxy capability token. Falls
    * back to `HMAC_SECRET` (HKDF label keeps them independent). Neither set → the