Build the session runtime from platform ports instead of DurableObjectState (COL-83)

packages/control-plane/src/cloudflare/session-platform.test.ts

@@ -0,0 +1,100 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import type { SqlDatabase } from "../db/sql-database";
+import type { Logger } from "../logger";
+import { createDurableObjectSessionPlatform } from "./session-platform";
+
+/** Stand-in for the Workers runtime's request/response pair. */
+class FakeRequestResponsePair {
+  constructor(
+    readonly request: string,
+    readonly response: string
+  ) {}
+}
+
+function createFakeState() {
+  const storage = {
+    sql: { exec: vi.fn() },
+    transactionSync: vi.fn(<T>(closure: () => T): T => closure()),
+    getAlarm: vi.fn(async () => null),
+    setAlarm: vi.fn(async () => {}),
+    deleteAlarm: vi.fn(async () => {}),
+  };
+  const calls = {
+    id: { toString: () => "do-id" },
+    storage,
+    acceptWebSocket: vi.fn(),
+    getTags: vi.fn(() => ["sandbox", "sid:sb-1"]),
+    getWebSockets: vi.fn(() => []),
+    setWebSocketAutoResponse: vi.fn(),
+    waitUntil: vi.fn(),
+  };
+  return { state: calls as unknown as DurableObjectState, storage, calls };
+}
+
+describe("createDurableObjectSessionPlatform", () => {
+  beforeEach(() => {
+    vi.stubGlobal("WebSocketRequestResponsePair", FakeRequestResponsePair);
+  });
+  afterEach(() => {
+    vi.unstubAllGlobals();
+  });
+
+  it("exposes the object's id, SQL, transaction, alarm store, and db", () => {
+    const { state, storage } = createFakeState();
+    const db = {} as SqlDatabase;
+
+    const platform = createDurableObjectSessionPlatform(state, db);
+
+    expect(platform.id).toBe("do-id");
+    expect(platform.sql).toBe(storage.sql);
+    expect(platform.db).toBe(db);
+    expect(platform.alarmStore).toBe(storage);
+    expect(platform.transactionSync(() => 42)).toBe(42);
+    expect(storage.transactionSync).toHaveBeenCalledTimes(1);
+  });
+
+  it("delegates socket acceptance, tags, and enumeration, passing the tag filter through", () => {
+    const { state, calls } = createFakeState();
+    const ws = {} as WebSocket;
+
+    const platform = createDurableObjectSessionPlatform(state, null);
+    platform.sockets.accept(ws, ["sandbox", "sid:sb-1"]);
+    platform.sockets.all();
+    platform.sockets.all("sandbox");
+
+    expect(calls.acceptWebSocket).toHaveBeenCalledWith(ws, ["sandbox", "sid:sb-1"]);
+    expect(platform.sockets.tags(ws)).toEqual(["sandbox", "sid:sb-1"]);
+    expect(calls.getTags).toHaveBeenCalledWith(ws);
+    expect(calls.getWebSockets.mock.calls).toEqual([[undefined], ["sandbox"]]);
+  });
+
+  it("installs the auto-response as a request/response pair", () => {
+    const { state, calls } = createFakeState();
+
+    const platform = createDurableObjectSessionPlatform(state, null);
+    platform.sockets.setAutoResponse('{"type":"ping"}', '{"type":"pong"}');
+
+    expect(calls.setWebSocketAutoResponse).toHaveBeenCalledTimes(1);
+    const pair = calls.setWebSocketAutoResponse.mock.calls[0][0] as FakeRequestResponsePair;
+    expect(pair).toBeInstanceOf(FakeRequestResponsePair);
+    expect(pair.request).toBe('{"type":"ping"}');
+    expect(pair.response).toBe('{"type":"pong"}');
+  });
+
+  it("builds background tasks over the object's event lifetime that report to the given logger", async () => {
+    const { state, calls } = createFakeState();
+    const logger = { error: vi.fn() } as unknown as Logger;
+
+    const platform = createDurableObjectSessionPlatform(state, null);
+    platform.createBackgroundTasks(logger).submit(() => Promise.reject(new Error("boom")), {
+      name: "session.task",
+    });
+
+    expect(calls.waitUntil).toHaveBeenCalledTimes(1);
+    await calls.waitUntil.mock.calls[0][0];
+    expect(logger.error).toHaveBeenCalledWith(
+      "background_task.failed",
+      expect.objectContaining({ task_name: "session.task" })
+    );
+  });
+});

packages/control-plane/src/cloudflare/session-platform.ts

