feat(runtime-cf): pin a container's transport per repo, so rpc can be proved on one

apps/dispatcher/src/workflow.ts

@@ -444,6 +444,24 @@ export class RunWorkflow extends WorkflowEntrypoint<Env> {
       }
     }
 
+    // Per-repo container transport — the canary knob for `rpc`.
+    //
+    // The SDK's streaming file APIs exist only on its `rpc` client (`http` is
+    // the "route-based compatibility" path), which is why the R2 dependency
+    // cache misses on every run: its restore hands `writeFile` a stream, the SDK
+    // routes any stream to `writeFileStream`, and that raises off `rpc`.
+    // `SANDBOX_TRANSPORT=rpc` as a Worker var would fix it for every repo at
+    // once; this lets one prove it first.
+    //
+    // A value the SDK does not recognise is ignored by it with a warning, so a
+    // typo degrades to the default rather than breaking the run.
+    // Container path only — the substrate owns its own transport, so a run
+    // there should not pay a KV read to be told about one it cannot use.
+    const sandboxTransport =
+      substrateFacade === undefined
+        ? await this.env.CONFIG_KV?.get(`sandbox.transport:${payload.github.repo}`)
+        : undefined;
+
     const runtime = makeCFRuntimeLive({
       db,
       bucket: this.env.RUNS_STORAGE,
@@ -542,6 +560,11 @@ export class RunWorkflow extends WorkflowEntrypoint<Env> {
         ? { aiGatewayAuthToken: this.env.AI_GATEWAY_AUTH_TOKEN }
         : {}),
       sandboxPreviewHostname: this.env.SANDBOX_PREVIEW_HOSTNAME,
+      ...(sandboxTransport === "rpc" ||
+      sandboxTransport === "websocket" ||
+      sandboxTransport === "http"
+        ? { sandboxTransport }
+        : {}),
       ...(publicOrigin !== undefined ? { publicOrigin } : {}),
       ...(logsBaseUrl !== undefined ? { logsViewerBase: logsBaseUrl } : {}),
       // Wire the live OIDC signing Layer when both the JWK + issuer URL are

packages/runtime-cf/src/runtime.ts

@@ -172,6 +172,13 @@ export type CFRuntimeLiveOptions = {
    * Non-browser runs never touch the surface.
    */
   readonly sandboxPreviewHostname?: string;
+  /**
+   * Pin this execution's containers to a transport instead of the Worker-wide
+   * `SANDBOX_TRANSPORT` default — see `makeSandboxCloudflareLive`'s `transport`
+   * option for why the choice is per-execution rather than per-Worker.
+   * Container path only; the substrate facade owns its own transport.
+   */
+  readonly sandboxTransport?: "http" | "websocket" | "rpc";
   /**
    * The dispatcher's own public origin (e.g.
    * `https://<worker>.<account>.workers.dev`) — prefixed onto the
@@ -315,6 +322,7 @@ export const makeCFRuntimeLive = (opts: CFRuntimeLiveOptions): Layer.Layer<RunCo
           cloneAuth,
           opts.sandboxPreviewHostname,
           opts.logsViewerBase,
+          opts.sandboxTransport,
         );
   const stepRunner = makeStepRunnerCloudflare(opts.workflowStep, opts.executionId);
   const checks = makeChecksGithubLive(opts.checks);

packages/runtime-cf/src/sandbox-cf.test.ts

@@ -65,11 +65,20 @@ const makeFakeBox = (opts: {
           name: undefined,
         })),
     ),
+    // The SDK's per-container transport pin. Recorded rather than stubbed away:
+    // which containers got which transport is the whole assertion for the
+    // canary knob.
+    setTransport: vi.fn(async (t: string) => {
+      transportCalls.push(t);
+    }),
     _getLogs: getLogs,
     _waitForPort: waitForPort,
   };
 };
 
+/** Every `setTransport` the Layer issued, in order. */
+const transportCalls: string[] = [];
+
 // A stand-in for the SDK's `SessionTerminatedError` (thrown when a command's
 // shell exits). `vi.hoisted` so it exists before the hoisted `vi.mock` factory
 // runs and so the tests can construct instances the Layer will `instanceof`.
@@ -314,6 +323,63 @@ describe("makeSandboxCloudflareLive — container routing", () => {
     }),
   );
 
+  it.effect("no transport option leaves the container on the Worker-wide default", () =>
+    Effect.gen(function* () {
+      currentBox = makeFakeBox({ proc: null });
+      transportCalls.length = 0;
+      yield* Effect.flatMap(SandboxTag, (s) => s.acquire({})).pipe(
+        Effect.provide(makeSandboxCloudflareLive(ns, makeBucket().bucket, "route-2")),
+      );
+      // Untouched: the default is whatever `SANDBOX_TRANSPORT` says, and a
+      // Layer that pinned it unasked would silently change every consumer.
+      expect(transportCalls).toEqual([]);
+    }),
+  );
+
+  it.effect("a transport option pins EVERY container the execution acquires", () =>
+    Effect.gen(function* () {
+      currentBox = makeFakeBox({ proc: null });
+      transportCalls.length = 0;
+      yield* Effect.flatMap(SandboxTag, (s) =>
+        Effect.all([s.acquire({}), s.acquire({ key: "features" })]),
+      ).pipe(
+        Effect.provide(
+          makeSandboxCloudflareLive(ns, makeBucket().bucket, "route-3", undefined, undefined, undefined, "rpc"),
+        ),
+      );
+      // Per CONTAINER, not per Layer: a keyed container is a different DO and
+      // carries its own transport, so pinning the first would leave the stage
+      // containers on the compatibility client — which is where the streaming
+      // file APIs do not exist.
+      expect(transportCalls).toEqual(["rpc", "rpc"]);
+    }),
+  );
+
+  it.effect("re-acquiring a container does not re-pin it", () =>
+    Effect.gen(function* () {
+      currentBox = makeFakeBox({ proc: null });
+      transportCalls.length = 0;
+      yield* Effect.flatMap(SandboxTag, (s) =>
+        // Three acquires, two containers. `acquire` is also how a caller derives
+        // an id without provisioning — `ensureWorkspace` on a rebuild, and the
+        // stage reaper naming the container it is about to destroy. Each of
+        // those would otherwise wake a Durable Object to re-assert a setting it
+        // already has.
+        Effect.all([
+          s.acquire({}),
+          s.acquire({}),
+          s.acquire({ key: "features" }),
+          s.acquire({ key: "features" }),
+        ]),
+      ).pipe(
+        Effect.provide(
+          makeSandboxCloudflareLive(ns, makeBucket().bucket, "route-4", undefined, undefined, undefined, "rpc"),
+        ),
+      );
+      expect(transportCalls).toEqual(["rpc", "rpc"]);
+    }),
+  );
+
   it.effect("exec routes by the handle it is given, not by the execution", () =>
     Effect.gen(function* () {
       currentBox = makeFakeBox({ proc: null });

packages/runtime-cf/src/sandbox-cf.ts

@@ -308,6 +308,31 @@ export const makeSandboxCloudflareLive = (
    * historical message. Built by the dispatcher (it owns the token secret).
    */
   logsViewerBase?: string,
+  /**
+   * Pin every container this execution acquires to a transport, rather than
+   * taking the Worker-wide `SANDBOX_TRANSPORT` default.
+   *
+   * WHY THIS EXISTS, and why it is per-execution. The SDK's streaming file APIs
+   * live only on the `rpc` client — its own comment calls `rpc` the "primary
+   * container-control client" and `http`/`websocket` the "route-based
+   * compatibility client". `writeFileStream` is a bare `throw` off `rpc`, which
+   * is why the R2 dependency cache has missed on every run since it was added:
+   * the restore hands `writeFile` a `ReadableStream`, the SDK routes any stream
+   * to `writeFileStream`, and it raises.
+   *
+   * `SANDBOX_TRANSPORT=rpc` as a Worker var would fix that in one line and
+   * change the control path for EVERY repo this dispatcher serves at once.
+   * This threads the same choice per execution instead, so one consumer can
+   * prove it before the rest follow. That is the whole difference between a
+   * rollout and a flip.
+   *
+   * Note the precedence the SDK applies on cold start: a transport written to
+   * the DO's storage WINS over the env-derived default. `setTransport` persists,
+   * so a container pinned here stays pinned for its lifetime regardless of what
+   * the var says — which is what makes this safe to scope, and what would make
+   * it sticky if it were ever pointed at the wrong value.
+   */
+  transport?: "http" | "websocket" | "rpc",
 ): Layer.Layer<SandboxTag> => {
   // The Durable Object / sandbox id. `getSandbox` routes the DO by this id AND
   // the SDK embeds it in the `exposePort` preview URL's DNS label, which must
@@ -337,6 +362,9 @@ export const makeSandboxCloudflareLive = (
   // in line with them.
   const boxFor = (container?: Container): Sandbox => getSandbox(ns, container?.id ?? sandboxId);
 
+  /** Container ids whose transport this Layer has already pinned — see `acquire`. */
+  const pinned = new Set<string>();
+
 
   // `exec` log keys are unique within a run: the first exec is `exec.ndjson`
   // (the name the plan's acceptance pins), subsequent execs `exec-2.ndjson`, …
@@ -501,12 +529,37 @@ export const makeSandboxCloudflareLive = (
     // dispatcher's end-of-run teardown knows the execution's own id and cannot
     // know what a run named.
     acquire: (opts) =>
-      Effect.succeed({
-        id:
+      Effect.gen(function* () {
+        const id =
           opts.key === undefined
             ? sandboxId
-            : previewSafeSandboxId(`${executionId}:${opts.key}`),
-      } satisfies Container),
+            : previewSafeSandboxId(`${executionId}:${opts.key}`);
+        // Pin the container's transport before anything runs in it, when the
+        // caller asked for one. See `transport` in this Layer's options for why
+        // this is per-container rather than a Worker-wide env var.
+        //
+        // ONCE PER CONTAINER, not once per acquire. `acquire` is also how a
+        // caller derives an id without provisioning anything — `ensureWorkspace`
+        // re-acquires on a rebuild, and `offload-test`'s stage reaper acquires
+        // purely to name the container it is about to destroy. Without this memo
+        // each of those would wake a Durable Object to re-assert a setting it
+        // already has.
+        //
+        // Best-effort: a container that stays on the default transport works,
+        // it just cannot stream a file. Failing acquisition over a transport
+        // preference would trade a slow cache for a dead run.
+        if (transport !== undefined && !pinned.has(id)) {
+          pinned.add(id);
+          yield* Effect.promise(async () => {
+            try {
+              await getSandbox(ns, id).setTransport(transport);
+            } catch (cause) {
+              console.warn(`setTransport(${transport}) failed for ${id} — staying on the default: ${String(cause)}`);
+            }
+          });
+        }
+        return { id } satisfies Container;
+      }),
 
     // Best-effort and idempotent — destroying a container that was never
     // provisioned, or is already gone, is a success. A teardown failure must

runs/README.md

@@ -263,6 +263,37 @@ Staged mode is webhook-only: a dispatch that passes `command` skips the config
 read and stays single-exec. Stages run inside one workflow instance posting one
 check-run — sequentially by default, concurrently when the next rung says so.
 
+### Container transport (`sandbox.transport:<repo>`)
+
+```bash
+wrangler kv key put --binding=CONFIG_KV "sandbox.transport:owner/repo" "rpc"
+```
+
+Pins every container that repo's executions acquire to one of `http` (the
+default), `websocket`, or `rpc`. **Any other value is dropped by the dispatcher
+before it reaches the SDK** — silently, and the run proceeds on the default. A
+typo therefore degrades rather than breaks, but it also says nothing: check the
+key back if a transport change appears to have had no effect.
+
+**Why it exists.** The SDK's streaming file APIs live only on its `rpc` client —
+its own comment calls `rpc` the "primary container-control client" and
+`http`/`websocket` the "route-based compatibility client", and `writeFileStream`
+is a bare `throw` off `rpc`. That is why the R2 dependency cache misses on every
+run: `installCached`'s restore hands `writeFile` a `ReadableStream`, the SDK
+routes any stream to `writeFileStream`, it raises, and `composeRestoreOr`
+records a miss.
+
+`SANDBOX_TRANSPORT=rpc` as a Worker var fixes that in one line — and changes the
+control path for every repo this dispatcher serves, at once. This key is the
+same choice scoped to one consumer, so a change with that blast radius can be
+proved before it is taken.
+
+**It is sticky.** `setTransport` persists to the container's Durable Object
+storage, and the SDK prefers a stored transport over the env-derived default on
+cold start. A container pinned here keeps that transport for its lifetime;
+since ids are per execution, the pin is re-applied per run and costs one DO call
+at acquire.
+
 ### A stage does not assume its checkout
 
 Container disk is ephemeral, and a staged run spanning forty minutes of durable