/**
 * RoomDO — one Durable Object per spreadsheet room.
 *
 * Phase 5 deliverable: snapshot + log + audit + chat + ecell storage, backed
 * by SocialCalc via `@ethercalc/socialcalc-headless`. Exposes an internal
 * HTTP API on `/_do/*` that the Worker's room-level routes dispatch to.
 *
 * Key patterns come from `@ethercalc/shared/storage-keys`:
 *   - `snapshot`               → string (SocialCalc save)
 *   - `meta:updated_at`        → number (Date.now())
 *   - `log:<seq>`              → string (one command batch per entry)
 *   - `audit:<seq>`            → string (never truncated)
 *   - `chat:<seq>`             → string (room chat message)
 *   - `ecell:<user>`           → string (cell coordinate)
 *
 * Sequence counters (`nextLogSeq`, `nextAuditSeq`, `nextChatSeq`) are lazily
 * initialized from `storage.list({prefix})` on first write — cheap because
 * DO storage keeps the SQLite index in memory for warm instances.
 *
 * The in-memory SocialCalc `HeadlessSpreadsheet` is hydrated lazily from the
 * stored snapshot (if any) on first use and cached for the lifetime of the
 * isolate. Mutations that would drift cache vs storage are wrapped in
 * `state.blockConcurrencyWhile` to keep the DO serialized.
 */
import {
  encodeMessage,
  MAX_WS_FRAME_CHARS,
  isStorageSafeCommand,
  parseClientMessage,
  type ServerMessage,
} from "@ethercalc/shared/messages";
import {
  decodeFrame,
  nativeToSocketIoEvent,
  PacketType,
  socketIoEventToNative,
} from "@ethercalc/socketio-shim";
import {
  auditKey,
  chatKey,
  ecellKey,
  logKey,
  STORAGE_KEYS,
  snapshotChunkKey,
} from "@ethercalc/shared/storage-keys";
import { createSpreadsheet, HeadlessSpreadsheet } from "@ethercalc/socialcalc-headless";
import type { Env } from "./env.ts";
import { buildEmailSender } from "./handlers/cron.ts";
import { parseSeedPayload } from "./handlers/migrate.ts";
import { verifyAuth } from "./lib/auth.ts";
import { verifyAuthSession } from "./lib/auth-session.ts";
import { authorize as authorizeRoom } from "./lib/authorize.ts";
import { hydrateCrossSheetRefs, readBoundedResponseText } from "./lib/cross-sheet.ts";
import { isCommandBatchWithinLimits, isSnapshotWithinSheetLimits } from "./lib/command-limits.ts";
import { PARENT_READ_HEADER } from "./lib/do-dispatch.ts";
import { neutralizeCSVDocument } from "./lib/csv-encode.ts";
import { parseCSV } from "./lib/csv-parse.ts";
import { parseSendemail } from "./lib/email.ts";
import { formdataSiblingRoom } from "./lib/formdata-sibling.ts";
import { verifyMigrateToken } from "./lib/migrate-auth.ts";
import { csvToMarkdown } from "./lib/md.ts";
import { getMaxSubsheetIndex } from "./lib/multi-sheet-import.ts";
import { workbookToLoadClipboardCommand } from "./lib/xlsx-import.ts";
import { MAX_MULTI_SHEETS, MAX_MULTI_SHEET_TITLE_LENGTH } from "@ethercalc/shared";
import { bookmarkStorage, isPitrUnavailableError, parsePitrRequest } from "./lib/pitr.ts";
import { encodeRoom, isValidRoomName } from "./lib/room-name.ts";
import { deleteRoomFromD1, mirrorRoomToD1 } from "./lib/rooms-index.ts";
import {
  isSandstormEnforced,
  sandstormAllowsWsWrite,
  sandstormCanModify,
} from "./lib/sandstorm-access.ts";
import type { SessionPrincipal } from "./lib/session.ts";
import {
  AUDIT_HISTORY_KEEP,
  appendAuditRows,
  appendChatRows,
  CHAT_HISTORY_KEEP,
  deleteAuditRows,
  deleteChatRows,
  type SeqRow,
} from "./lib/seq-store.ts";
import {
  hasSnapshot,
  SNAPSHOT_CHUNK_BYTES,
  MAX_SNAPSHOT_CHUNKS,
  readSnapshot,
  readSnapshotMeta,
  type SnapshotMeta,
  snapshotEntries,
} from "./lib/snapshot-storage.ts";
import {
  deleteStorageKeysBatched,
  putStorageEntriesBatched,
  replaceStorageEntriesBatched,
} from "./lib/storage-batch.ts";
import { isFilteredExecuteCommand } from "./lib/ws-dispatch.ts";
import {
  dispatchWsMessage,
  type WsContext,
  type WsSiblingDO,
  type WsStorage,
} from "./lib/ws-handlers.ts";
import { upgradeLegacySocketIo, upgradeWebSocket, type WsAttachment } from "./lib/ws-upgrade.ts";
import {
  BINARY_CONTENT_TYPES,
  type BinaryFormat,
  sheetViewToBinaryWorkbook,
} from "./lib/xlsx-build.ts";

/** Shape returned from `GET /_do/log`. */
export interface RoomLogSnapshot {
  readonly log: readonly string[];
  readonly chat: readonly string[];
}

/** Content type used for plain-text bodies returned from the DO. */
const PLAIN_TEXT = "text/plain; charset=utf-8";
const APP_JSON = "application/json";
const TEXT_CSV = "text/csv; charset=utf-8";
const TEXT_HTML = "text/html; charset=utf-8";
const TEXT_MARKDOWN = "text/x-markdown; charset=utf-8";

/**
 * How long one AuthDO session verification stays reusable inside a
 * RoomDO isolate, and how many distinct tokens it retains.
 *
 * Without a cache, one inbound frame in a private room costs `1 + N`
 * subrequests against the deployment-wide `idFromName('auth')` object
 * (one per peer re-checked by `#broadcast`), so a single authorized
 * client sitting at the room message-rate ceiling could saturate the
 * global login path. The TTL is the resulting revocation-latency
 * ceiling — the same bounded-staleness tradeoff already accepted for
 * `#accessMeta`. Session *expiry* never depends on this cache:
 * `#closeExpiredSessionSocket` reads `sessionExp` off the socket
 * attachment synchronously, before any verification runs.
 */
const SESSION_VERIFY_TTL_MS = 3_000;
const SESSION_VERIFY_CAP = 256;

type SessionVerification = {
  readonly at: number;
  readonly principal: Promise<SessionPrincipal | null>;
};

/**
 * Ring-buffer length for the command log. The stored snapshot is
 * authoritative on hydrate (see `#getSpreadsheet`), so `log:` is now a
 * pure client-catch-up buffer: `ask.log` returns the recent tail alongside
 * the snapshot, and any client that has fallen further behind than this
 * window resets to the snapshot instead of replaying. `#appendCommand`
 * deletes `log:<seq - LOG_RING>` as it writes `log:<seq>`, so the live
 * log never holds more than `LOG_RING` entries. `audit:` is NEVER trimmed
 * here (it is the append-only record).
 */
const MULTI_RESERVATION_PREFIX = "meta:multi-reservation:";
const MULTI_RESERVATION_TTL_MS = 10 * 60 * 1000;
const MAX_MULTI_RESERVATIONS = 16;
const LOG_RING = 1024;

/**
 * Cap on distinct `ecell:<user>` keys retained per room. ecells are keyed
 * by an arbitrary client-supplied username, so without a bound an attacker
 * could blow per-room storage by cycling usernames. We evict the
 * least-recently-written entry once the cap is exceeded (`#trackEcell`).
 */
const ECELL_CAP = 256;

/**
 * Maximum concurrent WebSocket connections accepted per room. Past this we
 * reject the upgrade — a coarse DoS backstop complementing the CF platform
 * layer (AGENTS.md §13 Q7 keeps real rate limiting at the edge).
 */
const MAX_CONN = 128;

/** Hibernation-safe per-socket message-rate window. */
const WS_RATE_WINDOW_MS = 10_000;
const MAX_WS_MESSAGES_PER_WINDOW = 300;
const MAX_ROOM_WS_MESSAGES_PER_WINDOW = 1_500;

/**
 * Cadence (ms) at which the housekeeping alarm re-fires while a room stays
 * active. One hour keeps chat-trim/TTL checks cheap without a tight loop.
 */
const ALARM_INTERVAL_MS = 60 * 60 * 1000;

/**
 * Give the accepted restore response time to cross the DO boundary before
 * aborting the instance that produced it. The exact delay is not semantic.
 */
// Stryker disable next-line all : any short positive delay has the same contract
const PITR_ABORT_DELAY_MS = 100;

function plainResponse(body: string, status = 200): Response {
  return new Response(body, {
    status,
    headers: { "Content-Type": PLAIN_TEXT },
  });
}

function jsonResponse(body: unknown, status = 200): Response {
  return new Response(JSON.stringify(body), {
    status,
    headers: { "Content-Type": APP_JSON },
  });
}

function notFound(): Response {
  return plainResponse("", 404);
}

function textResponse(body: string, contentType: string, status = 200): Response {
  return new Response(body, { status, headers: { "Content-Type": contentType } });
}

function binaryResponse(bytes: Uint8Array, contentType: string, status = 200): Response {
  // workerd accepts a `Uint8Array` as a `BodyInit` at runtime; the cast
  // satisfies the stricter `Uint8Array<ArrayBufferLike>` lib typing that
  // doesn't structurally match `BodyInit` (the DO → worker hop streams it).
  return new Response(bytes as unknown as BodyInit, {
    status,
    headers: { "Content-Type": contentType },
  });
}

/**
 * Fold a base snapshot + since-base command log into a single
 * authoritative SocialCalc save. Used on ingest by `#postSeed` and
 * `#postInstall` so the stored snapshot already incorporates every log
 * command (the hydrate path no longer replays the log over a present
 * snapshot — see `#getSpreadsheet`). `createSpreadsheet({snapshot, log})`
 * applies each log line exactly once on top of the base, then we
 * serialise the result.
 */
function foldSnapshot(snapshot: string, log: readonly string[]): string {
  const ss = createSpreadsheet(snapshot ? { snapshot, log } : { log });
  return ss.createSpreadsheetSave();
}

/**
 * Parse the legacy `--expire` / `ETHERCALC_EXPIRE` value (a TTL in
 * SECONDS, matching the old Redis `EXPIRE` semantics) into milliseconds.
 * Returns `null` when unset, non-numeric, or non-positive — in those
 * cases the alarm handler skips TTL expiry entirely (rooms live forever,
 * the production default).
 */
function parseExpireMs(raw: string | undefined): number | null {
  if (raw === undefined) return null;
  const seconds = Number(raw);
  if (!Number.isFinite(seconds) || seconds <= 0) return null;
  return seconds * 1000;
}

export class RoomDO implements DurableObject {
  readonly #state: DurableObjectState;
  readonly #env: Env;
  readonly #instanceNonce = crypto.randomUUID();

