fix(runs): contain state Worker background tasks

deploy/cloudflare-memory/backgroundTasks.test.ts

@@ -0,0 +1,22 @@
+import { describe, expect, it, vi } from "vitest";
+import { assertNoPendingBackgroundTasks, holdBackgroundTask } from "./backgroundTasks.ts";
+
+describe("memory Worker pending task guard", () => {
+  it("fails on a fixture promise that is still pending when the test ends", async () => {
+    let release!: () => void;
+    const fixture = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    const waitUntil = vi.fn<(task: Promise<unknown>) => void>();
+
+    holdBackgroundTask({ waitUntil }, "fixture pending promise", fixture);
+
+    expect(waitUntil).toHaveBeenCalledOnce();
+    expect(() => assertNoPendingBackgroundTasks()).toThrow(/fixture pending promise/);
+
+    release();
+    await fixture;
+    await Promise.resolve();
+    expect(() => assertNoPendingBackgroundTasks()).not.toThrow();
+  });
+});

deploy/cloudflare-memory/backgroundTasks.ts

@@ -0,0 +1,25 @@
+type WaitUntilContext = Pick<DurableObjectState, "waitUntil">;
+
+const TASKS = Symbol.for("switchboard.memory-worker.pending-background-tasks");
+type PendingTasks = Map<Promise<unknown>, string>;
+
+const pendingTasks = (): PendingTasks => {
+  const root = globalThis as typeof globalThis & { [TASKS]?: PendingTasks };
+  return (root[TASKS] ??= new Map());
+};
+
+/** Keep deliberately detached Worker work in the runtime's request lifetime.
+ * The process-wide label set is also the test seam: a case cannot finish while
+ * work it started is still able to contend with the next case. */
+export function holdBackgroundTask(context: WaitUntilContext, label: string, task: Promise<unknown>): void {
+  const pending = pendingTasks();
+  const held = task.finally(() => pending.delete(held));
+  pending.set(held, label);
+  context.waitUntil(held);
+}
+
+export function assertNoPendingBackgroundTasks(): void {
+  const labels = [...pendingTasks().values()];
+  if (labels.length === 0) return;
+  throw new Error(`test ended with ${labels.length} pending background task(s): ${labels.join(", ")}`);
+}

deploy/cloudflare-memory/runLedger.test.ts

@@ -4,6 +4,7 @@ import type { RunRecord } from "../../src/core/runRecord.ts";
 import { FRICTION_CATEGORIES } from "../../src/core/runFriction.ts";
 import { LEASE_MS } from "../../src/core/runLedger/types.ts";
 import type { CoordinatorInstance, CoordinatorUnit } from "../../src/core/coordinator/contract.ts";
+import { assertNoPendingBackgroundTasks } from "./backgroundTasks.ts";
 import type { RunHistoryDO, SessionLogDO } from "./worker.ts";
 
 // Feature: docs/reference/specs/run-history.md items 28–34 — the live-run ledger on the
