feat(ws): persist WebSocket connections in SQLite with recovery/rollback

Add a `WsConnection` module that durably persists WebSocket connection
state, outbound message queues, and fragmented inbound streams to the
existing opencode SQLite database.

Key features:

- **Connection lifecycle** – `open()` / `close()` / `get()` / `setState()`
  track each connection with a cryptographically-random recovery token
  and a configurable recovery window (default 1 h).

- **Durable message queue** – `enqueue()` / `markSent()` / `ack()`
  implement a pending → sent → acked state machine.  Backpressure is
  handled by keeping "sent" messages in the queue until the client
  explicitly acks them, so nothing is lost if the network drops.

- **Connection recovery** – `recover()` looks up a connection by token
  and replays all non-acked messages since `last_acked_seq`, enabling
  clients that reconnect mid-stream to continue without data loss.

- **Stream / fragment support** – `beginStream()` / `appendFragment()` /
  `commitStream()` accumulate out-of-order fragments atomically and
  enqueue the fully-assembled message only when all fragments arrive.

- **Stale-stream rollback** – `rollbackStaleStreams()` deletes uncommitted
  streams older than a given threshold, so partial transmissions from
  senders that disconnect mid-message do not accumulate indefinitely.

- **Housekeeping** – `cleanup()` removes closed connections (and their
  cascaded messages / streams) outside the recovery window.

Schema (migration 20260225000000):
  ws_connection  – connection state, token, acked cursor
  ws_message     – ordered outbound queue with delivery tracking
  ws_stream      – fragmented inbound stream accumulator

All functions accept an optional `_db` parameter for isolated unit
testing without needing a full Instance context.  54 tests cover the
full API surface including the end-to-end recovery and mid-stream-drop
scenarios.

https://claude.ai/code/session_01Urv7ih36PJX12fvs5DAQZn

Author: claude

packages/opencode/migration/20260225000000_websocket_persistence/migration.sql

@@ -0,0 +1,53 @@
+CREATE TABLE `ws_connection` (
+	`id` text PRIMARY KEY,
+	`user_id` text,
+	`session_id` text,
+	`recovery_token` text NOT NULL UNIQUE,
+	`recovery_expires_at` integer NOT NULL,
+	`state` text,
+	`last_acked_seq` integer NOT NULL DEFAULT 0,
+	`opened_at` integer NOT NULL,
+	`closed_at` integer,
+	`last_activity_at` integer NOT NULL,
+	`time_created` integer NOT NULL,
+	`time_updated` integer NOT NULL
+);
+--> statement-breakpoint
+CREATE TABLE `ws_message` (
+	`seq` integer PRIMARY KEY AUTOINCREMENT,
+	`connection_id` text NOT NULL,
+	`type` text NOT NULL,
+	`payload` text NOT NULL,
+	`status` text NOT NULL DEFAULT 'pending',
+	`delivery_attempts` integer NOT NULL DEFAULT 0,
+	`acked_at` integer,
+	`time_created` integer NOT NULL,
+	`time_updated` integer NOT NULL,
+	CONSTRAINT `ws_message_connection_id_fk` FOREIGN KEY (`connection_id`) REFERENCES `ws_connection`(`id`) ON DELETE CASCADE
+);
+--> statement-breakpoint
+CREATE TABLE `ws_stream` (
+	`id` text PRIMARY KEY,
+	`connection_id` text NOT NULL,
+	`message_type` text NOT NULL,
+	`fragments` text NOT NULL DEFAULT '[]',
+	`total_fragments` integer,
+	`committed` integer NOT NULL DEFAULT 0,
+	`time_created` integer NOT NULL,
+	`time_updated` integer NOT NULL,
+	CONSTRAINT `ws_stream_connection_id_fk` FOREIGN KEY (`connection_id`) REFERENCES `ws_connection`(`id`) ON DELETE CASCADE
+);
+--> statement-breakpoint
+CREATE INDEX `ws_connection_recovery_idx` ON `ws_connection` (`recovery_token`);
+--> statement-breakpoint
+CREATE INDEX `ws_connection_session_idx` ON `ws_connection` (`session_id`);
+--> statement-breakpoint
+CREATE INDEX `ws_connection_closed_idx` ON `ws_connection` (`closed_at`);
+--> statement-breakpoint
+CREATE INDEX `ws_message_connection_status_idx` ON `ws_message` (`connection_id`, `status`);
+--> statement-breakpoint
+CREATE INDEX `ws_message_connection_seq_idx` ON `ws_message` (`connection_id`, `seq`);
+--> statement-breakpoint
+CREATE INDEX `ws_stream_connection_idx` ON `ws_stream` (`connection_id`);
+--> statement-breakpoint
+CREATE INDEX `ws_stream_committed_idx` ON `ws_stream` (`connection_id`, `committed`);