@@ -0,0 +1,30 @@
+import type { SqlDatabase } from "../db/sql-database";
+import type { SessionPlatform } from "../session/platform";
+import { createCloudflareBackgroundTasks } from "./background-tasks";
+
+/**
+ * A Durable Object's storage, hibernatable sockets, alarm, and event lifetime
+ * as the session platform.
+ */
+export function createDurableObjectSessionPlatform(
+  ctx: DurableObjectState,
+  db: SqlDatabase | null
+): SessionPlatform {
+  return {
+    id: ctx.id.toString(),
+    sql: ctx.storage.sql,
+    transactionSync: <T>(closure: () => T): T => ctx.storage.transactionSync(closure),
+    db,
+    alarmStore: ctx.storage,
+    sockets: {
+      accept: (ws, tags) => ctx.acceptWebSocket(ws, tags),
+      tags: (ws) => ctx.getTags(ws),
+      all: (tag) => ctx.getWebSockets(tag),
+      // Hibernation-level auto-response: matched by the runtime without
+      // waking the object.
+      setAutoResponse: (request, response) =>
+        ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair(request, response)),
+    },
+    createBackgroundTasks: (log) => createCloudflareBackgroundTasks(ctx, log),
+  };
+}

packages/control-plane/src/session/components.ts

@@ -50,6 +50,7 @@ import { requireRepoSecretsEncryptionKey, requireTokenEncryptionKey } from "../e
 import type { Env, ClientInfo } from "../types";
 import type { SessionRow } from "./types";
 import type { SqlDatabase } from "../db/sql-database";
+import type { SessionPlatform } from "./platform";
 import { SessionCoreRepository } from "./session-core-repository";
 import { SandboxRepository } from "./sandbox-repository";
 import { SessionAttachmentRepository } from "./session-attachment-repository";
@@ -80,7 +81,6 @@ import { CallbackNotificationService } from "./callback-notification-service";
 import { UserEnvResolver } from "./user-env-resolver";
 import { resolveSessionRepoId } from "./repo-id-resolution";
 import { Scheduler } from "../scheduler/scheduler";
-import { createCloudflareBackgroundTasks } from "../cloudflare/background-tasks";
 import { PresenceService } from "./presence-service";
 import { SessionMessageQueue } from "./message-queue";
 import { SandboxArtifactEventHandler } from "./sandbox-events/artifact.handler";
@@ -138,13 +138,6 @@ import { AuthorizationError, AuthorizationService } from "../authorization/servi
  */
 const WS_AUTH_TIMEOUT_MS = 30000; // 30 seconds
 
-/** The platform surface the session graph is built over. */
-export interface SessionPlatform {
-  ctx: DurableObjectState;
-  sql: SqlStorage;
-  db: SqlDatabase | null;
-}
-
 /**
  * What the platform adapter (SessionDO) is allowed to touch. Everything else
  * stays inside the factory; `internals` exists for integration tests that
@@ -211,9 +204,15 @@ function resolveExecutionTimeoutMs(
 
 /** Build the session runtime, including authorization verification and lease expiry handling. */
 export function createSessionRuntime(platform: SessionPlatform, env: Env): SessionRuntime {
-  const { ctx, sql, db } = platform;
-  const durableObjectId = ctx.id.toString();
-  const transaction = <T>(closure: () => T): T => ctx.storage.transactionSync(closure);
+  const {
+    id: durableObjectId,
+    sql,
+    transactionSync: transaction,
+    db,
+    alarmStore,
+    sockets: socketPlatform,
+    createBackgroundTasks,
+  } = platform;
 
   // Tier 1 — repositories and alarm persistence (leaves over SqlStorage).
   const attachmentRepository = new SessionAttachmentRepository(sql);
@@ -249,29 +248,27 @@ export function createSessionRuntime(platform: SessionPlatform, env: Env): Sessi
     createLogger("session-do", {}, parseLogLevel(env.LOG_LEVEL)),
     getPublicSessionId
   );
-  const backgroundTasks = createCloudflareBackgroundTasks(ctx, log);
+  const backgroundTasks = createBackgroundTasks(log);
   // The sandbox repository validates the status it reads and warns on anything
   // unmodelled, so it needs the session logger — and it owns encrypt-at-rest
   // for access secrets, so it takes the key.
   const sandboxRepository = new SandboxRepository(sql, log, repoSecretsEncryptionKey);
 
   // Tier 2 — sockets and alarm scheduling.
-  const alarmScheduler = createEarliestAlarmScheduler(ctx.storage, alarmDeadlines);
+  const alarmScheduler = createEarliestAlarmScheduler(alarmStore, alarmDeadlines);
   const wsManager: SessionWebSocketManager = new SessionWebSocketManagerImpl(
-    ctx,
+    socketPlatform,
     sandboxRepository,
     wsClientMappingRepository,
     alarmScheduler,
     log,
     { authTimeoutMs: WS_AUTH_TIMEOUT_MS }
   );
