JSONbored · JSONbored · Jul 11, 2026
@@ -15,3 +15,6 @@ CREATE TABLE IF NOT EXISTS impact_map_query_cache (
   fetched_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
   PRIMARY KEY (project, repo, query_fingerprint)
 );
+
+CREATE INDEX IF NOT EXISTS impact_map_query_cache_repo_fetched_at_idx
+  ON impact_map_query_cache (project, repo, fetched_at);
@@ -1492,5 +1492,6 @@ export const impactMapQueryCache = sqliteTable(
   },
   (table) => ({
     primary: primaryKey({ columns: [table.project, table.repo, table.queryFingerprint] }),
+    repoFetchedAtIdx: index("impact_map_query_cache_repo_fetched_at_idx").on(table.project, table.repo, table.fetchedAt),
   }),
 );
@@ -65,6 +65,10 @@ function buildSymbolQueryText(file: FileChangedSymbols): string {
 // re-embed within that window, and any longer would risk masking a real index update for no added benefit.
 const IMPACT_MAP_QUERY_CACHE_MAX_AGE_MS = 30 * 60 * 1000;
 
+function impactMapQueryCacheCutoffIso(): string {
+  return new Date(Date.now() - IMPACT_MAP_QUERY_CACHE_MAX_AGE_MS).toISOString();
+}
+
 /** One query's cache key: every input that affects retrieveContextWithMetrics' result. topK/minScore/reranker
  *  are constants for this module's own calls, but are still hashed (not assumed) so this function stays
  *  correct if a future caller ever varies them. excludePaths is sorted before hashing so argument order never
@@ -88,13 +92,16 @@ async function getCachedImpactMapQuery(
 ): Promise<RagRetrievalResult | null> {
   try {
     const row = await storage
-      .prepare("SELECT context, metrics_json AS metricsJson, fetched_at AS fetchedAt FROM impact_map_query_cache WHERE project = ? AND repo = ? AND query_fingerprint = ?")
+      .prepare("SELECT metrics_json AS metricsJson, fetched_at AS fetchedAt FROM impact_map_query_cache WHERE project = ? AND repo = ? AND query_fingerprint = ?")
       .bind(project, repo, fingerprint)
-      .first<{ context: string; metricsJson: string; fetchedAt: string }>();
+      .first<{ metricsJson: string; fetchedAt: string }>();
     if (!row) return null;
     const ageMs = Date.now() - Date.parse(row.fetchedAt);
-    if (!Number.isFinite(ageMs) || ageMs >= IMPACT_MAP_QUERY_CACHE_MAX_AGE_MS) return null;
-    return { context: row.context, metrics: JSON.parse(row.metricsJson) as RagRetrievalResult["metrics"] };
+    if (!Number.isFinite(ageMs) || ageMs >= IMPACT_MAP_QUERY_CACHE_MAX_AGE_MS) {
+      await storage.prepare("DELETE FROM impact_map_query_cache WHERE project = ? AND repo = ? AND query_fingerprint = ?").bind(project, repo, fingerprint).run();
+      return null;
+    }
+    return { context: "", metrics: JSON.parse(row.metricsJson) as RagRetrievalResult["metrics"] };
   } catch {
     return null; // fail-safe: a storage error degrades to "no cache", never blocks the query
   }
@@ -108,14 +115,15 @@ async function putCachedImpactMapQuery(
   result: RagRetrievalResult,
 ): Promise<void> {
   try {
+    await storage.prepare("DELETE FROM impact_map_query_cache WHERE project = ? AND repo = ? AND fetched_at < ?").bind(project, repo, impactMapQueryCacheCutoffIso()).run();
     await storage
       .prepare(
         `INSERT INTO impact_map_query_cache (project, repo, query_fingerprint, context, metrics_json, fetched_at)
          VALUES (?, ?, ?, ?, ?, ?)
          ON CONFLICT(project, repo, query_fingerprint) DO UPDATE SET
            context = excluded.context, metrics_json = excluded.metrics_json, fetched_at = excluded.fetched_at`,
       )
-      .bind(project, repo, fingerprint, result.context, JSON.stringify(result.metrics), nowIso())
+      .bind(project, repo, fingerprint, "", JSON.stringify(result.metrics), nowIso())
       .run();
   } catch {
     // fail-safe: a write failure only means this ONE result isn't cached -- never blocks the review

@@ -60,6 +60,15 @@ function cachingStorageStub(count: number, fetchedAtOverride?: string): { storag
         all: async () =>
           /SELECT id, text/i.test(sql) ? { results: args.map((id) => ({ id: String(id), text: `body for ${String(id)}` })) } : { results: [] },
         run: async () => {
+          if (/DELETE FROM impact_map_query_cache/i.test(sql)) {
+            const [project, repo, third] = args as [string, string, string];
+            for (const [key, row] of rows) {
+              const [rowProject, rowRepo, rowFingerprint] = key.split("|");
+              const matchesExactRow = /query_fingerprint = \?/i.test(sql) && rowProject === project && rowRepo === repo && rowFingerprint === third;
+              const matchesExpiredRepoRow = /fetched_at < \?/i.test(sql) && rowProject === project && rowRepo === repo && row.fetchedAt < third;
+              if (matchesExactRow || matchesExpiredRepoRow) rows.delete(key);
+            }
+          }
           if (/INSERT INTO impact_map_query_cache/i.test(sql)) {
             const [project, repo, fingerprint, context, metricsJson, fetchedAt] = args as [string, string, string, string, string, string];
             rows.set(`${project}|${repo}|${fingerprint}`, { context, metricsJson, fetchedAt: fetchedAtOverride ?? fetchedAt });
@@ -387,6 +396,45 @@ describe("computeImpactMap", () => {
     expect(queryCalls).toBe(2);
   });
 
+  it("REGRESSION: expired impact-map cache rows are evicted instead of accumulating indefinitely", async () => {
+    const staleIso = new Date(Date.now() - 60 * 60 * 1000).toISOString();
+    const freshIso = new Date().toISOString();
+    const { storage, rows } = cachingStorageStub(5);
+    rows.set("acme|widgets|expired-one", { context: "old context", metricsJson: "{}", fetchedAt: staleIso });
+    rows.set("acme|widgets|expired-two", { context: "old context", metricsJson: "{}", fetchedAt: staleIso });
+    rows.set("other|widgets|expired-other-project", { context: "old context", metricsJson: "{}", fetchedAt: staleIso });
+    rows.set("acme|widgets|fresh-one", { context: "fresh context", metricsJson: "{}", fetchedAt: freshIso });
+    const infra: RagInfra = {
+      storage,
+      vector: vectorStub([{ id: "src/review/caller.ts::0", score: 0.9, metadata: { path: "src/review/caller.ts" } }]),
+      inference: ai1024,
+    };
+
+    await computeImpactMap(testEnv(), [{ path: "src/review/impact-map.ts", symbols: ["computeImpactMap"] }], { infra, project: "acme", repo: "widgets" });
+
+    const acmeWidgetRows = [...rows.keys()].filter((key) => key.startsWith("acme|widgets|"));
+    expect(acmeWidgetRows).toHaveLength(2);
+    expect(acmeWidgetRows).toContain("acme|widgets|fresh-one");
+    expect(acmeWidgetRows).not.toContain("acme|widgets|expired-one");
+    expect(acmeWidgetRows).not.toContain("acme|widgets|expired-two");
+    expect(rows.has("other|widgets|expired-other-project")).toBe(true);
+  });
+
+  it("REGRESSION: impact-map cache rows retain only metrics, not the retrieved context body", async () => {
+    const { storage, rows } = cachingStorageStub(5);
+    const infra: RagInfra = {
+      storage,
+      vector: vectorStub([{ id: "src/review/caller.ts::0", score: 0.9, metadata: { path: "src/review/caller.ts" } }]),
+      inference: ai1024,
+    };
+
+    await computeImpactMap(testEnv(), [{ path: "src/review/impact-map.ts", symbols: ["computeImpactMap"] }], { infra, project: "acme", repo: "widgets" });
+
+    expect([...rows.values()]).toHaveLength(1);
+    expect([...rows.values()][0]?.context).toBe("");
+    expect([...rows.values()][0]?.metricsJson).toContain("src/review/caller.ts");
+  });
+
   describe("cache hit/miss telemetry (#4448)", () => {
     afterEach(() => resetMetrics());
 

@@ -1,6 +1,6 @@
 /* eslint-disable */
 // Generated by Wrangler by running `wrangler types` (hash: 31e2a5aa781186e51d85d560f8a9f25b)
-// Runtime types generated with workerd@1.20260701.1 2026-05-28 nodejs_compat
+// Runtime types generated with workerd@1.20260708.1 2026-05-28 nodejs_compat
 interface __BaseEnv_Env {
 	DB: D1Database;
 	JOBS: Queue;
@@ -619,7 +619,7 @@ declare abstract class DurableObjectNamespace<T extends Rpc.DurableObjectBranded
     getByName(name: string, options?: DurableObjectNamespaceGetDurableObjectOptions): DurableObjectStub<T>;
     jurisdiction(jurisdiction: DurableObjectJurisdiction): DurableObjectNamespace<T>;
 }
-type DurableObjectJurisdiction = "eu" | "fedramp" | "fedramp-high";
+type DurableObjectJurisdiction = "eu" | "fedramp" | "fedramp-high" | "us";
 interface DurableObjectNamespaceNewUniqueIdOptions {
     jurisdiction?: DurableObjectJurisdiction;
 }
@@ -12314,6 +12314,13 @@ interface ForwardableEmailMessage extends EmailMessage {
      * @returns A promise that resolves when the email message is replied.
      */
     reply(message: EmailMessage): Promise<EmailSendResult>;
+    /**
+     * Reply to the sender of this email message with a message built from the given
+     * fields. Threading headers (In-Reply-To/References) are set automatically.
+     * @param builder The reply message contents.
+     * @returns A promise that resolves when the email message is replied.
+     */
+    reply(builder: EmailReplyMessageBuilder): Promise<EmailSendResult>;
 }
 /** A file attachment for an email message */
 type EmailAttachment = {
@@ -12334,23 +12341,46 @@ interface EmailAddress {
     name: string;
     email: string;
 }
+/**
+ * Recipient fields for `SendEmail.send()`. At least one of `to`, `cc`, or
+ * `bcc` must be provided.
+ */
+type EmailDestinations = {
+    to?: string | EmailAddress | (string | EmailAddress)[];
+    cc?: string | EmailAddress | (string | EmailAddress)[];
+    bcc?: string | EmailAddress | (string | EmailAddress)[];
+} & ({
+    to: string | EmailAddress | (string | EmailAddress)[];
+} | {
+    cc: string | EmailAddress | (string | EmailAddress)[];
+} | {
+    bcc: string | EmailAddress | (string | EmailAddress)[];
+});
+/**
+ * Fields shared by all composed emails (no recipients). Used directly by
+ * `ForwardableEmailMessage.reply()`, which always replies to the original
+ * sender, and extended by `EmailMessageBuilder` for `SendEmail.send()`.
+ */
+interface EmailReplyMessageBuilder {
+    from: string | EmailAddress;
+    subject: string;
+    replyTo?: string | EmailAddress;
+    headers?: Record<string, string>;
+    text?: string;
+    html?: string;
+    attachments?: EmailAttachment[];
+}
+/**
+ * Fields for composing an email without constructing raw MIME, for
+ * `SendEmail.send()`. Requires at least one of `to`, `cc`, or `bcc`.
+ */
+type EmailMessageBuilder = EmailReplyMessageBuilder & EmailDestinations;
 /**
  * A binding that allows a Worker to send email messages.
  */
 interface SendEmail {
     send(message: EmailMessage): Promise<EmailSendResult>;
-    send(builder: {
-        from: string | EmailAddress;
-        to: string | EmailAddress | (string | EmailAddress)[];
-        subject: string;
-        replyTo?: string | EmailAddress;
-        cc?: string | EmailAddress | (string | EmailAddress)[];
-        bcc?: string | EmailAddress | (string | EmailAddress)[];
-        headers?: Record<string, string>;
-        text?: string;
-        html?: string;
-        attachments?: EmailAttachment[];
-    }): Promise<EmailSendResult>;
+    send(builder: EmailMessageBuilder): Promise<EmailSendResult>;
 }
 declare abstract class EmailEvent extends ExtendableEvent {
     readonly message: ForwardableEmailMessage;
@@ -13172,14 +13202,19 @@ declare namespace CloudflareWorkersModule {
     export type WorkflowDurationLabel = 'second' | 'minute' | 'hour' | 'day' | 'week' | 'month' | 'year';
     export type WorkflowSleepDuration = `${number} ${WorkflowDurationLabel}${'s' | ''}` | number;
     export type WorkflowDelayDuration = WorkflowSleepDuration;
+    export type WorkflowDynamicDelayContext = {
+        ctx: WorkflowStepContext<WorkflowDelayFunction>;
+        error: Error;
+    };
+    export type WorkflowDelayFunction = (input: WorkflowDynamicDelayContext) => WorkflowDelayDuration | Promise<WorkflowDelayDuration>;
     export type WorkflowTimeoutDuration = WorkflowSleepDuration;
     export type WorkflowRetentionDuration = WorkflowSleepDuration;
     export type WorkflowBackoff = 'constant' | 'linear' | 'exponential';
     export type WorkflowStepSensitivity = 'output';
     export type WorkflowStepConfig = {
         retries?: {
             limit: number;
-            delay: WorkflowDelayDuration | number;
+            delay: WorkflowDelayDuration | number | WorkflowDelayFunction;
             backoff?: WorkflowBackoff;
         };
         timeout?: WorkflowTimeoutDuration | number;
@@ -13205,13 +13240,22 @@ declare namespace CloudflareWorkersModule {
         type: string;
         sensitive?: WorkflowStepSensitivity;
     };
-    export type WorkflowStepContext = {
+    export type WorkflowStepContext<Delay = WorkflowDelayDuration | number> = {
         step: {
             name: string;
             count: number;
         };
         attempt: number;
-        config: WorkflowStepConfig;
+        config: {
+            retries?: {
+                limit: number;
+                backoff?: WorkflowBackoff;
+            } & (Delay extends WorkflowDelayFunction ? {} : {
+                delay: WorkflowDelayDuration | number;
+            });
+            timeout?: WorkflowTimeoutDuration | number;
+            sensitive?: WorkflowStepSensitivity;
+        };
     };
     export type WorkflowRollbackContext<T = unknown> = {
         ctx: WorkflowStepContext;
@@ -13227,7 +13271,9 @@ declare namespace CloudflareWorkersModule {
     };
     export abstract class WorkflowStep {
         do<T extends Rpc.Serializable<T>>(name: string, callback: (ctx: WorkflowStepContext) => Promise<T>, rollbackOptions?: WorkflowStepRollbackOptions<T>): Promise<T>;
-        do<T extends Rpc.Serializable<T>>(name: string, config: WorkflowStepConfig, callback: (ctx: WorkflowStepContext) => Promise<T>, rollbackOptions?: WorkflowStepRollbackOptions<T>): Promise<T>;
+        do<T extends Rpc.Serializable<T>, const C extends WorkflowStepConfig>(name: string, config: C, callback: (ctx: WorkflowStepContext<C['retries'] extends {
+            delay: infer D;
+        } ? D : WorkflowDelayDuration | number>) => Promise<T>, rollbackOptions?: WorkflowStepRollbackOptions<T>): Promise<T>;
         sleep: (name: string, duration: WorkflowSleepDuration) => Promise<void>;
         sleepUntil: (name: string, timestamp: Date | number) => Promise<void>;
         waitForEvent<T extends Rpc.Serializable<T>>(name: string, options: {
@@ -14137,6 +14183,7 @@ declare namespace TailStream {
         readonly dispatchNamespace?: string;
         readonly entrypoint?: string;
         readonly executionModel: string;
+        readonly durableObjectId?: string;
         readonly scriptName?: string;
         readonly scriptTags?: string[];
         readonly scriptVersion?: ScriptVersion;
@@ -14691,6 +14738,13 @@ interface WorkflowError {
     code?: number;
     message: string;
 }
+interface WorkflowInstanceTerminateOptions {
+    /**
+     * If true, run registered rollback handlers before terminating the instance.
+     * Only steps that registered rollback handlers are rolled back.
+     */
+    rollback?: boolean;
+}
 interface WorkflowInstanceRestartOptions {
     /**
      * Restart from a specific step. If omitted, the instance restarts from the beginning.
@@ -14724,8 +14778,9 @@ declare abstract class WorkflowInstance {
     public resume(): Promise<void>;
     /**
      * Terminate the instance. If it is errored, terminated or complete, an error will be thrown.
+     * @param options Options for termination, including whether registered rollback handlers should run.
      */
-    public terminate(): Promise<void>;
+    public terminate(options?: WorkflowInstanceTerminateOptions): Promise<void>;
     /**
      * Restart the instance. Optionally restart from a specific step, preserving
      * cached results for all steps before it.