packages/opencode/src/storage/schema.ts

@@ -2,3 +2,4 @@ export { ControlAccountTable } from "../control/control.sql"
 export { SessionTable, MessageTable, PartTable, TodoTable, PermissionTable } from "../session/session.sql"
 export { SessionShareTable } from "../share/share.sql"
 export { ProjectTable } from "../project/project.sql"
+export { WsConnectionTable, WsMessageTable, WsStreamTable } from "../ws/ws.sql"

packages/opencode/src/ws/ws.sql.ts

@@ -0,0 +1,94 @@
+import { sqliteTable, text, integer, index } from "drizzle-orm/sqlite-core"
+import { Timestamps } from "@/storage/schema.sql"
+
+/**
+ * Tracks the lifecycle of every WebSocket connection.
+ * Closed connections are kept for the duration of the recovery window so
+ * clients can reconnect and replay undelivered messages.
+ */
+export const WsConnectionTable = sqliteTable(
+  "ws_connection",
+  {
+    id: text().primaryKey(),
+    user_id: text(),
+    session_id: text(),
+    /** Opaque token the client stores and presents on reconnect. */
+    recovery_token: text().notNull().unique(),
+    /** Epoch-ms after which the recovery token is invalid. */
+    recovery_expires_at: integer().notNull(),
+    /** Arbitrary subscriber/room state, JSON-serialised. */
+    state: text({ mode: "json" }).$type<Record<string, unknown>>(),
+    /** Highest message seq the client has explicitly acknowledged. */
+    last_acked_seq: integer().notNull().$default(() => 0),
+    opened_at: integer().notNull(),
+    /** NULL while the connection is still active. */
+    closed_at: integer(),
+    last_activity_at: integer().notNull(),
+    ...Timestamps,
+  },
+  (t) => [
+    index("ws_connection_recovery_idx").on(t.recovery_token),
+    index("ws_connection_session_idx").on(t.session_id),
+    index("ws_connection_closed_idx").on(t.closed_at),
+  ],
+)
+
+/**
+ * Durable outbound message queue for a connection.
+ * Messages progress through: pending → sent → acked (or failed).
+ * Acked messages are retained briefly then purged by cleanup().
+ */
+export const WsMessageTable = sqliteTable(
+  "ws_message",
+  {
+    /** Monotonically-increasing delivery sequence number. */
+    seq: integer().primaryKey({ autoIncrement: true }),
+    connection_id: text()
+      .notNull()
+      .references(() => WsConnectionTable.id, { onDelete: "cascade" }),
+    type: text().notNull(),
+    payload: text({ mode: "json" }).notNull().$type<unknown>(),
+    status: text()
+      .notNull()
+      .$type<"pending" | "sent" | "acked" | "failed">()
+      .$default(() => "pending"),
+    delivery_attempts: integer().notNull().$default(() => 0),
+    acked_at: integer(),
+    ...Timestamps,
+  },
+  (t) => [
+    index("ws_message_connection_status_idx").on(t.connection_id, t.status),
+    index("ws_message_connection_seq_idx").on(t.connection_id, t.seq),
+  ],
+)
+
+/**
+ * Tracks fragmented/streaming inbound messages.
+ * When a large message arrives in fragments, each piece is appended here.
+ * commitStream() assembles the fragments into a single WsMessage.
+ * Fragments belonging to streams that never complete are rolled back by
+ * rollbackStaleStreams().
+ */
+export const WsStreamTable = sqliteTable(
+  "ws_stream",
+  {
+    id: text().primaryKey(),
+    connection_id: text()
+      .notNull()
+      .references(() => WsConnectionTable.id, { onDelete: "cascade" }),
+    message_type: text().notNull(),
+    /** Ordered list of received fragments, JSON-serialised. */
+    fragments: text({ mode: "json" })
+      .notNull()
+      .$type<{ index: number; data: string }[]>()
+      .$default(() => []),
+    /** Set on commitStream(); null means the total is not yet known. */
+    total_fragments: integer(),
+    committed: integer({ mode: "boolean" }).notNull().$default(() => false),
+    ...Timestamps,
+  },
+  (t) => [
+    index("ws_stream_connection_idx").on(t.connection_id),
+    index("ws_stream_committed_idx").on(t.connection_id, t.committed),
+  ],
+)