-  // Hibernation-level ping/pong: the runtime answers keepalives without
-  // waking the Durable Object. Platform-global wiring, so it lives here.
-  ctx.setWebSocketAutoResponse(
-    new WebSocketRequestResponsePair(
-      JSON.stringify({ type: "ping" }),
-      JSON.stringify({ type: "pong", timestamp: Date.now() })
-    )
+  // Platform-level ping/pong: keepalives are answered without waking the
+  // runtime. Session-wide wiring, so it lives here.
+  socketPlatform.setAutoResponse(
+    JSON.stringify({ type: "ping" }),
+    JSON.stringify({ type: "pong", timestamp: Date.now() })
   );
 
   // Tier 3 — outbound delivery over the socket registry.

packages/control-plane/src/session/durable-object.ts

@@ -10,25 +10,25 @@
 import { DurableObject } from "cloudflare:workers";
 import { initSchema } from "./schema";
 import type { Env } from "../types";
-import type { SqlDatabase } from "../db/sql-database";
+import { createDurableObjectSessionPlatform } from "../cloudflare/session-platform";
+import type { SessionPlatform } from "./platform";
 import { createSessionRuntime, type SessionRuntime } from "./components";
 
 export class SessionDO extends DurableObject<Env> {
-  private sql: SqlStorage;
   /**
-   * The DO's global-database handle — the single point where env.DB is read.
-   * Nullable to preserve the existing defensive guards against a missing
-   * binding at runtime. Distinct from `this.sql`, the DO-embedded SQLite.
+   * This object's storage, sockets, alarm, and event lifetime as the ports
+   * the runtime is built over. Its `db` is the single point where env.DB is
+   * read, nullable to preserve the existing defensive guards against a
+   * missing binding at runtime.
    */
-  private readonly db: SqlDatabase | null;
+  private readonly platform: SessionPlatform;
   // The per-activation runtime; null until ensureInitialized() builds it.
   private _runtime: SessionRuntime | null = null;
 
   constructor(ctx: DurableObjectState, env: Env) {
     super(ctx, env);
     // eslint-disable-next-line no-restricted-syntax -- composition root input: the DO's one env.DB read
-    this.db = env.DB ?? null;
-    this.sql = ctx.storage.sql;
+    this.platform = createDurableObjectSessionPlatform(ctx, env.DB ?? null);
   }
 
   /** The runtime, (re)built on first touch after construction or eviction. */
@@ -44,8 +44,8 @@ export class SessionDO extends DurableObject<Env> {
   private ensureInitialized(rehydrateAlarm = true): void {
     if (this._runtime) return;
     const initStart = performance.now();
-    initSchema(this.sql);
-    const runtime = createSessionRuntime({ ctx: this.ctx, sql: this.sql, db: this.db }, this.env);
+    initSchema(this.platform.sql);
+    const runtime = createSessionRuntime(this.platform, this.env);
     // Publish only after the graph is fully built: a throw above leaves the
     // activation uninitialized, so the next event retries initialization
     // instead of dereferencing an undefined runtime.

packages/control-plane/src/session/platform.ts

@@ -0,0 +1,50 @@
+/**
+ * The platform surface one session runtime is built over: what a host must
+ * supply for `createSessionRuntime` to assemble the collaborator graph. The
+ * Cloudflare adapter is `createDurableObjectSessionPlatform`
+ * (cloudflare/session-platform.ts); a Node host supplies the same record from
+ * its own storage, sockets, and process facilities.
+ */
+
+import type { Logger } from "../logger";
+import type { BackgroundTasks } from "../platform-ports";
+import type { SqlDatabase } from "../db/sql-database";
+import type { AlarmScheduleStore } from "./alarm/scheduler";
+import type { SqlStorage, TransactionSync } from "./sql-storage";
+
+/** Host socket operations the session's connection registry is built over. */
+export interface SocketPlatform {
+  /** Adopt `ws` into the runtime, tagged so its identity survives a restart. */
+  accept(ws: WebSocket, tags: string[]): void;
+  /** The tags `ws` was accepted with. */
+  tags(ws: WebSocket): string[];
+  /** Every accepted socket, or only those carrying `tag`. */
+  all(tag?: string): WebSocket[];
+  /**
+   * Answer `request` frames with `response` at the platform level, without
+   * waking the runtime.
+   */
+  setAutoResponse(request: string, response: string): void;
+}
+
+export interface SessionPlatform {
+  /**
+   * The host's identity for this runtime. It stands in for the session id
+   * until `init` writes the session row.
+   */
+  id: string;
+  /** The session's own SQLite store. */
+  sql: SqlStorage;
+  /** Run `closure` atomically against `sql`. */
+  transactionSync: TransactionSync;
+  /** The global store, or null when the deployment has none bound. */
+  db: SqlDatabase | null;
+  /** The runtime's single scheduled wake-up. */
+  alarmStore: AlarmScheduleStore;
+  sockets: SocketPlatform;
+  /**
+   * Build the deferred-work port for this runtime. Takes the session-scoped
+   * logger so failures of background work are attributed to the session.
+   */
+  createBackgroundTasks(log: Logger): BackgroundTasks;
+}