  #ss: HeadlessSpreadsheet | null = null;
  #nextLogSeq: number | null = null;
  #nextAuditSeq: number | null = null;
  #nextChatSeq: number | null = null;
  /**
   * In-memory LRU order of `ecell:<user>` keys (least-recently-written
   * first). Lazily seeded from storage on first ecell write so the cap is
   * enforced even after an isolate restart. Bounds distinct ecell keys to
   * `ECELL_CAP` so a client cycling arbitrary usernames can't grow storage
   * without limit.
   */
  #ecellOrder: string[] | null = null;
  /** Whether the housekeeping alarm is known to be armed (cheap dedupe). */
  #alarmArmed = false;
  #roomRateWindowStartedAt = 0;
  #roomRateMessageCount = 0;
  /**
   * Cached room name — set from `?name=…` on each request and retained
   * for cross-sheet formula resolution (so sibling DO lookups can skip
   * self-references without forcing every caller to thread the name
   * through `#getSpreadsheet`).
   */
  #ownName: string | undefined;
  /**
   * Memoized local access metadata. Parent *verdicts are never cached*: a
   * parent ACL revocation must take effect on its children immediately.
   */
  #accessMeta: {
    hasAccess: boolean;
    access: unknown;
    hasAcl: boolean;
    acl: unknown;
    hasParent: boolean;
    parent: unknown;
  } | null = null;
  /**
   * Isolate-local AuthDO verification memo, keyed by opaque session
   * token (see `SESSION_VERIFY_TTL_MS`). Insertion-ordered, so the
   * oldest entry is evicted once `SESSION_VERIFY_CAP` distinct tokens
   * are retained — an attacker cycling tokens pays the AuthDO round
   * trip they would have paid anyway, and cannot grow this map.
   */
  #sessionVerifications = new Map<string, SessionVerification>();

  constructor(state: DurableObjectState, env: Env) {
    this.#state = state;
    this.#env = env;
    // socket.io v0.9 heartbeat is a pure echo of `2::`. Auto-response lets
    // hibernated legacy sockets answer pings without waking the isolate
    // (and without a JS timer that would pin the DO awake).
    /* istanbul ignore next -- @preserve: WebSocketRequestResponsePair is a workerd global */
    if (typeof WebSocketRequestResponsePair === "function") {
      this.#state.setWebSocketAutoResponse(new WebSocketRequestResponsePair("2::", "2::"));
    }
  }

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    const path = url.pathname;
    // Every `doFetch` caller now threads the room name through as
    // `?name=…` so the DO can mirror to D1 without re-deriving it from
    // its opaque id. `/_do/ping` already used the same param.
    const roomName = url.searchParams.get("name");
    if (roomName) this.#ownName = roomName;
    const uid = request.headers.get("X-EC-Uid");
    // Worker-internal capability and operator paths bypass the generic ACL
    // gate. None serve sheet content: access returns only a verdict, PITR is
    // operator-authenticated, and fire-trigger is reached only by the Worker
    // scheduler (the public compatibility route requires the operator token).
    const isGateExemptPath =
      path === "/_do/access" ||
      path === "/_do/ping" ||
      path === "/_do/pitr-restore" ||
      path === "/_do/pitr-touch" ||
      path === "/_do/fire-trigger";
    const trustedParentRead = await this.#isTrustedParentRead(request, path);
    if (!isGateExemptPath && !trustedParentRead) {
      const purpose = request.method === "GET" ? "read" : "write";
      if (!(await this.#isAuthorized(request, purpose))) {
        return plainResponse("Forbidden", 403);
      }
    }

    if (path === "/_do/access" && request.method === "GET") {
      return this.#getAccessVerdict(request);
    }
    if (path === "/_do/ping") {
      return jsonResponse({
        id: this.#state.id.toString(),
        name: roomName,
        nonce: this.#instanceNonce,
      });
    }
    if (path === "/_do/pitr-restore" && request.method === "POST") {
      return this.#postPitrRestore(request);
    }
    if (path === "/_do/pitr-touch" && request.method === "POST") {
      return this.#postPitrTouch(roomName);
    }
    if (path === "/_do/snapshot") {
      if (request.method === "GET") return this.#getSnapshot();
      if (request.method === "PUT") return this.#putSnapshot(request, roomName);
    }
    if (path === "/_do/log" && request.method === "GET") {
      return this.#getLog();
    }
    if (path === "/_do/commands" && request.method === "POST") {
      return this.#postCommands(request, roomName);
    }
    if (path === "/_do/child-snapshot" && request.method === "POST") {
      return this.#postChildSnapshot(request, roomName);
    }
    if (path === "/_do/set-parent" && request.method === "POST") {
      return this.#postSetParent(request, roomName);
    }
    if (path === "/_do/backfill-children" && request.method === "POST") {
      return this.#postBackfillChildren(request, roomName);
    }
    if (path === "/_do/import-append-toc" && request.method === "POST") {
      return this.#postImportAppendToc(request, roomName);
    }
    if (path === "/_do/all" && request.method === "DELETE") {
      return this.#deleteAll(roomName, request.headers.get("X-EC-Uid"));
    }
    if (path === "/_do/exists" && request.method === "GET") {
      return this.#getExists();
    }
    if (path === "/_do/workbook-kind" && request.method === "GET") {
      return this.#getWorkbookKind(uid);
    }
    if (path === "/_do/cells" && request.method === "GET") {
      return this.#getCells(uid);
    }
    const cellMatch = path.match(/^\/_do\/cells\/(.+)$/);
    if (cellMatch && request.method === "GET") {
      try {
        return this.#getCell(decodeURIComponent(cellMatch[1]!), uid);
      } catch {
        return plainResponse("Bad Request", 400);
      }
    }
    // ─── Phase 8: export routes ────────────────────────────────────────
    if (path === "/_do/html" && request.method === "GET") {
      return this.#getHtml(uid);
    }
    if (path === "/_do/csv" && request.method === "GET") {
      return this.#getCsv(uid);
    }
    if (path === "/_do/csv.json" && request.method === "GET") {
      return this.#getCsvJson(uid);
    }
    if (path === "/_do/md" && request.method === "GET") {
      return this.#getMd(uid);
    }
    if (path === "/_do/xlsx" && request.method === "GET") {
      return this.#getBinary("xlsx", uid);
    }
    if (path === "/_do/ods" && request.method === "GET") {
      return this.#getBinary("ods", uid);
    }
    if (path === "/_do/fods" && request.method === "GET") {
      return this.#getBinary("fods", uid);
    }
    // ─── Phase 8.1: sheet-data for multi-sheet export ────────────────
    // Returns the structural SheetData (cells + valueformats + cellformats
    // + attribs) as JSON, for the top-level `/_/=:room/*` route to walk
    // cross-DO and build a multi-sheet workbook with formula fidelity.
    if (path === "/_do/sheet-data" && request.method === "GET") {
      return this.#getSheetData(uid);
    }
    // ─── Phase 6: cross-DO rename primitives ─────────────────────────
    // `set A\d+:B\d+ empty multi-cascade` in the HTTP command layer
    // moves snapshot/log/audit from <from> into <to> and wipes <from>.
    // `rename` runs on the source DO; `install` is the target-side
    // receiver. Both additive; no existing path shape changes.
    if (path === "/_do/rename" && request.method === "POST") {
      return this.#postRename(request);
    }
    if (path === "/_do/install" && request.method === "POST") {
      return this.#postInstall(request);
    }
    if (path === "/_do/clone" && request.method === "POST") {
      return this.#postClone(request);
    }
    // ─── Phase 11b: full-fidelity migration seed ─────────────────────
    // `POST /_do/seed` is the migration entry point — it replaces the
    // entire room (snapshot + log + audit + chat + ecell + meta
    // timestamp) in one shot, then mirrors the D1 `rooms` row. The
    // worker-level `PUT /_migrate/seed/:room` route authenticates the
    // caller before dispatching here; direct DO access from inside the
    // namespace (tests, future tooling) can bypass that.
    if (path === "/_do/seed" && request.method === "POST") {
      return this.#postSeed(request, roomName);
    }
    // ─── Phase 11b: client-side chunked snapshot upload ──────────────
    // Companion to `/_do/seed` for rooms whose raw SocialCalc save
    // exceeds CF's ~25 MB per-request body limit. The migrator first
    // POSTs `/_do/seed` with an empty `snapshot` (which deleteAll's
    // the DO and installs log/audit/chat/ecell), then streams N chunk
    // bodies here with `seq=<i>&chunks=<N>`. The final chunk flips
    // `snapshot:meta` over to the new layout; readers see either the
    // pre-migration state or the freshly-assembled save, never a
    // mix. See the `#postSnapshotChunk` doc for the full contract.
    if (path === "/_do/snapshot-chunk" && request.method === "POST") {
      return this.#postSnapshotChunk(request, roomName, url.searchParams);
    }
    // ─── Phase A: atomic private-room initialization ─────────────────
    if (path === "/_do/init-private" && request.method === "POST") {
      return this.#postInitPrivate(request);
    }
    // ─── Phase 7: native WebSocket upgrade ───────────────────────────
    if (path === "/_do/ws" && request.method === "GET") {
      return this.#acceptWebSocket(request);
    }
    // Legacy socket.io v0.9 WS — hibernation API with `legacy: true`
    // attachment so webSocketMessage/send use socket.io framing. Worker
    // routes `/socket.io/1/websocket/:sid` here on a sid-keyed RoomDO.
    if (path === "/_do/legacy-ws" && request.method === "GET") {
      return this.#acceptLegacyWebSocket(request);
    }
    // ─── Phase 9: cron fire-trigger hook ───────────────────────────────
    // `POST /_do/fire-trigger?cell=<coord>` — called from the
    // `scheduled()` handler (and from the backwards-compat
    // /_timetrigger HTTP endpoint) for each due row. Reads the
    // referenced cell's text, parses it as `sendemail <to> <subject>
    // <body>`, dispatches through the injected EmailSender, and
    // broadcasts the legacy `confirmemailsent` WS event to this
    // room's peers.
    if (path === "/_do/fire-trigger" && request.method === "POST") {
      return this.#fireTrigger(url.searchParams.get("cell") ?? "");
    }
    return new Response("Not implemented", { status: 501 });
  }

  // ─── Handlers ──────────────────────────────────────────────────────────

  async #isAuthorized(request: Request, purpose: "read" | "write"): Promise<boolean> {
    const uid = request.headers.get("X-EC-Uid");
    const access = await this.#effectiveAccess(uid);
    return purpose === "read" ? access.canRead : access.canWrite;
  }

  async #getAccessVerdict(request: Request): Promise<Response> {
    const uid = request.headers.get("X-EC-Uid");
    const access = await this.#effectiveAccess(uid);
    return jsonResponse({
      isPrivate: access.isPrivate,
      canRead: access.canRead,
      canWrite: access.canWrite,
    });
  }

  async #isTrustedParentRead(request: Request, path: string): Promise<boolean> {
    if (request.method !== "GET" || path !== "/_do/snapshot") return false;
    const claimedParent = request.headers.get(PARENT_READ_HEADER);
    const meta = await this.#getAccessMeta();
    return (
      meta.hasParent &&
      !meta.hasAccess &&
      !meta.hasAcl &&
      // Stryker disable next-line ConditionalExpression : equality to the
      // string-valued header below independently proves that the stored
      // parent is a string; this guard exists only for type clarity.
      typeof meta.parent === "string" &&
      meta.parent === claimedParent
    );
  }

  /**
   * Resolve the one authoritative access verdict for this room.
   *
   * A parent marker delegates the entire decision to the parent RoomDO.
   * Local access beside a parent marker is corruption, never an override.
   * Parent verdicts are intentionally uncached so revocation is immediate.
   */
  /**
   * Security regression proof (see matching names in `test/room.node.test.ts`):
   * six guards were inverted one at a time and each made its focused test red.
   * Ignoring `meta:parent` changed HTTP/export 403s to 200 and left the WS
   * open; permitting conflicting local metadata changed 403/deny to 200/allow;
   * bypassing the pristine-child guard changed 409 to 201; indexing a parented
   * child emitted a D1 INSERT; allowing a failed parent lookup changed deny to
   * allow; and ignoring the monotonic child index reused `.1` instead of `.2`.
   * Re-run those inversion checks before simplifying any of these guards.
   */
  async #effectiveAccess(uid: string | null): Promise<{
    isPrivate: boolean;
    canRead: boolean;
    canWrite: boolean;
    parented: boolean;
  }> {
    const meta = await this.#getAccessMeta();
    if (meta.hasParent) {
      if (meta.hasAccess || meta.hasAcl) {
        console.error("room-access-metadata-conflict", {
          room: this.#ownName ?? null,
        });
        return {
          isPrivate: true,
          canRead: false,
          canWrite: false,
          // Stryker disable next-line BooleanLiteral : this flag cannot affect
          // rename authorization because this same verdict denies both read
          // and write; those deny terms independently reject the operation.
          parented: true,
        };
      }
      if (
        typeof meta.parent !== "string" ||
        meta.parent.length === 0 ||
        meta.parent === this.#ownName
      ) {
        console.error("room-parent-metadata-invalid", {
          room: this.#ownName ?? null,
        });
        return {
          isPrivate: true,
          canRead: false,
          canWrite: false,
          // Stryker disable next-line BooleanLiteral : an invalid parent
          // verdict already denies both read and write, so the rename gate
          // rejects regardless of this redundant classification flag.
          parented: true,
        };
      }
      return this.#fetchParentAccess(meta.parent, uid);
    }

    const principal = uid === null ? null : { uid };
    return {
      isPrivate: meta.access === "private",
      canRead: authorizeRoom("read", principal, meta.access, meta.acl),
      canWrite: authorizeRoom("write", principal, meta.access, meta.acl),
      parented: false,
    };
  }

  async #fetchParentAccess(
    parent: string,
    uid: string | null,
  ): Promise<{
    isPrivate: boolean;
    canRead: boolean;
    canWrite: boolean;
    parented: true;
  }> {
    try {
      const stub = this.#env.ROOM.get(this.#env.ROOM.idFromName(encodeRoom(parent)));
      const headers = new Headers();
      if (uid !== null) headers.set("X-EC-Uid", uid);
      const response = await stub.fetch(
        `https://do.local/_do/access?name=${encodeURIComponent(parent)}`,
        {
          method: "GET",
          headers,
          signal: AbortSignal.timeout(5_000),
        },
      );
      if (response.status !== 200) {
        throw new Error(`parent access ${response.status}`);
      }
      const verdict: unknown = await response.json();
      if (
        verdict === null ||
        typeof verdict !== "object" ||
        !("isPrivate" in verdict) ||
        typeof verdict.isPrivate !== "boolean" ||
        !("canRead" in verdict) ||
        typeof verdict.canRead !== "boolean" ||
        !("canWrite" in verdict) ||
        typeof verdict.canWrite !== "boolean"
      ) {
        throw new Error("parent access malformed");
      }
      return {
        isPrivate: verdict.isPrivate,
        canRead: verdict.canRead,
        canWrite: verdict.canWrite,
        parented: true,
      };
    } catch (error) {
      console.error("room-parent-access-failed", {
        room: this.#ownName ?? null,
        parent,
        reason: error instanceof Error ? error.message : "unknown",
      });
      return {
        isPrivate: true,
        canRead: false,
        canWrite: false,
        // Stryker disable next-line BooleanLiteral : lookup failure denies
        // read and write, so the rename gate rejects independently of this
        // redundant classification flag.
        parented: true,
      };
    }
  }

  /**
   * Memoized local access-plane read. Parent verdicts are never memoized.
   */
  async #getAccessMeta(): Promise<{
    hasAccess: boolean;
    access: unknown;
    hasAcl: boolean;
    acl: unknown;
    hasParent: boolean;
    parent: unknown;
  }> {
    if (this.#accessMeta) return this.#accessMeta;
    const stored = await this.#state.storage.get<unknown>([
      STORAGE_KEYS.metaAccess,
      STORAGE_KEYS.metaAcl,
      STORAGE_KEYS.metaParent,
    ]);
    const meta = {
      hasAccess: stored.has(STORAGE_KEYS.metaAccess),
      access: stored.get(STORAGE_KEYS.metaAccess),
      hasAcl: stored.has(STORAGE_KEYS.metaAcl),
      acl: stored.get(STORAGE_KEYS.metaAcl),
      hasParent: stored.has(STORAGE_KEYS.metaParent),
      parent: stored.get(STORAGE_KEYS.metaParent),
    };
    this.#accessMeta = meta;
    return meta;
  }

  async #readAccessEntries(): Promise<Record<string, unknown>> {
    const keys = [
      STORAGE_KEYS.metaAccess,
      STORAGE_KEYS.metaAcl,
      STORAGE_KEYS.metaParent,
      STORAGE_KEYS.metaNextChildIndex,
      STORAGE_KEYS.metaGroup,
    ];
    const stored = await this.#state.storage.get<unknown>(keys);
    const entries: Record<string, unknown> = {};
    for (const key of keys) {
      if (stored.has(key)) entries[key] = stored.get(key);
    }
    return entries;
  }
  async #postChildSnapshot(request: Request, roomName: string | null): Promise<Response> {
    if (!roomName) return plainResponse("room name required", 400);
    let raw: unknown;
    try {
      raw = await request.json();
    } catch {
      return plainResponse("child-snapshot body must be valid JSON", 400);
    }
    if (raw === null || typeof raw !== "object") {
      return plainResponse("child-snapshot body must be an object", 400);
    }
    const parent = "parent" in raw ? raw.parent : null;
    const snapshot = "snapshot" in raw ? raw.snapshot : null;
    const mode = "mode" in raw ? raw.mode : null;
    if (
      !isValidRoomName(parent) ||
      typeof snapshot !== "string" ||
      (mode !== "create" && mode !== "replace" && mode !== "verify")
    ) {
      return plainResponse("child-snapshot body must be {parent,snapshot,mode}", 400);
    }
    if (!this.#isManagedChildName(parent, roomName)) {
      return plainResponse("child room is outside parent namespace", 400);
    }
    if (!isSnapshotWithinSheetLimits(snapshot)) {
      return plainResponse("child snapshot exceeds sheet limits", 413);
    }
    const uid = request.headers.get("X-EC-Uid");
    if (!(await this.#fetchParentAccess(parent, uid)).canWrite) {
      return plainResponse("Forbidden", 403);
    }

    type ChildSnapshotOutcome =
      | "created"
      | "idempotent"
      | "replace"
      | "occupied"
      | "unparented"
      | "local-access"
      | "different-parent";
    const outcome = await this.#state.blockConcurrencyWhile(
      async (): Promise<ChildSnapshotOutcome> => {
        const stored = await this.#state.storage.get<unknown>([
          STORAGE_KEYS.metaAccess,
          STORAGE_KEYS.metaAcl,
          STORAGE_KEYS.metaParent,
        ]);
        if (stored.has(STORAGE_KEYS.metaAccess) || stored.has(STORAGE_KEYS.metaAcl)) {
          return "local-access";
        }
        if (stored.has(STORAGE_KEYS.metaParent)) {
          if (stored.get(STORAGE_KEYS.metaParent) !== parent) {
            return "different-parent";
          }
          return mode === "replace" ? "replace" : "idempotent";
        }
        if (mode === "verify") return "unparented";
        // Stryker disable next-line ObjectLiteral : removing `limit: 1` reads
        // more keys but cannot change the only observed predicate, `size > 0`.
        if ((await this.#state.storage.list({ limit: 1 })).size > 0) {
          return "occupied";
        }
        await this.#state.storage.transaction(async (txn) => {
          await putStorageEntriesBatched(txn, {
            ...snapshotEntries(snapshot),
            [STORAGE_KEYS.metaParent]: parent,
            [STORAGE_KEYS.metaUpdatedAt]: Date.now(),
          });
        });
        return "created";
      },
    );
    if (outcome === "local-access") {
      return plainResponse("Child has local access metadata", 409);
    }
    if (outcome === "different-parent") {
      return plainResponse("Child belongs to a different parent", 409);
    }
    if (outcome === "unparented") {
      return plainResponse("Child is not initialized for this parent", 409);
    }
    if (outcome === "occupied") {
      return plainResponse(
        mode === "replace"
          ? "Child must be backfilled before replacement"
          : "Child room is occupied",
        409,
      );
    }
    if (outcome === "replace") {
      const replaced = await this.#putSnapshot(
        new Request(request.url, { method: "PUT", body: snapshot }),
        roomName,
      );
      if (!replaced.ok) return replaced;
    }
    this.#resetVolatile();
    await this.#armAlarm();
    return jsonResponse({ created: outcome === "created", parent }, 201);
  }

  #isManagedChildName(parent: string, child: string): boolean {
    const prefix = `${parent}.`;
    return child.startsWith(prefix) && /^[1-9]\d*$/.test(child.slice(prefix.length));
  }

  async #postSetParent(request: Request, roomName: string | null): Promise<Response> {
    if (!roomName) return plainResponse("room name required", 400);
    const token = verifyMigrateToken(
      this.#env.ETHERCALC_MIGRATE_TOKEN,
      request.headers.get("Authorization"),
    );
    if (token.kind === "disabled") return plainResponse("Not Found", 404);
    if (token.kind !== "ok") return plainResponse("Unauthorized", 401);
    let raw: unknown;
    try {
      raw = await request.json();
    } catch {
      return plainResponse("set-parent body must be valid JSON", 400);
    }
    if (raw === null || typeof raw !== "object") {
      return plainResponse("set-parent body must be an object", 400);
    }
    const parent = "parent" in raw ? raw.parent : null;
    const approved = "approved" in raw ? raw.approved : false;
    const dryRun = "dryRun" in raw ? raw.dryRun : false;
    if (!isValidRoomName(parent) || approved !== true || typeof dryRun !== "boolean") {
      return plainResponse("set-parent body must be {parent,approved:true,dryRun?:boolean}", 400);
    }
    if (!this.#isManagedChildName(parent, roomName)) {
      return plainResponse("child room is outside parent namespace", 400);
    }

    const uid = request.headers.get("X-EC-Uid");
    if (uid === null) return plainResponse("Forbidden", 403);
    try {
      const parentStub = this.#env.ROOM.get(this.#env.ROOM.idFromName(encodeRoom(parent)));
      const approvalResponse = await parentStub.fetch(
        `https://do.local/_do/backfill-children?name=${encodeURIComponent(parent)}`,
        {
          method: "POST",
          headers: {
            Authorization: request.headers.get("Authorization")!,
            "Content-Type": "application/json",
            "X-EC-Uid": uid,
          },
          body: JSON.stringify({
            children: [roomName],
            approved: true,
            dryRun: true,
          }),
          signal: AbortSignal.timeout(5_000),
        },
      );
      if (approvalResponse.status === 501) return plainResponse("Not implemented", 501);
      if (approvalResponse.status !== 200) return plainResponse("Forbidden", 403);
      const approval: unknown = await approvalResponse.json();
      if (
        approval === null ||
        typeof approval !== "object" ||
        !("children" in approval) ||
        !Array.isArray(approval.children) ||
        !approval.children.includes(roomName)
      ) {
        return plainResponse("Child is not in the parent TOC", 409);
      }
    } catch {
      return plainResponse("Forbidden", 403);
    }

    const outcome = await this.#state.blockConcurrencyWhile(async () => {
      return this.#state.storage.transaction(async (txn) => {
        const stored = await txn.get<unknown>([
          STORAGE_KEYS.metaAccess,
          STORAGE_KEYS.metaAcl,
          STORAGE_KEYS.metaParent,
        ]);
        if (stored.has(STORAGE_KEYS.metaAccess) || stored.has(STORAGE_KEYS.metaAcl)) {
          return "local-access" as const;
        }
        if (stored.has(STORAGE_KEYS.metaParent) && stored.get(STORAGE_KEYS.metaParent) !== parent) {
          return "different-parent" as const;
        }
        if (stored.get(STORAGE_KEYS.metaParent) === parent) {
          return "already-parented" as const;
        }
        if (!dryRun) await txn.put(STORAGE_KEYS.metaParent, parent);
        return dryRun ? ("eligible" as const) : ("set" as const);
      });
    });
    if (outcome === "local-access") {
      return plainResponse("Child has local access metadata", 409);
    }
    if (outcome === "different-parent") {
      return plainResponse("Child belongs to a different parent", 409);
    }
    if (outcome === "set") {
      this.#resetVolatile();
      await this.#deleteIndex(roomName);
    }
    return jsonResponse({
      child: roomName,
      parent,
      status: outcome,
    });
  }

  async #postBackfillChildren(request: Request, roomName: string | null): Promise<Response> {
    if (!roomName) return plainResponse("room name required", 400);
    const token = verifyMigrateToken(
      this.#env.ETHERCALC_MIGRATE_TOKEN,
      request.headers.get("Authorization"),
    );
    if (token.kind === "disabled") return plainResponse("Not Found", 404);
    if (token.kind !== "ok") return plainResponse("Unauthorized", 401);
    const uid = request.headers.get("X-EC-Uid");
    const local = await this.#getAccessMeta();
    if (
      uid === null ||
      local.hasParent ||
      local.access !== "private" ||
      local.acl === null ||
      typeof local.acl !== "object" ||
      !("owner" in local.acl) ||
      local.acl.owner !== uid
    ) {
      return plainResponse("Forbidden", 403);
    }

    let raw: unknown;
    try {
      raw = await request.json();
    } catch {
      return plainResponse("backfill body must be valid JSON", 400);
    }
    if (raw === null || typeof raw !== "object") {
      return plainResponse("backfill body must be an object", 400);
    }
    const children = "children" in raw ? raw.children : null;
    const approved = "approved" in raw ? raw.approved : false;
    const dryRun = "dryRun" in raw ? raw.dryRun : false;
    if (
      !Array.isArray(children) ||
      !children.every((child) => typeof child === "string") ||
      approved !== true ||
      typeof dryRun !== "boolean"
    ) {
      return plainResponse("backfill body must be {children,approved:true,dryRun?:boolean}", 400);
    }

    const ss = await this.#getSpreadsheet(uid);
    const tocLinks = new Set(
      parseCSV(ss.exportCSV())
        .slice(1)
        .map((row) => row[0])
        .filter((link): link is string => typeof link === "string"),
    );
    const eligible: string[] = [];
    const skipped: string[] = [];
    for (const child of children) {
      if (this.#isManagedChildName(roomName, child) && tocLinks.has(`/${child}`)) {
        eligible.push(child);
      } else {
        skipped.push(child);
      }
    }
    return jsonResponse({
      parent: roomName,
      dryRun,
      children: eligible,
      skipped,
    });
  }

  /**
   * Resolve or schedule a SQLite DO PITR bookmark. A successful restore is
   * applied only after this instance restarts, so return the target + undo
   * bookmark first and abort on a short timer.
   */
  async #postPitrRestore(request: Request): Promise<Response> {
    let body: unknown;
    try {
      body = await request.json();
    } catch {
      return plainResponse("body must be valid JSON", 400);
    }
    const parsed = parsePitrRequest(body);
    if (!parsed.ok) return plainResponse(parsed.error, 400);
    const accessMeta = await this.#getAccessMeta();
    if (accessMeta.hasParent || accessMeta.access === "private") {
      return plainResponse("PITR restore is not supported for protected room metadata", 409);
    }

    const storage = bookmarkStorage(this.#state.storage);
    if (!storage) {
      return plainResponse("PITR is unavailable on this deployment", 501);
    }

    let bookmark: string;
    try {
      if ("at" in parsed.value) {
        bookmark = await storage.getBookmarkForTime(parsed.value.at);
      } else {
        bookmark = parsed.value.bookmark;
        if (parsed.value.dryRun) await storage.getBookmarkForTime(Date.now());
      }
    } catch (error) {
      if (isPitrUnavailableError(error)) {
        return plainResponse("PITR is unavailable on this deployment", 501);
      }
      return plainResponse("PITR target is unavailable", 400);
    }

    if (parsed.value.dryRun) {
      return jsonResponse({ dryRun: true, bookmark });
    }

    let undoBookmark: string;
    try {
      undoBookmark = await storage.onNextSessionRestoreBookmark(bookmark);
    } catch (error) {
      if (isPitrUnavailableError(error)) {
        return plainResponse("PITR is unavailable on this deployment", 501);
      }
      return plainResponse("PITR target is unavailable", 400);
    }

    const { promise, resolve } = Promise.withResolvers<void>();
    this.#state.waitUntil(promise);
    setTimeout(() => {
      resolve();
      this.#state.abort("PITR restore scheduled");
    }, PITR_ABORT_DELAY_MS);
    return jsonResponse({
      bookmark,
      undoBookmark,
      nonce: this.#instanceNonce,
    });
  }

  /**
   * Rebuild metadata that lives outside the restored timeline. This endpoint
   * is called only after the public route observes a replacement instance.
   */
  async #postPitrTouch(roomName: string | null): Promise<Response> {
    if (!(await hasSnapshot(this.#state.storage))) {
      await this.#state.storage.deleteAlarm();
      this.#alarmArmed = false;
      await this.#deleteIndex(roomName);
      return jsonResponse({ exists: false });
    }
    const updatedAt = Date.now();
    await this.#state.storage.put(STORAGE_KEYS.metaUpdatedAt, updatedAt);
    const accessMeta = await this.#getAccessMeta();
    if (accessMeta.hasParent) await this.#deleteIndex(roomName);
    else {
      const access = await this.#effectiveAccess(null);
      if (access.isPrivate || !access.canRead) await this.#deleteIndex(roomName);
      else await this.#mirrorIndex(roomName, updatedAt);
    }
    await this.#armAlarm();
    return jsonResponse({ exists: true, updatedAt });
  }

  async #getSnapshot(): Promise<Response> {
    // Fast path: single-key. Small snapshots stay materialized (they
    // fit under CF's 96 MB DO-response body ceiling with plenty of
    // headroom).
    const single = await this.#state.storage.get<string>(STORAGE_KEYS.snapshot);
    if (typeof single === "string") return plainResponse(single);
    const meta = await readSnapshotMeta(this.#state.storage);
    if (meta === null) return notFound();
    // Chunked path: stream the reassembled save. Materializing a 148 MB
    // string into a Response body hits workerd's DO-response-size limit
    // (empirically ~96 MB on the paid plan); a streamed body bypasses
    // that because the response is transferred frame-by-frame.
    const storage = this.#state.storage;
    const encoder = new TextEncoder();
    const total = meta.chunks;
    let i = 0;
    const stream = new ReadableStream<Uint8Array>({
      async pull(controller) {
        if (i >= total) {
          controller.close();
          return;
        }
        const part = await storage.get<string>(snapshotChunkKey(i));
        if (typeof part !== "string") {
          controller.error(new Error(`snapshot chunk ${i} missing`));
          return;
        }
        controller.enqueue(encoder.encode(part));
        i += 1;
      },
    });
    return new Response(stream, {
      status: 200,
      headers: { "Content-Type": PLAIN_TEXT },
    });
  }

  async #putSnapshot(request: Request, roomName: string | null): Promise<Response> {
    const body = await request.text();
    let updatedAt = 0;
    if (!isSnapshotWithinSheetLimits(body)) {
      return plainResponse("snapshot exceeds sheet limits", 413);
    }
    await this.#state.blockConcurrencyWhile(async () => {
      const accessEntries = await this.#readAccessEntries();
      updatedAt = Date.now();
      await replaceStorageEntriesBatched(this.#state.storage, {
        ...accessEntries,
        ...snapshotEntries(body),
        [STORAGE_KEYS.metaUpdatedAt]: updatedAt,
      });
      this.#ss = null;
      this.#nextLogSeq = 0;
      this.#nextAuditSeq = 0;
      this.#nextChatSeq = 0;
      this.#resetVolatile();
    });
    await this.#mirrorIndex(roomName, updatedAt);
    // Keep housekeeping active so a room created/replaced via PUT and never
    // subsequently edited still gets TTL expiry.
    await this.#armAlarm();
    return plainResponse("OK", 201);
  }

  async #getLog(): Promise<Response> {
    const [log, chat] = await Promise.all([
      this.#listPrefix(STORAGE_KEYS.logPrefix),
      this.#listPrefix(STORAGE_KEYS.chatPrefix),
    ]);
    return jsonResponse({ log, chat });
  }

  async #postCommands(request: Request, roomName: string | null): Promise<Response> {
    const body = await request.text();
    if (!body) return plainResponse("", 202);
    const applied = await this.#applyCommandAndMirror(
      roomName,
      body,
      request.headers.get("X-EC-Uid"),
    );
    if (!applied) return plainResponse("command exceeds sheet limits", 413);
    // HTTP commands have no originating socket to exclude. Fan out only
    // after persistence succeeds; native WS commands use handleExecute's
    // separate broadcast path and never enter #postCommands.
    const broadcastRoom = roomName ?? this.#ownName;
    if (broadcastRoom) {
      await this.#broadcastAll({
        type: "execute",
        room: broadcastRoom,
        user: "",
        cmdstr: body,
      });
    }
    return plainResponse("", 202);
  }

  /**
   * Two-phase append reservation. Reserve assigns collision-free child ids
   * without publishing TOC rows; callers initialize every child, then commit.
   * A failed child therefore cannot leave the parent pointing at public data.
   */
  async #postImportAppendToc(request: Request, roomName: string | null): Promise<Response> {
    if (!roomName) return plainResponse("room name required", 400);
    const uid = request.headers.get("X-EC-Uid");
    let raw: unknown;
    try {
      raw = await request.json();
    } catch {
      return plainResponse("invalid JSON body", 400);
    }
    if (raw === null || typeof raw !== "object" || !("phase" in raw)) {
      return plainResponse("phase required", 400);
    }
    const phase = raw.phase;
    const requestId = "requestId" in raw ? raw.requestId : null;
    if (
      typeof requestId !== "string" ||
      requestId.length === 0 ||
      requestId.length > 128 ||
      !/^[A-Za-z0-9_-]+$/.test(requestId)
    ) {
      return plainResponse("valid requestId required", 400);
    }
    const reservationKey = `${MULTI_RESERVATION_PREFIX}${requestId}`;
    type ReservedSheet = { subroom: string; link: string; title: string };
    type Reservation = {
      uid: string | null;
      createdAt: number;
      firstIndex: number;
      requestedTitles: string[];
      tocStartRow?: number;
      committedAt?: number;
      sheets: ReservedSheet[];
    };

    if (phase === "reserve") {
      const requested = "sheets" in raw && Array.isArray(raw.sheets) ? raw.sheets : null;
      if (requested === null || requested.length === 0) {
        return plainResponse("sheets array required", 400);
      }
      const titles: string[] = [];
      for (const sheet of requested) {
        if (
          sheet === null ||
          typeof sheet !== "object" ||
          !("title" in sheet) ||
          typeof sheet.title !== "string"
        ) {
          return plainResponse("sheets array required", 400);
        }
        titles.push(sheet.title.slice(0, MAX_MULTI_SHEET_TITLE_LENGTH));
      }
      if (titles.length > MAX_MULTI_SHEETS) {
        return plainResponse(`workbook exceeds ${MAX_MULTI_SHEETS} sheets (${titles.length})`, 413);
      }
      const reserved = await this.#state.blockConcurrencyWhile(
        async (): Promise<
          { ok: true; reservation: Reservation } | { ok: false; status: number; message: string }
        > => {
          const existing = await this.#state.storage.get<Reservation>(reservationKey);
          if (existing) {
            if (
              existing.uid !== uid ||
              existing.requestedTitles.length !== titles.length ||
              existing.requestedTitles.some((title, index) => title !== titles[index])
            ) {
              return {
                ok: false,
                status: 409,
                message: "requestId already reserved",
              };
            }
            return { ok: true, reservation: existing };
          }

          const now = Date.now();
          const pending = await this.#state.storage.list<Reservation>({
            prefix: MULTI_RESERVATION_PREFIX,
          });
          const live: Reservation[] = [];
          for (const [key, reservation] of pending) {
            if (now - reservation.createdAt >= MULTI_RESERVATION_TTL_MS) {
              await this.#state.storage.delete(key);
            } else if (reservation.committedAt === undefined) {
              live.push(reservation);
            }
          }
          if (live.length >= MAX_MULTI_RESERVATIONS) {
            return {
              ok: false,
              status: 429,
              message: "too many pending sheet operations",
            };
          }

          const ss = await this.#getSpreadsheet(uid);
          const grid = parseCSV(ss.exportCSV());
          const existingLinks: string[] = [];
          const currentTitles: string[] = [];
          for (const row of grid.slice(1)) {
            const link = row[0];
            if (!link) continue;
            existingLinks.push(link);
            currentTitles.push(row[1] ?? "");
          }
          const pendingCount = live.reduce(
            (sum, reservation) => sum + reservation.sheets.length,
            0,
          );
          if (existingLinks.length + pendingCount + titles.length > MAX_MULTI_SHEETS) {
            return {
              ok: false,
              status: 413,
              message: `workbook exceeds ${MAX_MULTI_SHEETS} sheets`,
            };
          }
          const pendingLinks = live.flatMap((reservation) =>
            reservation.sheets.map((sheet) => sheet.link),
          );
          const inferredNext =
            getMaxSubsheetIndex([...existingLinks, ...pendingLinks], roomName) + 1;
          const storedNext = await this.#state.storage.get<number>(STORAGE_KEYS.metaNextChildIndex);
          const firstIndex =
            typeof storedNext === "number" && Number.isSafeInteger(storedNext) && storedNext > 0
              ? Math.max(storedNext, inferredNext)
              : inferredNext;
          const lastIndex = firstIndex + titles.length - 1;
          if (!Number.isSafeInteger(lastIndex) || !isValidRoomName(`${roomName}.${lastIndex}`)) {
            return {
              ok: false,
              status: 413,
              message: "child room namespace exhausted",
            };
          }
          for (const reservation of live) {
            currentTitles.push(...reservation.sheets.map((sheet) => sheet.title));
          }
          const sheets: ReservedSheet[] = [];
          for (let index = 0; index < titles.length; index++) {
            const childIndex = firstIndex + index;
            const preferred = titles[index]!.trim() || `Sheet${childIndex}`;
            let title = preferred;
            if (
              currentTitles
                .map((candidate) => candidate.toLowerCase())
                .includes(title.toLowerCase())
            ) {
              title = `${title}_${childIndex}`;
            }
            currentTitles.push(title);
            const subroom = `${roomName}.${childIndex}`;
            sheets.push({ subroom, link: `/${subroom}`, title });
          }
          const reservation: Reservation = {
            uid,
            createdAt: now,
            firstIndex,
            requestedTitles: titles,
            sheets,
          };
          await this.#state.storage.put({
            [reservationKey]: reservation,
            [STORAGE_KEYS.metaNextChildIndex]: firstIndex + titles.length,
          });
          return { ok: true, reservation };
        },
      );
      if (!reserved.ok) {
        return plainResponse(reserved.message, reserved.status);
      }
      return jsonResponse(
        {
          requestId,
          firstIndex: reserved.reservation.firstIndex,
          sheets: reserved.reservation.sheets,
        },
        201,
      );
    }

    if (phase !== "commit") return plainResponse("invalid phase", 400);
    const reserved = await this.#state.storage.get<Reservation>(reservationKey);
    if (!reserved) return plainResponse("reservation not found", 409);
    if (reserved.uid !== uid) return plainResponse("Forbidden", 403);

    const verifyHeaders = new Headers({ "Content-Type": "application/json" });
    if (uid !== null) verifyHeaders.set("X-EC-Uid", uid);
    for (const sheet of reserved.sheets) {
      const child = this.#env.ROOM.get(this.#env.ROOM.idFromName(encodeRoom(sheet.subroom)));
      let verified: Response;
      try {
        verified = await child.fetch(
          `https://do.local/_do/child-snapshot?name=${encodeURIComponent(sheet.subroom)}`,
          {
            method: "POST",
            headers: verifyHeaders,
            body: JSON.stringify({
              parent: roomName,
              snapshot: "",
              mode: "verify",
            }),
            signal: AbortSignal.timeout(5_000),
          },
        );
      } catch {
        return new Response("child verification failed", {
          status: 503,
          headers: { "Content-Type": PLAIN_TEXT, "Retry-After": "5" },
        });
      }
      if (verified.status === 501) return plainResponse("Not implemented", 501);
      if (verified.status !== 201) {
        return plainResponse(
          (await verified.text().catch(() => "child verification failed")) ||
            "child verification failed",
          verified.status >= 400 ? verified.status : 500,
        );
      }
    }

    const committed = await this.#state.blockConcurrencyWhile(async () => {
      const reservation = await this.#state.storage.get<Reservation>(reservationKey);
      if (!reservation) return { kind: "missing" } as const;
      if (reservation.uid !== uid) return { kind: "forbidden" } as const;
      if (reservation.committedAt !== undefined) {
        return { kind: "already", reservation } as const;
      }
      const ss = await this.#getSpreadsheet(uid);
      const grid = parseCSV(ss.exportCSV());
      const hasTocContent = grid.some((row) => row.some((cell) => cell !== ""));
      const tocStartRow = hasTocContent ? grid.length + 1 : 2;
      const csvRows = reservation.sheets.map(
        ({ link, title }) => `"${link.replace(/"/g, '""')}","${title.replace(/"/g, '""')}"`,
      );
      if (!hasTocContent) csvRows.unshift('"#url","#title"');
      const loadcmd = workbookToLoadClipboardCommand(new TextEncoder().encode(csvRows.join("\n")));
      if (!loadcmd) return { kind: "too-large" } as const;
      const commandText = `${loadcmd}\npaste A${hasTocContent ? grid.length + 1 : 1} all`;
      const applied = await this.#appendCommand(commandText, uid, {
        [reservationKey]: {
          ...reservation,
          committedAt: Date.now(),
          tocStartRow,
        } satisfies Reservation,
      });
      if (applied === null) return { kind: "too-large" } as const;
      return { kind: "applied", commandText, applied, tocStartRow } as const;
    });
    if (committed.kind === "missing") {
      return plainResponse("reservation not found", 409);
    }
    if (committed.kind === "forbidden") return plainResponse("Forbidden", 403);
    if (committed.kind === "too-large") {
      return plainResponse("command exceeds sheet limits", 413);
    }
    if (committed.kind === "applied") {
      await this.#mirrorIndex(roomName, committed.applied.ts);
      await this.#mirrorAudit(roomName, [
        {
          seq: committed.applied.auditSeq,
          ts: committed.applied.ts,
          body: committed.applied.auditBody,
        },
      ]);
      await this.#armAlarm();
      await this.#broadcastAll({
        type: "execute",
        room: roomName,
        user: "",
        cmdstr: committed.commandText,
      });
    }
    const tocStartRow =
      committed.kind === "applied" ? committed.tocStartRow : committed.reservation.tocStartRow;
    if (
      typeof tocStartRow !== "number" ||
      !Number.isSafeInteger(tocStartRow) ||
      tocStartRow < 2
    ) {
      return plainResponse("reservation row is unavailable", 500);
    }
    return jsonResponse(
      {
        requestId,
        firstIndex: reserved.firstIndex,
        sheets: reserved.sheets.map((sheet, index) => ({
          ...sheet,
          row: tocStartRow + index,
        })),
      },
      201,
    );
  }

  /**
   * Apply a command batch + mirror to D1. Shared between the HTTP path
   * (`POST /_do/commands`) and the WS path (`execute` frame). Centralizing
   * this ensures both paths update the `rooms` index; without mirroring on
   * the WS path, `/_rooms` and `/_roomtimes` go stale whenever a browser
   * client edits a fresh room (found during 2026-04-20 browser smoke).
   */
  async #applyCommandAndMirror(
    roomName: string | null,
    cmdstr: string,
    uid: string | null,
  ): Promise<boolean> {
    const applied = await this.#state.blockConcurrencyWhile(() => this.#appendCommand(cmdstr, uid));
    if (applied === null) return false;
    await this.#mirrorIndex(roomName, applied.ts);
    // Offload the audit entry to D1 (the durable record) so the alarm's DO
    // audit-trim doesn't lose it. Best-effort, outside the lock.
    await this.#mirrorAudit(roomName, [
      {
        seq: applied.auditSeq,
        ts: applied.ts,
        body: applied.auditBody,
      },
    ]);
    if (applied.auditSeq >= AUDIT_HISTORY_KEEP) {
      await this.#state.storage.delete(auditKey(applied.auditSeq - AUDIT_HISTORY_KEEP));
    }
    return true;
  }

  async #deleteAll(roomName: string | null, uid: string | null): Promise<Response> {
    await this.#deleteAllAndUnindex(roomName, uid);
    return plainResponse("OK", 201);
  }

  /**
   * Wipe the entire room + delete its D1 index row. Shared between the
   * HTTP path (`DELETE /_do/all`) and the WS path (`stopHuddle` frame).
   * Centralizing this ensures both paths drop the D1 row; without it,
   * `/_rooms` kept listing rooms that had been stopHuddle'd through the
   * WS (discovered during 2026-04-20 browser smoke).
   *
   * The local private ACL or immutable parent marker is retained as a
   * permanent tombstone so neither a protected room nor a child can be
   * re-created as public. `uid` is forwarded to the formdata sibling so a
   * private sibling accepts the cascade.
   */
  async #deleteAllAndUnindex(roomName: string | null, uid: string | null): Promise<void> {
    await this.#state.blockConcurrencyWhile(async () => {
      const accessEntries = await this.#readAccessEntries();
      const keptProtectedTombstone =
        accessEntries[STORAGE_KEYS.metaAccess] === "private" ||
        typeof accessEntries[STORAGE_KEYS.metaParent] === "string";
      await this.#state.storage.deleteAll();
      if (keptProtectedTombstone) {
        accessEntries[STORAGE_KEYS.metaUpdatedAt] = Date.now();
        await this.#state.storage.put(accessEntries);
      }
      this.#ss = null;
      this.#nextLogSeq = 0;
      this.#nextAuditSeq = 0;
      this.#nextChatSeq = 0;
      this.#resetVolatile();
    });
    // Protected tombstones are permanent: reclaiming a private parent's ACL
    // would declassify every surviving child, and reclaiming a child's parent
    // marker would make a surviving TOC reopen it as standalone public.
    await this.#deleteIndex(roomName);
    await this.#deleteAuditChatFromD1(roomName);
    await this.#deleteFormdataSibling(roomName, uid);
  }

  /**
   * Best-effort wipe of the submitform sibling `<room>_formdata` DO when
   * the main room is deleted. Skips when `roomName` is already a form-data
   * sibling or missing (issue #442).
   */
  async #deleteFormdataSibling(roomName: string | null, uid: string | null): Promise<void> {
    if (!roomName) return;
    const sibling = formdataSiblingRoom(roomName);
    if (!sibling) return;
    try {
      const id = this.#env.ROOM.idFromName(encodeRoom(sibling));
      const stub = this.#env.ROOM.get(id);
      await stub.fetch(`https://do.local/_do/all?name=${encodeURIComponent(sibling)}`, {
        method: "DELETE",
        ...(uid === null ? {} : { headers: { "X-EC-Uid": uid } }),
      });
    } catch {
      // Sibling may not exist; legacy delete was silent on orphans too.
    }
  }

  async #getExists(): Promise<Response> {
    return jsonResponse({
      exists: (await hasSnapshot(this.#state.storage)) ? 1 : 0,
    });
  }

  /**
   * Classify the default-room entry without sending its full snapshot or CSV
   * through the Worker. Legacy TOCs put `#url/#title` in row 1; workbooks
   * initialized through the current cold-room CSV path put the same header in
   * row 2 because that compatibility path pastes at A2.
   */
  async #getWorkbookKind(uid: string | null): Promise<Response> {
    if (!(await hasSnapshot(this.#state.storage))) {
      return jsonResponse({ kind: "absent" });
    }
    const ss = await this.#getSpreadsheet(uid);
    const a1 = (ss.exportCell("A1") as { readonly datavalue?: unknown } | null | undefined)
      ?.datavalue;
    const b1 = (ss.exportCell("B1") as { readonly datavalue?: unknown } | null | undefined)
      ?.datavalue;
    const a2 = (ss.exportCell("A2") as { readonly datavalue?: unknown } | null | undefined)
      ?.datavalue;
    const b2 = (ss.exportCell("B2") as { readonly datavalue?: unknown } | null | undefined)
      ?.datavalue;
    const isMulti = (a1 === "#url" && b1 === "#title") || (a2 === "#url" && b2 === "#title");
    return jsonResponse({ kind: isMulti ? "multi" : "single" });
  }

  async #getCells(uid: string | null): Promise<Response> {
    const ss = await this.#getSpreadsheet(uid);
    // Legacy (src/sc.ls:361): `JSON.stringify(window.ss.sheet.cells)`.
    // Unwrapped — not `{cells: ...}`. External clients parse the map
    // directly as `response.A1.datavalue`, etc.
    return jsonResponse(ss.exportCells());
  }

  async #getCell(coord: string, uid: string | null): Promise<Response> {
    const ss = await this.#getSpreadsheet(uid);
    return jsonResponse(ss.exportCell(coord));
  }

  // ─── Export handlers (Phase 8) ────────────────────────────────────────
  //
  // Every export derives from the in-memory `HeadlessSpreadsheet` — no
  // caching beyond what the SocialCalc instance itself does. That keeps
  // each GET deterministic after any mutation (POST /_do/commands).

  async #getHtml(uid: string | null): Promise<Response> {
    const ss = await this.#getSpreadsheet(uid);
    return textResponse(ss.createSheetHTML(), TEXT_HTML);
  }

  async #getCsv(uid: string | null): Promise<Response> {
    const ss = await this.#getSpreadsheet(uid);
    // The `.csv` download is opened in desktop spreadsheet apps, so defang
    // formula/DDE injection (`=`, `+`, `-`, `@`) before emitting. csv.json
    // stays faithful — it's consumed as JSON, not evaluated as a formula.
    return textResponse(neutralizeCSVDocument(ss.exportCSV()), TEXT_CSV);
  }

  async #getCsvJson(uid: string | null): Promise<Response> {
    const ss = await this.#getSpreadsheet(uid);
    return jsonResponse(parseCSV(ss.exportCSV()));
  }

  async #getMd(uid: string | null): Promise<Response> {
    const ss = await this.#getSpreadsheet(uid);
    return textResponse(csvToMarkdown(ss.exportCSV()), TEXT_MARKDOWN);
  }

  async #getBinary(format: BinaryFormat, uid: string | null): Promise<Response> {
    const ss = await this.#getSpreadsheet(uid);
    // Walk the raw SocialCalc sheet rather than going through CSV — that
    // preserves formulas, number formats, merges, and comments. See
    // `sheetViewToWorksheet` for the graceful-degrade-to-value rules.
    const bytes = sheetViewToBinaryWorkbook(ss.exportSheetData(), format);
    return binaryResponse(bytes, BINARY_CONTENT_TYPES[format]);
  }

  async #getSheetData(uid: string | null): Promise<Response> {
    const ss = await this.#getSpreadsheet(uid);
    return jsonResponse(ss.exportSheetData());
  }

  // ─── Rename primitives (Phase 6) ─────────────────────────────────────
  //
  // Legacy `set A\d+:B\d+ empty multi-cascade` (src/main.ls:425-436)
  // renamed the Redis keys `snapshot-<from>` -> `snapshot-<from>.bak`
  // plus the `log-*` and `audit-*` siblings, then re-ran the command.
  // In the DO world each "room" IS its own DO, so the equivalent is a
  // cross-DO state transfer orchestrated by the source DO.
  //
  // Design:
  //   POST /_do/rename body {to}  -- runs on source, dumps own
  //     snapshot/log/audit, fetches target's POST /_do/install with
  //     those as JSON, then deleteAll's own storage.
  //   POST /_do/install body {snapshot, log, audit}  -- wipes own
  //     storage and installs the payload verbatim.
  //
  // Chat and ecell are NOT carried over (legacy kept those under
  // different Redis prefixes so they stayed with the original room's
  // logical identity).

  async #postRename(request: Request): Promise<Response> {
    const parsed = (await request.json()) as { to?: unknown };
    const to = parsed.to;
    if (!isValidRoomName(to)) {
      return new Response("rename body must be {to: string}", { status: 400 });
    }
    const access = await this.#effectiveAccess(request.headers.get("X-EC-Uid"));
    if (access.parented || access.isPrivate || !access.canRead || !access.canWrite) {
      return new Response("Private room rename is not supported", {
        status: 409,
      });
    }
    const [snapshot, log, audit] = await Promise.all([
      readSnapshot(this.#state.storage),
      this.#listPrefix(STORAGE_KEYS.logPrefix),
      this.#listPrefix(STORAGE_KEYS.auditPrefix),
    ]);
    if (snapshot === null) {
      // No-op: legacy `if snapshot` guard at main.ls:427 -- nothing to rename.
      return new Response(null, { status: 204 });
    }
    const targetStub = this.#env.ROOM.get(this.#env.ROOM.idFromName(encodeRoom(to)));
    const installRes = await targetStub.fetch("https://do.local/_do/install", {
      method: "POST",
      body: JSON.stringify({ snapshot, log, audit }),
      headers: { "Content-Type": "application/json" },
    });
    if (!installRes.ok) {
      return new Response(`install failed: ${installRes.status}`, { status: 502 });
    }
    await this.#state.blockConcurrencyWhile(async () => {
      await this.#state.storage.deleteAll();
      this.#ss = null;
      this.#nextLogSeq = 0;
      this.#nextAuditSeq = 0;
      this.#nextChatSeq = 0;
      this.#resetVolatile();
    });
    return plainResponse("OK", 201);
  }

  /**
   * Snapshot-only DO-to-DO copy for `GET /:template/form` (legacy
   * main.ls:287-293). Leaves the source room intact.
   */
  async #postClone(request: Request): Promise<Response> {
    const parsed = (await request.json()) as { to?: unknown };
    const to = parsed.to;
    if (!isValidRoomName(to)) {
      return new Response("clone body must be {to: string}", { status: 400 });
    }
    // A clone copies this room's snapshot into a PUBLIC target — on a
    // private source that would declassify content (a writer-only
    // member could exfiltrate a sheet they cannot read). Same Phase A
    // stopgap as rename: refuse outright.
    const access = await this.#effectiveAccess(request.headers.get("X-EC-Uid"));
    if (access.isPrivate || !access.canRead || !access.canWrite) {
      return new Response("Private room clone is not supported", {
        status: 409,
      });
    }
    const snapshot = await readSnapshot(this.#state.storage);
    const targetId = this.#env.ROOM.idFromName(encodeRoom(to));
    const targetStub = this.#env.ROOM.get(targetId);
    const putRes = await targetStub.fetch(
      `https://do.local/_do/snapshot?name=${encodeURIComponent(to)}`,
      { method: "PUT", body: snapshot ?? "" },
    );
    if (!putRes.ok) {
      return new Response(`clone failed: ${putRes.status}`, { status: 502 });
    }
    return plainResponse("OK", 201);
  }

  /**
   * Phase 11b — full-fidelity room seed. Accepts the complete payload
   * derived from a legacy Redis dump (see `@ethercalc/migrate`) and
   * installs it verbatim, replacing any existing state in this DO.
   *
   * Differences from `#postInstall` (rename path):
   *   - carries chat + ecell + explicit updatedAt
   *   - mirrors the D1 `rooms` row via `?name=<room>`
   *   - snapshot field is optional; a log-only room (legacy rooms with
   *     commands recorded but no snapshot yet folded) seeds with an
   *     empty snapshot and no `snapshot` storage key.
   *
   * Idempotent — re-running against the same room overwrites. The
   * migrator calls this exactly once per room per run.
   */
  async #postSeed(request: Request, roomName: string | null): Promise<Response> {
    let raw: unknown;
    try {
      raw = await request.json();
    } catch {
      return new Response("seed body must be valid JSON", { status: 400 });
    }
    const parsed = parseSeedPayload(raw, () => Date.now());
    if (!parsed.ok) {
      return new Response(parsed.error, { status: 400 });
    }
    const payload = parsed.value;
    // Fold base+log into one authoritative snapshot on ingest. Since
    // `#getSpreadsheet` no longer replays the log over a present
    // snapshot, a seeded room that carried a base snapshot + since-base
    // log would otherwise read back as just the base (log silently
    // dropped) — or, under the old double-apply hydrate, with every log
    // command applied twice. Folding here makes the stored snapshot the
    // single source of truth.
    //
    // Two shapes survive folding:
    //   - base snapshot present (with or without log) → fold to a save.
    //   - log-only room (no base snapshot) → fold the log onto an empty
    //     sheet, producing a real snapshot. This collapses the legacy
    //     "commands but never folded" rooms into a normal snapshot room.
    //   - neither → empty room; keep the "no snapshot" shape.
    const hasState = payload.snapshot.length > 0 || payload.log.length > 0;
    const foldedSnapshot = hasState ? foldSnapshot(payload.snapshot, payload.log as string[]) : "";
    const logTail = (payload.log as string[]).slice(-LOG_RING);
    await this.#state.blockConcurrencyWhile(async () => {
      const accessEntries = await this.#readAccessEntries();
      const entries: Record<string, unknown> = {
        ...accessEntries,
        [STORAGE_KEYS.metaUpdatedAt]: payload.updatedAt,
      };
      // Chunk the snapshot when it exceeds the DO-storage 128 KiB
      // per-value ceiling. `snapshotEntries` returns either `{snapshot:
      // …}` (fast path) or `{snapshot:meta:{chunks}, snapshot:chunk:<i>:
      // …}` (>100 KiB, split). Skipped entirely for empty snapshots so
      // truly empty rooms keep the "no snapshot" shape.
      if (foldedSnapshot.length > 0) {
        Object.assign(entries, snapshotEntries(foldedSnapshot));
      }
      // Keep only the bounded recent log tail for client catch-up — the
      // folded snapshot already incorporates the full log.
      for (let i = 0; i < logTail.length; i++) {
        entries[logKey(i)] = logTail[i] as string;
      }
      for (let i = 0; i < payload.audit.length; i++) {
        entries[auditKey(i)] = payload.audit[i] as string;
      }
      for (let i = 0; i < payload.chat.length; i++) {
        entries[chatKey(i)] = payload.chat[i] as string;
      }
      for (const [user, cell] of Object.entries(payload.ecell)) {
        entries[ecellKey(user)] = cell;
      }

      await replaceStorageEntriesBatched(this.#state.storage, entries);
      this.#ss = null;
      this.#nextLogSeq = logTail.length;
      this.#nextAuditSeq = payload.audit.length;
      this.#nextChatSeq = payload.chat.length;
      this.#resetVolatile();
    });
    await this.#armAlarm();
    // The D1 mirror is a cross-DO write that happens on every seed. Two
    // opt-outs, from the caller's perspective:
    //   - `payload.skipIndex=true` — the caller plans to batch index
    //     writes via `PUT /_migrate/bulk-index`. Don't touch D1 at all.
    //   - default (skipIndex=false) — fire-and-forget via `waitUntil`
    //     so this 201 returns as soon as DO storage is durable; the D1
    //     write drains on the DO's background execution context.
    // At 1.8M rooms the second option alone cuts ~50 ms off the critical
    // path of each PUT, and the batched path cuts the D1 chokepoint by
    // 100× (see AGENTS.md §14 2026-04-21).
    if (!payload.skipIndex) {
      this.#state.waitUntil(this.#mirrorIndex(roomName, payload.updatedAt));
    }
    // Mirror the seeded audit + chat into the durable D1 stores so a migrated
    // room's history survives the DO-tail trims. Done regardless of skipIndex
    // (which only governs the rooms index) — fire-and-forget. Skipped when
    // empty (the common dir-migration case keeps log/chat in-memory only).
    if (payload.audit.length > 0) {
      this.#state.waitUntil(
        this.#mirrorAudit(
          roomName,
          payload.audit.map((body, i) => ({
            seq: i,
            ts: payload.updatedAt,
            body: body as string,
          })),
        ),
      );
    }
    if (payload.chat.length > 0) {
      this.#state.waitUntil(
        this.#mirrorChat(
          roomName,
          payload.chat.map((body, i) => ({
            seq: i,
            ts: payload.updatedAt,
            body: body as string,
          })),
        ),
      );
    }
    return plainResponse("OK", 201);
  }

  /**
   * Phase 11b — client-side chunked snapshot upload.
   *
   * Contract (enforced by `routes/migrate.ts` before we ever see the
   * request, re-checked here for defense-in-depth and because unit
   * tests construct the DO directly):
   *   - `seq` and `chunks` are integers, `0 ≤ seq < chunks`, `chunks ≥ 1`.
   *   - Body is the verbatim chunk payload (plain UTF-8 text).
   *
   * Flow when called N times in order:
   *   1. seq=0..N-2 → land `snapshot:chunk:<padSeq(seq)>`. Meta is NOT
   *      written yet; readers still see whatever snapshot (if any) was
   *      there before the upload started.
   *   2. seq=N-1 → batched put of the final chunk + `snapshot:meta =
   *      {chunks: N}` + `meta:updated_at`. Atomically flips the DO's
   *      "current snapshot" over to the new chunked layout. Then
   *      cleans up any legacy single-key `snapshot` from a prior seed
   *      and any higher-seq chunks from a prior larger chunked save.
   *      Finally mirrors the D1 `rooms` row so cross-room indexes pick
   *      up the new `updated_at`.
   *
   * Re-migrating the same room: safe. The stale-cleanup step drops
   * both layouts of any leftover snapshot state. We don't reset
   * `#nextLogSeq` / `#nextAuditSeq` / `#nextChatSeq` — those counters
   * track DO-local append sequence and are unrelated to the snapshot
   * body itself.
   */
  async #postSnapshotChunk(
    request: Request,
    roomName: string | null,
    searchParams: URLSearchParams,
  ): Promise<Response> {
    const seqRaw = searchParams.get("seq");
    const chunksRaw = searchParams.get("chunks");
    // `URLSearchParams.get` returns `null` for missing params;
    // `Number(null) === 0` would otherwise slip through the range check.
    const seq = seqRaw === null ? NaN : Number(seqRaw);
    const chunks = chunksRaw === null ? NaN : Number(chunksRaw);
    if (
      !Number.isInteger(seq) ||
      seq < 0 ||
      !Number.isInteger(chunks) ||
      chunks < 1 ||
      chunks > MAX_SNAPSHOT_CHUNKS ||
      seq >= chunks
    ) {
      return new Response("seq/chunks must be integers with 0 ≤ seq < chunks", {
        status: 400,
      });
    }
    const bytes = await request.arrayBuffer();
    if (bytes.byteLength > SNAPSHOT_CHUNK_BYTES) {
      return plainResponse("snapshot chunk exceeds 100 KiB", 413);
    }
    const body = new TextDecoder().decode(bytes);
    const isFinal = seq === chunks - 1;
    let updatedAt = 0;
    await this.#state.blockConcurrencyWhile(async () => {
      if (!isFinal) {
        await this.#state.storage.put(snapshotChunkKey(seq), body);
        return;
      }
      // Read the prior meta BEFORE we overwrite it, so we know which
      // higher-seq chunks (if any) are stale from a larger previous
      // chunked save. A prior single-key `snapshot` is always cleaned
      // up regardless — either it's leftover from a pre-chunked seed
      // or it's absent, both fine.
      const priorMeta = await readSnapshotMeta(this.#state.storage);
      updatedAt = Date.now();
      const stale: string[] = [STORAGE_KEYS.snapshot];
      if (priorMeta !== null) {
        for (let i = chunks; i < priorMeta.chunks; i++) {
          stale.push(snapshotChunkKey(i));
        }
      }
      await this.#state.storage.transaction(async (txn) => {
        await txn.put({
          [snapshotChunkKey(seq)]: body,
          [STORAGE_KEYS.snapshotMeta]: { chunks } satisfies SnapshotMeta,
          [STORAGE_KEYS.metaUpdatedAt]: updatedAt,
        });
        await deleteStorageKeysBatched(txn, stale);
      });
      // Next `#getSpreadsheet` will rehydrate from the reassembled save.
      this.#ss = null;
    });
    if (isFinal) {
      await this.#mirrorIndex(roomName, updatedAt);
    }
    return plainResponse("OK", 201);
  }

  async #postInstall(request: Request): Promise<Response> {
    const parsed = (await request.json()) as {
      snapshot?: unknown;
      log?: unknown;
      audit?: unknown;
    };
    if (typeof parsed.snapshot !== "string") {
      return new Response("install body.snapshot must be string", { status: 400 });
    }
    const log = Array.isArray(parsed.log) ? (parsed.log as unknown[]) : [];
    const audit = Array.isArray(parsed.audit) ? (parsed.audit as unknown[]) : [];
    if (!log.every((e) => typeof e === "string")) {
      return new Response("install body.log must be string[]", { status: 400 });
    }
    if (!audit.every((e) => typeof e === "string")) {
      return new Response("install body.audit must be string[]", { status: 400 });
    }
    // Fold base+log into a single authoritative snapshot on ingest.
    // `#getSpreadsheet` no longer replays the log when a snapshot is
    // present, so the snapshot we store here must already incorporate the
    // incoming `log` commands — otherwise a renamed room would silently
    // lose every since-base command. The bounded recent tail (≤ LOG_RING)
    // is kept only for client catch-up; audit carries the full record.
    const foldedSnapshot = foldSnapshot(parsed.snapshot as string, log as string[]);
    const logTail = (log as string[]).slice(-LOG_RING);
    await this.#state.blockConcurrencyWhile(async () => {
      const accessEntries = await this.#readAccessEntries();
      const entries: Record<string, unknown> = {
        ...accessEntries,
        ...snapshotEntries(foldedSnapshot),
        [STORAGE_KEYS.metaUpdatedAt]: Date.now(),
      };
      for (let i = 0; i < logTail.length; i++) {
        entries[logKey(i)] = logTail[i] as string;
      }
      for (let i = 0; i < audit.length; i++) {
        entries[auditKey(i)] = audit[i] as string;
      }
      await replaceStorageEntriesBatched(this.#state.storage, entries);
      this.#ss = null;
      this.#nextLogSeq = logTail.length;
      this.#nextAuditSeq = audit.length;
      this.#nextChatSeq = 0;
      this.#resetVolatile();
    });
    await this.#armAlarm();
    return plainResponse("OK", 201);
  }

  /**
   * Phase A — atomic private-room initialization. The one route that
   * creates `meta:access`/`meta:acl`/`meta:group`, and only on a DO
   * with ZERO storage keys — a tombstoned or occupied room 409s, so
   * emptiness (not authorization) is the claim guard. The caller's
   * trusted uid (`X-EC-Uid`, minted by the Worker from a verified
   * session) must own the supplied ACL so nobody can mint rooms owned
   * by someone else.
   */
  async #postInitPrivate(request: Request): Promise<Response> {
    const uid = request.headers.get("X-EC-Uid");
    if (uid === null || uid.length === 0) {
      return plainResponse("Forbidden", 403);
    }
    let raw: unknown;
    try {
      raw = await request.json();
    } catch {
      return plainResponse("init-private body must be valid JSON", 400);
    }
    if (raw === null || typeof raw !== "object") {
      return plainResponse("init-private body must be an object", 400);
    }
    const snapshot = "snapshot" in raw && typeof raw.snapshot === "string" ? raw.snapshot : null;
    if (snapshot === null) {
      return plainResponse("init-private body.snapshot must be string", 400);
    }
    if (!isSnapshotWithinSheetLimits(snapshot)) {
      return plainResponse("init-private snapshot exceeds sheet limits", 413);
    }
    const acl = "acl" in raw ? raw.acl : null;
    if (
      acl === null ||
      typeof acl !== "object" ||
      !("owner" in acl) ||
      typeof acl.owner !== "string" ||
      acl.owner.length === 0 ||
      !("readers" in acl) ||
      !Array.isArray(acl.readers) ||
      !acl.readers.every((r) => typeof r === "string" && r.length > 0) ||
      !("writers" in acl) ||
      !Array.isArray(acl.writers) ||
      !acl.writers.every((w) => typeof w === "string" && w.length > 0)
    ) {
      return plainResponse("init-private body.acl is malformed", 400);
    }
    const group = "group" in raw ? raw.group : undefined;
    if (group !== undefined && typeof group !== "string") {
      return plainResponse("init-private body.group must be string", 400);
    }
    if (acl.owner !== uid) {
      return plainResponse("Forbidden", 403);
    }
    const privateEntries = {
      ...snapshotEntries(snapshot),
      [STORAGE_KEYS.metaAccess]: "private",
      [STORAGE_KEYS.metaAcl]: {
        owner: acl.owner,
        readers: acl.readers,
        writers: acl.writers,
      },
      ...(group === undefined ? {} : { [STORAGE_KEYS.metaGroup]: group }),
      [STORAGE_KEYS.metaUpdatedAt]: Date.now(),
    };
    const created = await this.#state.blockConcurrencyWhile(() =>
      this.#state.storage.transaction(async (txn) => {
        const existing = await txn.list({ limit: 1 });
        if (existing.size > 0) return false;
        await putStorageEntriesBatched(txn, privateEntries);
        return true;
      }),
    );
    if (!created) {
      return plainResponse("Room already exists", 409);
    }
    this.#ss = null;
    this.#nextLogSeq = 0;
    this.#nextAuditSeq = 0;
    this.#nextChatSeq = 0;
    this.#resetVolatile();
    await this.#armAlarm();
    return plainResponse("OK", 201);
  }

  // ─── Cron fire-trigger (Phase 9) ───────────────────────────────────────

  /**
   * `POST /_do/fire-trigger?cell=<coord>` — fire one due cron trigger.
   *
   * Legacy flow (src/sc.ls:360-370 + src/main.ls:196 + src/sc.ls:247-253):
   *   - `SC[room].triggerActionCell(cell, cb)` ran
   *     `SocialCalc.TriggerIoAction.Email('<coord>')` against the
   *     cell, which produced a URL-encoded `sendemail <to> <subject>
   *     <body>` string and passed it back via `cb`.
   *   - The server parsed that string, dispatched to `emailer.sendemail`,
   *     and broadcast `{type: confirmemailsent, message}` on
   *     `log-<room>`.
   *
   * We collapse all of that here:
   *   1. Look up the cell's datavalue/formula/comment and derive a
   *      `sendemail` command. SocialCalc's TriggerIoAction logic
   *      stores the email payload directly in the cell's `datavalue`
   *      (as a space-delimited, %20-encoded string starting with
   *      `sendemail `). If the cell doesn't hold one, the trigger is
   *      a no-op.
   *   2. Parse via `parseSendemail`, dispatch through `buildEmailSender(env)`.
   *   3. Broadcast `{type: 'confirmemailsent', message}` to every WS peer.
   *
   * Every failure path is swallowed into a `200 OK` so the cron
   * runner never retries on a malformed cell — the legacy handler
   * also moved on.
   */
  async #fireTrigger(cell: string): Promise<Response> {
    if (cell.length === 0) return plainResponse("", 200);
    const ss = await this.#getSpreadsheet();
    const cellRecord = ss.exportCell(cell) as { datavalue?: unknown; formula?: unknown } | null;
    if (!cellRecord) return plainResponse("", 200);
    // Legacy's TriggerIoAction.Email reconstructs the command from
    // formula-like payload stored in the cell. In practice clients put
    // the full `sendemail <to> <subject> <body>` URL-encoded string
    // into `formula` (for triggered cells) or `datavalue` (for plain
    // text). Try both.
    const candidate =
      (typeof cellRecord.formula === "string" && cellRecord.formula.length > 0
        ? cellRecord.formula
        : "") || (typeof cellRecord.datavalue === "string" ? cellRecord.datavalue : "");
    const parsed = parseSendemail(candidate);
    if (!parsed) return plainResponse("", 200);
    const sender = buildEmailSender(this.#env);
    const { message } = await sender.send(parsed.to, parsed.subject, parsed.body);
    await this.#broadcastAll({ type: "confirmemailsent", message });
    return plainResponse("", 200);
  }

  // ─── WebSocket acceptance ──────────────────────────────────────────────

  /**
   * `GET /_do/ws?user=<user>&auth=<hmac>` — upgrade to WebSocket using the
   * hibernation API. The attachment binds the room/user plus any verified
   * session token so private-room frames can honor logout revocation.
   */
  #acceptWebSocket(request: Request): Response {
    if (request.headers.get("Upgrade") !== "websocket") {
      return plainResponse("Expected Upgrade: websocket", 426);
    }
    // Per-room connection cap — a coarse DoS backstop. The hibernation API
    // keeps every accepted socket retrievable via `getWebSockets()`, so a
    // simple count is the live connection total for this room.
    if (this.#state.getWebSockets().length >= MAX_CONN) {
      return plainResponse("Too many connections", 503);
    }
    /* istanbul ignore next -- @preserve
     *   `upgradeWebSocket` needs `WebSocketPair`, `state.acceptWebSocket`,
     *   and a Workers `Response` accepting `status: 101` + `webSocket`.
     *   None of these exist in Node; end-to-end coverage lives in the
     *   workers-pool integration tests (`test/ws.test.ts`,
     *   `test/legacy-socketio.test.ts`, `test/room.test.ts`).
     */
    const uid = request.headers.get("X-EC-Uid");
    const sessionExpHeader = request.headers.get("X-EC-Session-Exp");
    const session = request.headers.get("X-EC-Session");
    /* istanbul ignore next -- workerd-only WebSocket attachment wiring */
    const sessionExp = sessionExpHeader === null ? null : Number(sessionExpHeader);
    /* istanbul ignore next -- workerd-only WebSocket attachment wiring */
    const wsOpts = {
      ...(isSandstormEnforced(this.#env)
        ? { sandstormModify: sandstormCanModify(request.headers) }
        : {}),
      ...(uid === null ? {} : { uid }),
      ...(sessionExp === null || !Number.isFinite(sessionExp) ? {} : { sessionExp }),
      ...(session === null ? {} : { session }),
    };
    return upgradeWebSocket(this.#state, request, wsOpts);
  }

  /**
   * `GET /_do/legacy-ws` — hibernatable socket.io v0.9 upgrade. Same
   * connection cap as the native path; framing differs (see attachment
   * `legacy: true` + `#sendTo` / `webSocketMessage`).
   */
  #acceptLegacyWebSocket(request: Request): Response {
    if (request.headers.get("Upgrade") !== "websocket") {
      return plainResponse("Expected Upgrade: websocket", 426);
    }
    if (this.#state.getWebSockets().length >= MAX_CONN) {
      return plainResponse("Too many connections", 503);
    }
    /* istanbul ignore next -- @preserve
     *   Same workerd-only surface as `#acceptWebSocket` (WebSocketPair +
     *   acceptWebSocket + 101 Response). Covered by workers-pool tests.
     */
    return upgradeLegacySocketIo(this.#state);
  }

  /**
   * Hibernation-api entrypoint. Parses the incoming frame, assembles a
   * `WsContext` bound to this socket, and delegates to the pure dispatch
   * layer in `src/lib/ws-handlers.ts` (Phase 7.1 extract).
   *
   * Legacy (`attachment.legacy`) sockets speak socket.io v0.9 framing:
   * heartbeats are auto-answered by `setWebSocketAutoResponse`; event
   * packets are unwrapped to native ClientMessage before dispatch.
   * Session-host DOs (sid-keyed, empty attachment.room) only forward
   * `execute` to the room named in the frame — matching the pre-hibernate
   * Worker-shim baseline — so spreadsheet state stays on the real room DO.
   */
  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): Promise<void> {
    const attachment = (ws.deserializeAttachment() as WsAttachment | null) ?? {
      user: "",
      room: "",
      auth: "",
    };
    const messageSize = typeof message === "string" ? message.length : message.byteLength;
    if (messageSize > MAX_WS_FRAME_CHARS) {
      try {
        ws.close(1009, "Message too large");
      } catch {
        // The peer may already be gone.
      }
      return;
    }
    if (this.#rateLimitSocket(ws, attachment)) return;
    if (typeof message !== "string") return;
    if (await this.#closeUnauthorizedPrivateSocket(ws, attachment, "read-or-write")) {
      return;
    }
    if (attachment.legacy) {
      await this.#handleLegacyFrame(ws, attachment, message);
      return;
    }
    const parsed = parseClientMessage(message);
    if (!parsed || !attachment.room) return;
    // Every room-labelled frame stays on its accepted socket. `ask.recalc`
    // is the one legacy exception: its room names a cross-sheet reference,
    // while this DO supplies the current cached snapshot.
    if (parsed.type !== "ask.recalc" && parsed.room !== attachment.room) return;
    // The handshake defines the cosmetic user identity for this socket.
    // Canonical parsing produced a fresh object, so rebinding cannot mutate
    // caller-owned state.
    if ("user" in parsed) parsed.user = attachment.user;
    const perMessageAuth =
      "auth" in parsed && typeof parsed.auth === "string" ? parsed.auth : attachment.auth;
    const ctx = this.#buildWsContext(ws, attachment, attachment.room, perMessageAuth);
    await dispatchWsMessage(ctx, parsed);
  }

  #rateLimitSocket(ws: WebSocket, attachment: WsAttachment): boolean {
    const now = Date.now();
    const startedAt = attachment.rateWindowStartedAt;
    const inCurrentWindow =
      typeof startedAt === "number" &&
      Number.isFinite(startedAt) &&
      now >= startedAt &&
      now - startedAt < WS_RATE_WINDOW_MS;
    const priorCount =
      typeof attachment.rateMessageCount === "number" &&
      Number.isFinite(attachment.rateMessageCount) &&
      attachment.rateMessageCount >= 0
        ? attachment.rateMessageCount
        : 0;
    const rateMessageCount = inCurrentWindow ? priorCount + 1 : 1;
    const rateWindowStartedAt = inCurrentWindow ? startedAt : now;

    // A hibernated object can be reconstructed between frames. Recover the
    // aggregate window once from per-socket attachments so eviction cannot
    // reset the room-wide amplification limit.
    if (this.#roomRateWindowStartedAt === 0) {
      let restoredCount = 0;
      let restoredStartedAt = now;
      let sawCurrentSocket = false;
      for (const peer of this.#state.getWebSockets()) {
        if (peer === ws) sawCurrentSocket = true;
        let peerAttachment: WsAttachment | null = null;
        try {
          peerAttachment = peer.deserializeAttachment() as WsAttachment | null;
        } catch {
          continue;
        }
        const peerStartedAt = peerAttachment?.rateWindowStartedAt;
        const peerCount = peerAttachment?.rateMessageCount;
        if (
          typeof peerStartedAt === "number" &&
          Number.isFinite(peerStartedAt) &&
          now >= peerStartedAt &&
          now - peerStartedAt < WS_RATE_WINDOW_MS &&
          typeof peerCount === "number" &&
          Number.isFinite(peerCount) &&
          peerCount >= 0
        ) {
          restoredStartedAt = Math.min(restoredStartedAt, peerStartedAt);
          restoredCount += Math.min(peerCount, MAX_WS_MESSAGES_PER_WINDOW);
        }
      }
      if (!sawCurrentSocket && inCurrentWindow) {
        restoredStartedAt = Math.min(restoredStartedAt, rateWindowStartedAt);
        restoredCount += Math.min(priorCount, MAX_WS_MESSAGES_PER_WINDOW);
      }
      this.#roomRateWindowStartedAt = restoredStartedAt;
      this.#roomRateMessageCount = restoredCount;
    }

    const roomInCurrentWindow =
      now >= this.#roomRateWindowStartedAt &&
      now - this.#roomRateWindowStartedAt < WS_RATE_WINDOW_MS;
    if (roomInCurrentWindow) {
      this.#roomRateMessageCount += 1;
    } else {
      this.#roomRateWindowStartedAt = now;
      this.#roomRateMessageCount = 1;
    }
    try {
      ws.serializeAttachment({
        ...attachment,
        rateWindowStartedAt,
        rateMessageCount,
      } satisfies WsAttachment);
    } catch {
      try {
        ws.close(1011, "Attachment state unavailable");
      } catch {
        // The peer may already be gone.
      }
      return true;
    }
    const socketExceeded = rateMessageCount > MAX_WS_MESSAGES_PER_WINDOW;
    const roomExceeded = this.#roomRateMessageCount > MAX_ROOM_WS_MESSAGES_PER_WINDOW;
    if (!socketExceeded && !roomExceeded) return false;
    try {
      ws.close(1008, roomExceeded ? "Room message rate exceeded" : "Message rate exceeded");
    } catch {
      // The peer may already be gone.
    }
    return true;
  }

  /**
   * Decode one socket.io v0.9 frame and either dispatch it locally (when
   * this DO is the room the frame names) or forward `execute` to the
   * named room DO (session-host / sid-keyed case).
   */
  async #handleLegacyFrame(ws: WebSocket, attachment: WsAttachment, raw: string): Promise<void> {
    const packet = decodeFrame(raw);
    if (!packet) return;
    if (packet.type === PacketType.Disconnect) {
      try {
        ws.close(1000, "client disconnected");
      } catch {
        /* already closed */
      }
      return;
    }
    // Heartbeats are answered by setWebSocketAutoResponse without waking
    // us; if one does arrive (auto-response unset in tests), ignore it.
    if (packet.type === PacketType.Heartbeat) return;
    if (packet.type !== PacketType.Event) return;
    const parsed = socketIoEventToNative(packet);
    if (!parsed) return;

    // Session host: attachment.room is empty because the upgrade was
    // sid-keyed, not room-keyed. Forward executes to the real room DO
    // (preserves spreadsheet locality); other types need two-way state
    // and are dropped here just as the pre-hibernate baseline did.
    if (!attachment.room) {
      if (parsed.type !== "execute") return;
      if (isFilteredExecuteCommand(parsed.cmdstr)) return;
      const room = parsed.room;
      const auth = parsed.auth ?? attachment.auth;
      if (!(await verifyAuth(this.#env.ETHERCALC_KEY, room, auth))) return;
      try {
        const stub = this.#env.ROOM.get(this.#env.ROOM.idFromName(encodeRoom(room)));
        await stub.fetch(`https://do.local/_do/commands?name=${encodeURIComponent(room)}`, {
          method: "POST",
          body: parsed.cmdstr,
        });
      } catch {
        // Best-effort; a missing ROOM binding in unit tests no-ops.
      }
      return;
    }

    // Room-scoped legacy socket: full native dispatch with framing on send.
    // Same room-binding invariant as the native path. Cross-sheet
    // `ask.recalc` keeps its legacy target-room label.
    if (parsed.type !== "ask.recalc" && parsed.room !== attachment.room) return;
    if ("user" in parsed) parsed.user = attachment.user;
    const perMessageAuth =
      "auth" in parsed && typeof parsed.auth === "string" ? parsed.auth : attachment.auth;
    const ctx = this.#buildWsContext(ws, attachment, attachment.room, perMessageAuth);
    await dispatchWsMessage(ctx, parsed);
  }

  /**
   * Hibernation-api close hook. We intentionally do NOT remove the user's
   * `ecell:<user>` entry here — legacy left the last-known cursor in place
   * so a reconnecting client could resume where they left off.
   */
  async webSocketClose(ws: WebSocket): Promise<void> {
    void ws;
  }

  // ─── WsContext assembly ────────────────────────────────────────────────

  /**
   * Build the `WsContext` surface for one WS frame. Everything here is
   * deliberately small: captured closures over `this` that translate
   * handler calls into storage I/O, socket sends, and sibling-DO fetches.
   *
   * State-changing frames are room-bound before this method is called.
   * Read/poll message room fields remain decorative for legacy parity.
   */
  #buildWsContext(
    ws: WebSocket,
    attachment: WsAttachment,
    messageRoom: string,
    messageAuth: string,
  ): WsContext {
    const storage: WsStorage = {
      listPrefix: (prefix) => this.#listPrefix(prefix),
      listHash: (prefix) => this.#listHash(prefix),
      putHash: async (prefix, key, value) => {
        // ecell writes carry an attacker-controllable username key, so
        // they go through the LRU-capped path. Any other prefix (none
        // today) writes straight through.
        /* istanbul ignore else -- @preserve: only ecell: reaches putHash today */
        if (prefix === STORAGE_KEYS.ecellPrefix) {
          await this.#putEcellCapped(key, value);
        } else {
          await this.#state.storage.put(`${prefix}${key}`, value);
        }
      },
      appendLog: (prefix, value) => this.#appendLogEntry(prefix, value, attachment.room),
      getSnapshot: async () => (await readSnapshot(this.#state.storage)) ?? undefined,
      deleteAll: async () => {
        // WS `stopHuddle` is the hot path. Mirror the HTTP DELETE flow:
        // wipe storage AND drop the D1 index row so `/_rooms`,
        // `/_roomlinks`, and `/_roomtimes` stop listing the dead room.
        // Always the handshake room: a frame naming another room never
        // reaches dispatch (`webSocketMessage` drops it), so the
        // attacker-supplied `room` field can't unindex a third party.
        // The handler enforces auth before calling this, so we don't
        // need to re-verify here.
        await this.#deleteAllAndUnindex(attachment.room, attachment.uid ?? null);
      },
    };
    const env = this.#env;
    return {
      room: messageRoom,
      user: attachment.user,
      auth: messageAuth,
      storage,
      applyCommand: async (cmdstr: string) => {
        // Mirror the DO's own room (from the WS handshake attachment),
        // not the per-frame `room` field, because the append lands in
        // *this* DO's storage regardless of what room the frame names.
        const applied = await this.#applyCommandAndMirror(
          attachment.room,
          cmdstr,
          attachment.uid ?? null,
        );
        if (!applied) {
          try {
            ws.close(1008, "Command exceeds sheet limits");
          } catch {
            // The peer may already be gone.
          }
        }
        return applied;
      },
      broadcast: async (msg, includeSelf) => {
        if (includeSelf) await this.#broadcastAll(msg);
        else await this.#broadcast(ws, msg);
      },
      reply: async (msg) => {
        await this.#sendTo(ws, msg);
      },
      verifyAuth: async () => {
        // Sandstorm viewers lack `modify` — block WS writes (SH-6).
        if (!sandstormAllowsWsWrite(env, attachment.sandstormModify)) {
          return false;
        }
        // Legacy explicit view-only sentinel (`auth === '0'`) stays an
        // absolute veto on both public and private paths.
        if (messageAuth === "0") return false;
        const uid = attachment.uid ?? null;
        const access = await this.#effectiveAccess(uid);
        if (!access.canWrite) return false;
        if (access.isPrivate) {
          // Private access (local or parent-derived) replaces legacy HMAC.
          return true;
        }
        // Execute/ecell/stopHuddle carry their own per-message `auth`
        // string; legacy verified against that field (src/main.ls:516).
        // Messages without an auth field pass empty string → rejected by
        // `verifyAuth` as view-only unless KEY is unset (identity path).
        return await verifyAuth(env.ETHERCALC_KEY, messageRoom, messageAuth);
      },
      allowSubmitForm: async () => {
        const access = await this.#effectiveAccess(attachment.uid ?? null);
        return !access.isPrivate && access.canWrite;
      },
      siblingDo: (room: string): WsSiblingDO => {
        const id = env.ROOM.idFromName(room);
        const stub = env.ROOM.get(id);
        const uid = attachment.uid;
        return {
          async fetch(path, init) {
            // Thread the sender's verified identity so a private
            // formdata sibling accepts the write.
            const headers = new Headers(init?.headers);
            headers.delete("X-EC-Uid");
            if (uid !== undefined) headers.set("X-EC-Uid", uid);
            return await stub.fetch(path, { ...init, headers });
          },
        };
      },
    };
  }

  // ─── WS broadcast primitives ───────────────────────────────────────────

  #closeExpiredSessionSocket(ws: WebSocket, attachment?: WsAttachment | null): boolean {
    let stored = attachment;
    if (stored === undefined) {
      try {
        stored = ws.deserializeAttachment() as WsAttachment | null;
      } catch {
        try {
          ws.close(1008, "Session expired");
        } catch {
          // The socket may already be closed; it must still receive no data.
        }
        return true;
      }
    }
    if (!stored || stored.uid === undefined) return false;
    const { sessionExp } = stored;
    if (typeof sessionExp === "number" && Number.isFinite(sessionExp) && Date.now() < sessionExp) {
      return false;
    }
    try {
      ws.close(1008, "Session expired");
    } catch {
      // The socket may already be closed; it must still receive no data.
    }
    return true;
  }

  /**
   * Verify an opaque session token against the singleton AuthDO,
   * reusing a recent in-isolate result when one is still fresh.
   */
  #verifySession(session: string): Promise<SessionPrincipal | null> {
    const now = Date.now();
    const cached = this.#sessionVerifications.get(session);
    if (cached !== undefined && now - cached.at < SESSION_VERIFY_TTL_MS) {
      return cached.principal;
    }
    // `verifyAuthSession` folds every failure into `null`, so a cached
    // promise can never be a sticky rejection.
    const principal = verifyAuthSession(this.#env, session);
    // Re-insert so the refreshed token becomes the newest entry, then
    // drop from the oldest end until the retention cap holds.
    this.#sessionVerifications.delete(session);
    this.#sessionVerifications.set(session, { at: now, principal });
    for (const token of this.#sessionVerifications.keys()) {
      if (this.#sessionVerifications.size <= SESSION_VERIFY_CAP) break;
      this.#sessionVerifications.delete(token);
    }
    return principal;
  }

  async #closeUnauthorizedPrivateSocket(
    ws: WebSocket,
    attachment?: WsAttachment | null,
    purpose: "read" | "read-or-write" = "read",
  ): Promise<boolean> {
    let stored = attachment;
    if (stored === undefined) {
      try {
        stored = ws.deserializeAttachment() as WsAttachment | null;
      } catch {
        try {
          ws.close(1008, "Session invalid");
        } catch {
          // The socket may already be closed.
        }
        return true;
      }
    }
    if (this.#closeExpiredSessionSocket(ws, stored)) return true;
    const access = await this.#effectiveAccess(stored?.uid ?? null);
    if (!access.isPrivate && access.canRead) return false;
    if (!stored || stored.uid === undefined || stored.session === undefined) {
      try {
        ws.close(1008, "Session invalid");
      } catch {
        // The socket may already be closed.
      }
      return true;
    }
    const principal = await this.#verifySession(stored.session);
    const authorized =
      principal !== null &&
      principal.uid === stored.uid &&
      principal.exp === stored.sessionExp &&
      (access.canRead || (purpose === "read-or-write" && access.canWrite));
    if (authorized) return false;
    try {
      ws.close(1008, "Session invalid");
    } catch {
      // The socket may already be closed.
    }
    return true;
  }

  async #sendTo(ws: WebSocket, msg: ServerMessage): Promise<void> {
    // The inbound frame already passed the private-session check. Re-check
    // expiry synchronously before replying without a duplicate AuthDO call.
    if (this.#closeExpiredSessionSocket(ws)) return;
    try {
      ws.send(this.#frameFor(ws, msg));
    } catch {
      // Socket closed underfoot; hibernation will fire webSocketClose.
    }
  }

  /** Encode a ServerMessage for `ws` — socket.io event frame when legacy. */
  #frameFor(ws: WebSocket, msg: ServerMessage): string {
    const att = ws.deserializeAttachment() as WsAttachment | null;
    return att?.legacy ? nativeToSocketIoEvent(msg) : encodeMessage(msg);
  }

  /** Send to every authorized peer except `skip`. */
  async #broadcast(skip: WebSocket, msg: ServerMessage): Promise<void> {
    const peers = this.#state.getWebSockets().filter((peer) => peer !== skip);
    const denied = await Promise.all(
      peers.map((peer) => this.#closeUnauthorizedPrivateSocket(peer)),
    );
    for (let i = 0; i < peers.length; i += 1) {
      if (denied[i]) continue;
      try {
        peers[i]!.send(this.#frameFor(peers[i]!, msg));
      } catch {
        // Skip dead peer; the rest still receive.
      }
    }
  }

  /** Send to every authorized peer (including the sender). */
  async #broadcastAll(msg: ServerMessage): Promise<void> {
    const peers = this.#state.getWebSockets();
    const denied = await Promise.all(
      peers.map((peer) => this.#closeUnauthorizedPrivateSocket(peer)),
    );
    for (let i = 0; i < peers.length; i += 1) {
      if (denied[i]) continue;
      try {
        peers[i]!.send(this.#frameFor(peers[i]!, msg));
      } catch {
        // Best-effort; individual peer errors are expected.
      }
    }
  }

  /**
   * Append one entry under `prefix`, the transport-agnostic fan-out for
   * `ctx.storage.appendLog`. Today only `chat:` is reachable from the
   * handler layer — `log:`/`audit:` writes go through `applyCommand` →
   * `#appendCommand` so they stay serialized alongside the SocialCalc
   * state-save. Any future handler that needs a non-chat prefix should
   * route through this method so the concurrency guard stays in one
   * place.
   */
  async #appendLogEntry(prefix: string, value: string, roomName: string | null): Promise<void> {
    // Today only `chat:` is reachable from the handler layer —
    // `log:`/`audit:` writes go through `applyCommand` → `#appendCommand`.
    // The non-chat fallthrough is a silent no-op rather than a throw to
    // keep `WsStorage.appendLog` honest about its prefix arg.
    /* istanbul ignore else -- @preserve */
    if (prefix === STORAGE_KEYS.chatPrefix) {
      await this.appendChat(value, roomName);
    }
  }

  /** List entries under `prefix` as a `{key-without-prefix: value}` map. */
  async #listHash(prefix: string): Promise<Record<string, string>> {
    const map = await this.#state.storage.list<string>({ prefix });
    const out = Object.create(null) as Record<string, string>;
    for (const [k, v] of map) out[k.slice(prefix.length)] = v;
    return out;
  }

  /**
   * Append a batch of commands to log+audit, run it through SocialCalc, and
   * write the resulting snapshot + meta timestamp. `atomicEntries` lets a
   * caller commit protocol state (currently the append reservation) in that
   * same storage transaction, eliminating a crash window between the sheet
   * mutation and its idempotency marker. Caller is responsible for wrapping
   * this in `blockConcurrencyWhile` when it needs serialization.
   */
  async #appendCommand(
    body: string,
    uid: string | null,
    atomicEntries?: Readonly<Record<string, unknown>>,
  ): Promise<{
    auditSeq: number;
    ts: number;
    auditBody: string;
  } | null> {
    const ss = await this.#getSpreadsheet(uid);
    const attribs = ss.exportSheetData().attribs;
    if (
      !isCommandBatchWithinLimits(body, {
        rows: Number(attribs["lastrow"] ?? 1),
        columns: Number(attribs["lastcol"] ?? 1),
      })
    ) {
      return null;
    }
    await this.#ensureSeqs();
    const logSeq = this.#nextLogSeq!;
    const auditSeq = this.#nextAuditSeq!;
    const ts = Date.now();
    const auditBody = isStorageSafeCommand(body)
      ? body
      : `[oversized command omitted: ${body.length} UTF-16 code units]`;
    ss.executeCommand(body);
    // Cross-sheet formulas added in this command may have resolved to
    // `#NAME?` because the referenced siblings weren't in the formula
    // cache during the initial recalc. Fetch them now and recalc so the
    // stored snapshot carries the live value.
    await this.#hydrateCrossSheetRefs(ss, this.#ownName, uid);
    const newSnapshot = ss.createSpreadsheetSave();

    // Build one atomic snapshot/log/audit replacement. Both transitions need
    // cleanup: chunked → single removes meta/chunks, while single → chunked
    // removes the old fast-path key or readers would keep returning stale data.
    const priorMeta = await readSnapshotMeta(this.#state.storage);
    const newEntries = snapshotEntries(newSnapshot);
    const newMeta = newEntries[STORAGE_KEYS.snapshotMeta] as { chunks: number } | undefined;
    const newChunkCount = newMeta?.chunks ?? 0;
    const stale: string[] = [];
    if (newMeta) {
      stale.push(STORAGE_KEYS.snapshot);
    } else if (priorMeta !== null) {
      stale.push(STORAGE_KEYS.snapshotMeta);
    }
    if (priorMeta !== null) {
      for (let i = newChunkCount; i < priorMeta.chunks; i++) {
        stale.push(snapshotChunkKey(i));
      }
    }
    if (logSeq >= LOG_RING) {
      stale.push(logKey(logSeq - LOG_RING));
    }
    try {
      await this.#state.storage.transaction(async (txn) => {
        await putStorageEntriesBatched(txn, {
          ...newEntries,
          [STORAGE_KEYS.metaUpdatedAt]: ts,
          [logKey(logSeq)]: auditBody,
          [auditKey(auditSeq)]: auditBody,
          ...atomicEntries,
        });
        await deleteStorageKeysBatched(txn, stale);
      });
    } catch (error) {
      // `ss` already executed the command. Drop the mutated cache so a
      // rolled-back write cannot leak into later requests in this isolate.
      this.#ss = null;
      throw error;
    }

    this.#nextLogSeq = logSeq + 1;
    this.#nextAuditSeq = auditSeq + 1;
    await this.#armAlarm();
    return { auditSeq, ts, auditBody };
  }

  // ─── Internals ─────────────────────────────────────────────────────────

  /**
   * Return the cached HeadlessSpreadsheet, hydrating from the stored
   * snapshot if necessary. Idempotent + safe under concurrent calls because
   * the DO only services one request at a time.
   *
   * Phase 8.2 — hydrates sibling sheets referenced by any cross-sheet
   * formula (`'other'!A1`) into SocialCalc's formula-evaluator cache
   * BEFORE the final recalc, so cross-sheet refs compute to real values
   * instead of `#NAME?`. Fetches each sibling's save via the standard
   * `/_do/snapshot` DO-to-DO route; unresolved siblings just stay absent
   * and the formulas return `#NAME?` as graceful degrade.
   */
  async #getSpreadsheet(uid: string | null = null): Promise<HeadlessSpreadsheet> {
    if (this.#ss) return this.#ss;
    const snapshot = await readSnapshot(this.#state.storage);
    // The stored snapshot is AUTHORITATIVE on hydrate. `#appendCommand`
    // writes a freshly-serialized POST-command snapshot every time it
    // also appends to `log:`, so the snapshot already includes every
    // command the log holds. Replaying the log on top of it (the old
    // behaviour) double-applied every command — idempotent for plain
    // `set A1 …` but corrupting for `insertrow`/`paste`/`sort`/`move` on
    // a cold rehydrate (new isolate, or a seeded room). Build from the
    // snapshot ALONE; only replay the log in the no-snapshot (log-only)
    // case, where there is no folded state to replay onto.
    const ss = snapshot
      ? createSpreadsheet({ snapshot })
      : createSpreadsheet({ log: await this.#listPrefix(STORAGE_KEYS.logPrefix) });
    // #ownName is set by the fetch handler on every request; when it's
    // unset (e.g. direct construction in unit tests) we skip cross-sheet
    // resolution so tests don't require a full env.ROOM stub.
    await this.#hydrateCrossSheetRefs(ss, this.#ownName, uid);
    this.#ss = ss;
    return this.#ss;
  }

  /**
   * Fetch a sibling snapshot after confirming this room can read. When this
   * room is the sibling's declared parent, an internal assertion avoids a
   * child→parent authorization callback that would deadlock while the parent
   * is inside `blockConcurrencyWhile`. Worker dispatch strips this header
   * from ingress, and the child accepts it only for marker-matched GETs.
   */
  async #fetchSibling(name: string, uid: string | null): Promise<string | null> {
    const ownAccess = await this.#effectiveAccess(uid);
    if (!ownAccess.canRead) return null;
    const stub = this.#env.ROOM.get(this.#env.ROOM.idFromName(encodeRoom(name)));
    const headers = new Headers();
    if (uid !== null) headers.set("X-EC-Uid", uid);
    if (this.#ownName !== undefined) {
      headers.set(PARENT_READ_HEADER, this.#ownName);
    }
    const res = await stub.fetch(`https://do.local/_do/snapshot?name=${encodeURIComponent(name)}`, {
      method: "GET",
      headers,
    });
    if (res.status !== 200) return null;
    const text = await readBoundedResponseText(res);
    return text || null;
  }

  async #hydrateCrossSheetRefs(
    ss: HeadlessSpreadsheet,
    ownName: string | undefined,
    uid: string | null,
  ): Promise<void> {
    await hydrateCrossSheetRefs(ss, (name) => this.#fetchSibling(name, uid), ownName);
  }

  /** Ordered list of values stored under `prefix`, sorted by key. */
  async #listPrefix(prefix: string): Promise<string[]> {
    const map = await this.#state.storage.list<string>({ prefix });
    // DO storage guarantees lexicographic key order from list().
    return Array.from(map.values());
  }

  /**
   * Lazily populate the in-memory sequence counters by scanning storage on
   * first write. Each counter is derived from the HIGHEST existing key
   * index plus one — NOT the key count. The log ring-trim (`#appendCommand`)
   * and the chat trim (`#trimChat`) delete entries from the FRONT, so after
   * a trim the keys are non-contiguous (e.g. `log:0…0976 … log:0…1999`); a
   * count-based next-seq would then collide with a live key on the first
   * write after an isolate restart — silently overwriting recent data and
   * defeating the ring bound. Keys are fixed-width 16-digit zero-padded
   * (`@ethercalc/shared` padSeq), so the numeric suffix recovers the index
   * exactly. `audit:` is never trimmed (stays contiguous) but uses the same
   * derivation for uniformity.
   */
  async #ensureSeqs(): Promise<void> {
    if (this.#nextLogSeq === null) {
      this.#nextLogSeq = await this.#nextSeq(STORAGE_KEYS.logPrefix);
    }
    if (this.#nextAuditSeq === null) {
      this.#nextAuditSeq = await this.#nextSeq(STORAGE_KEYS.auditPrefix);
    }
    if (this.#nextChatSeq === null) {
      this.#nextChatSeq = await this.#nextSeq(STORAGE_KEYS.chatPrefix);
    }
  }

  /**
   * The next append sequence for `prefix`: one past the highest existing
   * key index, or 0 when the prefix is empty. Robust to the non-contiguous
   * key windows the ring/chat trims leave behind (see `#ensureSeqs`). Keys
   * are fixed-width 16-digit zero-padded, so `list()`'s lexicographic order
   * IS numeric order — the LAST key carries the highest index (the same
   * ordering guarantee `#listPrefix` relies on).
   */
  async #nextSeq(prefix: string): Promise<number> {
    const keys = Array.from((await this.#state.storage.list<string>({ prefix })).keys());
    const last = keys[keys.length - 1];
    return last === undefined ? 0 : Number(last.slice(prefix.length)) + 1;
  }

  /**
   * Mirror this room's latest `updatedAt` to the D1 `rooms` table
   * (Phase 5.1). No-ops when `env.DB` is unbound (Node unit tests
   * construct the DO without Miniflare) or when the DO wasn't told
   * its own room name (legacy `/_do/*` callers that pre-date the
   * `?name=` convention — new code always threads it).
   */
  async #mirrorIndex(roomName: string | null, updatedAt: number): Promise<void> {
    if (!this.#env.DB || !roomName) return;
    // Parented rooms are workbook-internal and never belong in the flat public
    // index, even when their parent is public. This avoids a parent DO hop on
    // every write and fails safe if the parent is temporarily unavailable.
    const meta = await this.#getAccessMeta();
    if (meta.hasParent) return;
    const access = await this.#effectiveAccess(null);
    // Stryker disable next-line LogicalOperator : effectiveAccess can produce
    // only public {false,true,true}, private {true,false,false}, or malformed
    // {false,false,false} here. Regrouping these three terms has the same
    // return result for every reachable triple.
    if (access.isPrivate || !access.canRead || !access.canWrite) return;
    await mirrorRoomToD1(this.#env.DB, roomName, updatedAt);
  }

  /** Symmetric delete for `#mirrorIndex`, used on `DELETE /_do/all`. */
  async #deleteIndex(roomName: string | null): Promise<void> {
    if (!this.#env.DB || !roomName) return;
    await deleteRoomFromD1(this.#env.DB, roomName);
  }

  /**
   * Single choke-point for best-effort D1 work scoped to this room. No-ops
   * without a DB binding or a room name; swallows transient errors. The DO
   * copy is authoritative for live state and the alarm only trims entries
   * already mirrored, so a dropped D1 write is non-fatal (matches the
   * rooms-index mirror's reliability model). Centralizing the guard + catch
   * keeps the audit/chat mirror + delete helpers one line each.
   */
  async #d1(
    roomName: string | null,
    op: (db: D1Database, room: string) => Promise<unknown>,
  ): Promise<void> {
    const db = this.#env.DB;
    if (!db || !roomName) return;
    try {
      await op(db, roomName);
    } catch {
      /* best-effort; DO copy persists and the alarm reconciles */
    }
  }

  /** Best-effort mirror of audit rows to the durable D1 `audit_log`. */
  #mirrorAudit(roomName: string | null, rows: readonly SeqRow[]): Promise<void> {
    return this.#d1(roomName, (db, room) => appendAuditRows(db, room, rows));
  }

  /** Best-effort mirror of chat rows to the durable D1 `chat_log`. */
  #mirrorChat(roomName: string | null, rows: readonly SeqRow[]): Promise<void> {
    return this.#d1(roomName, (db, room) => appendChatRows(db, room, rows));
  }

  /** Drop a room's durable `audit_log` + `chat_log` rows on deletion. */
  #deleteAuditChatFromD1(roomName: string | null): Promise<void> {
    return this.#d1(roomName, async (db, room) => {
      await deleteAuditRows(db, room);
      await deleteChatRows(db, room);
    });
  }

  // ─── Hooks used by future phases (chat/ecell) ──────────────────────────
  // Kept here because WS handlers (Phase 7) will poke these directly; the
  // method shapes are stable now and test coverage locks them in.

  /**
   * Append a chat message. Returns the stored sequence number. `roomName`
   * (the DO's own room) is threaded through so the message can be mirrored
   * to the durable D1 `chat_log`; when omitted (direct unit-test calls) the
   * D1 mirror is skipped.
   */
  async appendChat(message: string, roomName: string | null = null): Promise<number> {
    const seq = await this.#state.blockConcurrencyWhile(async () => {
      await this.#ensureSeqs();
      const s = this.#nextChatSeq!;
      await this.#state.storage.put(chatKey(s), message);
      this.#nextChatSeq = s + 1;
      // Arm the alarm so a chat-only room (no edits) still gets housekeeping
      // (chat trim / TTL); otherwise `chat:` would never be bounded.
      await this.#armAlarm();
      return s;
    });
    // Mirror to the durable D1 `chat_log` so the alarm's chat-trim drops only
    // entries that are already durable there. Best-effort, outside the lock.
    await this.#mirrorChat(roomName, [{ seq, ts: Date.now(), body: message }]);
    if (seq >= CHAT_HISTORY_KEEP) {
      await this.#state.storage.delete(chatKey(seq - CHAT_HISTORY_KEEP));
    }
    return seq;
  }

  /** Upsert an ecell value for a user (LRU-capped). */
  async putEcell(user: string, cell: string): Promise<void> {
    await this.#putEcellCapped(user, cell);
  }

  /**
   * Upsert one `ecell:<user>` entry with least-recently-written eviction
   * once the room exceeds `ECELL_CAP` distinct users. The LRU order is
   * lazily seeded from storage so the cap survives an isolate restart.
   *
   * Re-writing an existing user moves it to the most-recent slot without
   * growing the set. A brand-new user that overflows the cap evicts the
   * oldest entry from both the in-memory order and storage. This stops an
   * attacker cycling arbitrary usernames from blowing per-room storage.
   */
  async #putEcellCapped(user: string, cell: string): Promise<void> {
    if (this.#ecellOrder === null) {
      const map = await this.#state.storage.list<string>({
        prefix: STORAGE_KEYS.ecellPrefix,
      });
      // list() returns lexicographic key order — not write order — but
      // that's a stable, deterministic seed; subsequent writes refine it
      // toward true recency.
      this.#ecellOrder = Array.from(map.keys()).map((k) =>
        k.slice(STORAGE_KEYS.ecellPrefix.length),
      );
    }
    const order = this.#ecellOrder;
    const existing = order.indexOf(user);
    if (existing !== -1) order.splice(existing, 1);
    order.push(user);
    const evicted: string[] = [];
    while (order.length > ECELL_CAP) {
      // shift() over a non-empty array (length > cap ≥ 1) never returns
      // undefined; the `?? ''`-free assertion keeps the type honest.
      const victim = order.shift() as string;
      evicted.push(ecellKey(victim));
    }
    await this.#state.storage.put(ecellKey(user), cell);
    if (evicted.length > 0) await this.#state.storage.delete(evicted);
    await this.#armAlarm();
  }

  /**
   * Reset in-memory state that a full storage wipe (`deleteAll`) or
   * re-seed invalidates: the LRU ecell order (now empty / re-seeded) and
   * the cached alarm-armed flag (the writer re-arms afterwards). Kept
   * separate from the seq-counter resets above so each wipe site reads
   * clearly.
   */
  #resetVolatile(): void {
    this.#ecellOrder = null;
    this.#alarmArmed = false;
    this.#accessMeta = null;
  }

  /**
   * Arm the housekeeping alarm if it isn't already pending. Called from
   * every write path. We avoid a `getAlarm()` round-trip on the hot path
   * by caching the armed state in `#alarmArmed`; the alarm handler clears
   * it before re-arming, and the very first arm reads `getAlarm()` once to
   * recover state across isolate restarts.
   */
  async #armAlarm(): Promise<void> {
    if (this.#alarmArmed) return;
    const current = await this.#state.storage.getAlarm();
    if (current !== null) {
      this.#alarmArmed = true;
      return;
    }
    await this.#state.storage.setAlarm(Date.now() + ALARM_INTERVAL_MS);
    this.#alarmArmed = true;
  }

  /**
   * Housekeeping alarm (AGENTS.md §13 Q10). Fires roughly hourly while a
   * room stays active. Two jobs:
   *   (a) Trim `chat:*` / `audit:*` down to recent tails — both are
   *       mirrored to D1 beyond the DO lifetime, so the DO copy only
   *       needs to cover live catch-up.
   *   (b) TTL / `--expire`: when `ETHERCALC_EXPIRE` is set and the room
   *       hasn't been touched within the TTL, wipe it (same hammer as
   *       `DELETE /_do/all`).
   *
   * Re-arm gate (idle rooms must go fully dormant — no hourly wake):
   *   re-arm ONLY when at least one of:
   *     (1) any live WebSocket (`getWebSockets().length > 0`);
   *     (2) a TTL is configured (`parseExpireMs` non-null) so expiry can
   *         still fire;
   *     (3) this run actually trimmed rows (tails may still exceed caps
   *         next hour).
   *   Otherwise do NOT re-arm. Write paths re-arm via `#armAlarm()` so the
   *   next mutation resumes the cadence (invariant: every storage write
   *   that should keep the room "active" calls `#armAlarm` — see
   *   `#appendCommand`, `appendChat`, `#putEcellCapped`, `#putSnapshot`,
   *   `#postSeed`, `#postInstall`, `#postPitrTouch`).
   */
  async alarm(): Promise<void> {
    this.#alarmArmed = false;
    // (b) TTL expiry — if configured and the room is stale, drop it and
    // do NOT re-arm (the room is gone).
    const ttlMs = parseExpireMs(this.#env.ETHERCALC_EXPIRE);
    if (ttlMs !== null) {
      const updatedAt = await this.#state.storage.get<number>(STORAGE_KEYS.metaUpdatedAt);
      if (typeof updatedAt === "number" && Date.now() - updatedAt >= ttlMs) {
        await this.#deleteAllAndUnindex(this.#ownName ?? null, null);
        return;
      }
    }
    // (a) Catch up tails from migrated or pre-cap state. Normal append paths
    // evict one oldest row immediately; the alarm handles existing overflow.
    const chatTrimmed = await this.#trimTail(STORAGE_KEYS.chatPrefix, CHAT_HISTORY_KEEP);
    const auditTrimmed = await this.#trimTail(STORAGE_KEYS.auditPrefix, AUDIT_HISTORY_KEEP);
    // Invariant: write paths call `#armAlarm()`; this gate only decides
    // whether the idle cadence continues. See method doc above for the
    // three re-arm conditions.
    const hasSockets = this.#state.getWebSockets().length > 0;
    const ttlConfigured = ttlMs !== null;
    const trimmed = chatTrimmed || auditTrimmed;
    if (hasSockets || ttlConfigured || trimmed) {
      await this.#armAlarm();
    }
  }

  /**
   * Delete all but the most recent `keep` entries under `prefix`. Returns
   * true when at least one key was removed (so the alarm re-arm gate can
   * keep the cadence while a tail still needs draining). The full record
   * is mirrored to D1 at append time, so the dropped (oldest) entries
   * remain durable there — the DO copy is only a live-catch-up tail.
   */
  async #trimTail(prefix: string, keep: number): Promise<boolean> {
    const map = await this.#state.storage.list<string>({ prefix });
    const keys = Array.from(map.keys());
    if (keys.length <= keep) return false;
    // list() returns ascending key order; the oldest are at the front.
    await deleteStorageKeysBatched(this.#state.storage, keys.slice(0, keys.length - keep));
    return true;
  }

  /** Snapshot of all ecells as `{user → cell}`. */
  async listEcells(): Promise<Record<string, string>> {
    const map = await this.#state.storage.list<string>({
      prefix: STORAGE_KEYS.ecellPrefix,
    });
    const out = Object.create(null) as Record<string, string>;
    for (const [k, v] of map) {
      out[k.slice(STORAGE_KEYS.ecellPrefix.length)] = v;
    }
    return out;
  }
}