@@ -1429,6 +1430,14 @@ describe("the plane's admission stage — /plane/admit, reservations, the seal's
     await post("/runs/claim", claimBody(key, "r1", t));
     const q = (await admit(key, t, "two")).data.id as string;
     const pushes: { url: string; auth: string | null }[] = [];
+    let startedPush!: () => void;
+    let finishPush!: () => void;
+    const pushStarted = new Promise<void>((resolve) => {
+      startedPush = resolve;
+    });
+    const pushCanFinish = new Promise<void>((resolve) => {
+      finishPush = resolve;
+    });
     await runInDurableObject(env.RUNS.get(env.RUNS.idFromName(key)), async (inst: RunHistoryDO) => {
       // A BOT binding whose push dead-ends (the container down, an older bot
       // without the route): the fetch answers 404 and delivers nothing.
@@ -1438,14 +1447,21 @@ describe("the plane's admission stage — /plane/admit, reservations, the seal's
         BOT: {
           fetch: async (url: string, init: { headers: Record<string, string> }) => {
             pushes.push({ url: String(url), auth: init.headers.authorization ?? null });
+            startedPush();
+            await pushCanFinish;
             return new Response("not found", { status: 404 });
           },
         },
       };
     });
-    // The seal walks the queue and pushes the admit; the push fails.
-    await post("/runs/finish", { storeKey: key, runId: "r1", gen: "g1", record: record("r1", t) });
-    await new Promise((r) => setTimeout(r, 10)); // the push is fire-and-forget
+    // The seal walks the queue and pushes the admit. Its response stays
+    // independent, while waitUntil and the test guard own the unfinished I/O.
+    const finishing = post("/runs/finish", { storeKey: key, runId: "r1", gen: "g1", record: record("r1", t) });
+    await pushStarted;
+    expect(() => assertNoPendingBackgroundTasks()).toThrow(/plane effect push \(admit:/);
+    finishPush();
+    await finishing;
+    await vi.waitFor(() => expect(() => assertNoPendingBackgroundTasks()).not.toThrow());
     expect(pushes).toEqual([{ url: "https://bot/plane/effects", auth: "Bearer test-token" }]);
     // Nothing was acked: the offer stands and rides the next heartbeat answer.
     await runInDurableObject(env.RUNS.get(env.RUNS.idFromName(key)), async (inst: RunHistoryDO) => {
@@ -1953,15 +1969,24 @@ describe("the plane's checkpoint steers and the provider condition — the heart
     const key = storeKey();
     await post("/runs/claim", claimBody(key, "r5", "slack:C20:5.0"));
     await post("/plane/park", { storeKey: key, runId: "r5", provider: "anthropic" });
+    let pushed!: () => void;
+    const pushStarted = new Promise<void>((resolve) => {
+      pushed = resolve;
+    });
     await runInDurableObject(env.RUNS.get(env.RUNS.idFromName(key)), async (inst: RunHistoryDO) => {
       const withBot = inst as unknown as { env: Record<string, unknown> };
       withBot.env = {
         ...withBot.env,
-        BOT: { fetch: async () => new Response("down", { status: 503 }) },
+        BOT: {
+          fetch: async () => {
+            pushed();
+            return new Response("down", { status: 503 });
+          },
+        },
       };
     });
     await post("/plane/level", { storeKey: key, name: "provider", provider: "anthropic", side: "up" });
-    await new Promise((resolve) => setTimeout(resolve, 10));
+    await pushStarted;
     expect(await inbox(key, "r5")).toHaveLength(1);
     const beat = await post("/runs/heartbeat", { storeKey: key, runId: "r5", gen: "g1", leaseMs: LEASE_MS });
     expect(beat.data.effects).toMatchObject([{ id: "steer:r5:1", kind: "steer", seq: 1 }]);

deploy/cloudflare-memory/testSetup.ts

@@ -1,6 +1,14 @@
+import { afterEach } from "vitest";
+import { assertNoPendingBackgroundTasks } from "./backgroundTasks.ts";
+
 // Routine Worker logs are not test output. With console interception disabled
 // they would flood the runner; keep warnings and errors visible, while tests
 // that need a logger inject or spy on one explicitly.
 console.log = () => {};
 console.info = () => {};
 console.debug = () => {};
+
+// Every intentionally detached Worker operation must be registered through
+// holdBackgroundTask. Fail its own case rather than letting workerd charge its
+// unfinished work to a later Durable Object request in the shared isolate.
+afterEach(() => assertNoPendingBackgroundTasks());

deploy/cloudflare-memory/vitest.config.ts

@@ -32,6 +32,7 @@ export default defineConfig({
     // not correctness, was the only failure shape.
     fileParallelism: false,
     include: [
+      "backgroundTasks.test.ts",
       "worker.test.ts",
       "schedules.test.ts",
       "runs.test.ts",

deploy/cloudflare-memory/worker.ts

@@ -78,6 +78,7 @@ import {
   selectReclaim,
 } from "../../src/core/runLedger/decisions.ts";
 import { intakeReceiptRetentionMs, minutesToMs, PLANE } from "../../src/core/budgets.ts";
+import { holdBackgroundTask } from "./backgroundTasks.ts";
 import {
   causeOfClose,
   causeOfReclaim,
@@ -2387,21 +2388,24 @@ export class RunHistoryDO extends DurableObject<Env> {
 
   /** The transport's push (record 0064, "Where it lives"): committed effects
    *  are pushed to the bot Worker over the service binding, which forwards to
-   *  the container. Fire and forget — a push that fails is not retried by a
-   *  timer; the effect rides the next heartbeat or reclaim-sweep answer. */
+   *  the container. The response does not wait for this best-effort push, but
+   *  the actor does: waitUntil keeps its I/O inside this request's lifetime so
+   *  it cannot contend with an unrelated request after the caller moves on. */
   private pushPlaneEffects(effects: PlaneEffect[]): void {
     if (effects.length === 0) return;
     const bot = this.env.BOT;
     if (!bot) return;
-    void bot
-      .fetch("https://bot/plane/effects", {
-        method: "POST",
-        headers: {
-          "content-type": "application/json",
-          authorization: `Bearer ${this.env.MEMORY_TOKEN ?? ""}`,
-        },
-        body: JSON.stringify({ effects }),
-      })
+    const delivery = Promise.resolve()
+      .then(() =>
+        bot.fetch("https://bot/plane/effects", {
+          method: "POST",
+          headers: {
+            "content-type": "application/json",
+            authorization: `Bearer ${this.env.MEMORY_TOKEN ?? ""}`,
+          },
+          body: JSON.stringify({ effects }),
+        }),
+      )
       .then((r) => {
         if (!r.ok)
           console.warn(
@@ -2413,6 +2417,7 @@ export class RunHistoryDO extends DurableObject<Env> {
           `[plane/push] ${effects.length} effect(s) not delivered: ${err instanceof Error ? err.message : String(err)} — they ride the next heartbeat`,
         );
       });
+    holdBackgroundTask(this.ctx, `plane effect push (${effects.map((effect) => effect.id).join(", ")})`, delivery);
   }
 
   /** The unacknowledged effects, oldest first, at most `PLANE_EFFECTS_PER_ANSWER`