feat(cursor): Cursor provider (chat + Tab) with cross-request session reuse

apps/platform-cloudflare/entry.ts

@@ -13,6 +13,7 @@ import {
 // exported name on the Worker module. The wrangler `migrations.new_sqlite_classes`
 // entry must match this export.
 export { BroadcastDO } from './src/broadcast-do.ts';
+export { DurableHttpSessionDO } from './src/durable-http-session-do.ts';
 
 initBackgroundSchedulerResolver(c => promise => c.executionCtx.waitUntil(promise));
 

apps/platform-cloudflare/package.json

@@ -10,6 +10,7 @@
     "@floway-dev/gateway": "workspace:*",
     "@floway-dev/http": "workspace:*",
     "@floway-dev/platform": "workspace:*",
+    "@floway-dev/proxy": "workspace:*",
     "@floway-dev/protocols": "workspace:*",
     "hono": "^4"
   },

apps/platform-cloudflare/src/_held-socket.ts

@@ -0,0 +1,26 @@
+// Internal helper (underscore prefix = not a platform export): adapts a
+// DialedSocket into the DuplexStream shape `@floway-dev/http` consumes, plus
+// an idempotent close.
+//
+// Why this exists as its own ~tiny class: the DurableHttpSessionDO is the
+// first piece of code in the workspace that OWNS a live outbound socket across
+// inbound requests. If a future need arises for a raw persistent TCP duplex
+// (not HTTP), HeldSocket is the seam to promote into a platform `DurableSocketDial`
+// contract + a `DurableSocketDO` — a mechanical refactor rather than a rewrite.
+
+import type { DuplexStream } from '@floway-dev/http';
+import type { DialedSocket } from '@floway-dev/platform';
+
+export class HeldSocket {
+  constructor(private readonly dialed: DialedSocket) {}
+
+  /** The duplex view `fetchOnStream` reads/writes. */
+  asDuplex(): DuplexStream {
+    return { readable: this.dialed.readable, writable: this.dialed.writable };
+  }
+
+  /** Idempotent — safe to call on an already-closed/errored socket. */
+  close(): Promise<void> {
+    return this.dialed.close();
+  }
+}

apps/platform-cloudflare/src/bootstrap.ts

@@ -1,4 +1,5 @@
 import { DurableObjectChannelBroker, type BroadcastNamespace } from './do-channel-broker.ts';
+import { DurableObjectDurableHttpSession, type DurableHttpSessionNamespace } from './do-durable-http-session.ts';
 import { createCloudflareImageProcessor, type ImagesBinding } from './image-processor.ts';
 import { KvImageCache, type KvNamespace } from './kv-image-cache.ts';
 import { R2FileProvider, type R2BucketLike } from './r2-file-provider.ts';
@@ -10,6 +11,7 @@ import type { DumpMetadata } from '@floway-dev/gateway/dump-types';
 import { addTrustedRootCAs } from '@floway-dev/http';
 import {
   IMAGE_CACHE_POLICY,
+  initDurableHttpSession,
   initEnv,
   initFileProvider,
   initImageCacheStore,
@@ -25,15 +27,17 @@ export interface CloudflareEnv {
   IMAGES: ImagesBinding;
   KV: KvNamespace;
   BROADCAST_DO: BroadcastNamespace;
+  DURABLE_HTTP_SESSION: DurableHttpSessionNamespace;
   [key: string]: unknown;
 }
 
 // Every binding declared on `CloudflareEnv` is load-bearing — D1 holds all
 // config and telemetry, R2 holds spilled payloads, Images compresses inline
-// images, KV memoises compressed image results. A missing binding means
-// wrangler.jsonc drifted from the code, so we refuse to initialise rather
-// than 503 on first use of the absent binding.
-const REQUIRED_BINDINGS = ['DB', 'FILES', 'IMAGES', 'KV', 'BROADCAST_DO'] as const;
+// images, KV memoises compressed image results, BROADCAST_DO fans out dump
+// events, DURABLE_HTTP_SESSION holds cross-request upstream response streams
+// (cursor RunSSE). A missing binding means wrangler.jsonc drifted from the
+// code, so we refuse to initialise rather than 503 on first use.
+const REQUIRED_BINDINGS = ['DB', 'FILES', 'IMAGES', 'KV', 'BROADCAST_DO', 'DURABLE_HTTP_SESSION'] as const;
 
 export const bootstrapCloudflarePlatform = (env: CloudflareEnv): { db: SqlDatabase } => {
   const missing = REQUIRED_BINDINGS.filter(name => env[name] === undefined);
@@ -58,5 +62,6 @@ export const bootstrapCloudflarePlatform = (env: CloudflareEnv): { db: SqlDataba
   addTrustedRootCAs(cloudflareRuntimeRootCAs);
   initDumpStore(new FileDumpStore(env.DB, files));
   initDumpBroker(new DurableObjectChannelBroker<DumpMetadata>(env.BROADCAST_DO, dumpCodec));
+  initDurableHttpSession(new DurableObjectDurableHttpSession(env.DURABLE_HTTP_SESSION));
   return { db: env.DB };
 };

apps/platform-cloudflare/src/broadcast-do_test.ts

@@ -52,6 +52,12 @@ class FakeState {
   push(ws: FakeWebSocket): void {
     this.sockets.push(ws);
   }
+  // BroadcastDO never touches storage; present only to satisfy the shared
+  // DurableObjectState type (DurableHttpSessionDO uses setAlarm).
+  storage = {
+    async setAlarm(): Promise<void> {},
+    async deleteAlarm(): Promise<void> {},
+  };
 }
 
 test('BroadcastDO extends DurableObject so the runtime gates RPC dispatch on it', () => {

apps/platform-cloudflare/src/cloudflare-workers.d.ts

@@ -20,9 +20,14 @@ declare module 'cloudflare:workers' {
   }
 }
 
-// The runtime's `DurableObjectState` surface the actor touches — just the
-// WebSocket Hibernation entry points.
+// The runtime's `DurableObjectState` surface the actor touches — the
+// WebSocket Hibernation entry points and the alarm scheduler (used by
+// DurableHttpSessionDO for idle eviction).
 interface DurableObjectState {
   acceptWebSocket(server: WebSocket): void;
   getWebSockets(): WebSocket[];
+  storage: {
+    setAlarm(scheduledTime: number): Promise<void>;
+    deleteAlarm(): Promise<void>;
+  };
 }

apps/platform-cloudflare/src/do-durable-http-session.ts

@@ -0,0 +1,173 @@
+import type {
+  DurableHttpSessionStartInit,
+  DurableHttpSessionStartResult,
+} from './durable-http-session-do.ts';
+import type {
+  DurableHttpSession,
+  DurableHttpSessionAcquireOptions,
+  DurableHttpSessionHandle,
+  DurableHttpSessionInit,
+} from '@floway-dev/platform';
+
+// CF implementation of the DurableHttpSession contract, mirroring the
+// DurableObjectChannelBroker shape: resolve a per-sessionKey DO stub, then
+// drive it over RPC + a WebSocket body channel. The DO holds the live outbound
+// RunSSE response; this broker exposes it as the contract's ReadableStream.
+
+const DEFAULT_IDLE_TIMEOUT_MS = 5 * 60 * 1000;
+
+// One credit = "send me the next chunk". The broker sends exactly one per
+// ReadableStream pull; the DO treats any inbound message as a credit and ignores
+// the content, so a 1-byte sentinel is enough.
+const CREDIT_BYTE = new Uint8Array([1]);
+
+// Minimal namespace/stub surface — declared locally so this file stays off
+// `@cloudflare/workers-types` (same convention as do-channel-broker.ts).
+export interface DurableHttpSessionNamespace {
+  idFromName(name: string): unknown;
+  get(id: unknown): DurableHttpSessionStub;
+}
+
+interface DurableHttpSessionStub {
+  queryOrStart(
+    init: DurableHttpSessionStartInit | null,
+    idleTimeoutMs: number,
+  ): Promise<DurableHttpSessionStartResult | null>;
+  release(): Promise<void>;
+  discard(reason: string): Promise<void>;
+  fetch(request: Request): Promise<Response>;
+}
+
+export class DurableObjectDurableHttpSession implements DurableHttpSession {
+  constructor(private readonly namespace: DurableHttpSessionNamespace) {}
+
+  private stub(sessionKey: string): DurableHttpSessionStub {
+    return this.namespace.get(this.namespace.idFromName(sessionKey));
+  }
+
+  async acquire(
+    sessionKey: string,
+    init: DurableHttpSessionInit | null,
+    opts?: DurableHttpSessionAcquireOptions,
+  ): Promise<DurableHttpSessionHandle | null> {
+    const stub = this.stub(sessionKey);
+    const idleTimeoutMs = opts?.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS;
+
+    const meta = await stub.queryOrStart(init, idleTimeoutMs);
+    if (!meta) return null; // miss + init=null → caller degrades
+
+    const wsResponse = await stub.fetch(new Request('https://durable-http.do/body', {
+      headers: { Upgrade: 'websocket' },
+    }));
+    if (wsResponse.status !== 101) {
+      throw new Error(`DurableHttpSessionDO body upgrade returned HTTP ${wsResponse.status} instead of 101`);
+    }
+    const socket = (wsResponse as Response & { webSocket?: WebSocket }).webSocket;
+    if (!socket) throw new Error('DurableHttpSessionDO returned 101 without a webSocket');
+    socket.accept();
+
+    const body = wsToByteStream(socket, opts?.signal);
+
+    return {
+      status: meta.status,
+      headers: new Headers(meta.headers),
+      body,
+      async release(): Promise<void> {
+        try { socket.close(1000, 'released'); } catch { /* already closed */ }
+        await stub.release();
+      },
+      async discard(reason: string): Promise<void> {
+        try { socket.close(1000, 'discarded'); } catch { /* already closed */ }
+        await stub.discard(reason);
+      },
+    };
+  }
+}
+
+// Wrap the body WebSocket as a ReadableStream<Uint8Array>. Binary frames are
+// the body bytes; a close ends the stream; an error rejects the pending read.
+// Aborting the caller signal closes the socket and ends the stream — it does
+// NOT discard the DO session (sibling acquires keep working), matching the
+// contract's signal semantics.
+const wsToByteStream = (
+  socket: WebSocket,
+  signal: AbortSignal | undefined,
+): ReadableStream<Uint8Array> => {
+  // Head-index cursor over a push-only array so a burst that queues dozens of
+  // chunks before the consumer resumes doesn't pay O(n) per delivered chunk.
+  const queue: Uint8Array[] = [];
+  let queueHead = 0;
+  let pull: ((v: { value?: Uint8Array; done: boolean }) => void) | null = null;
+  let pendingError: unknown = null;
+  let closed = false;
+
+  const deliver = (chunk: Uint8Array | null): void => {
+    if (pull) {
+      const r = pull;
+      pull = null;
+      r(chunk ? { value: chunk, done: false } : { done: true });
+    } else if (chunk) {
+      queue.push(chunk);
+    }
+  };
+
+  const end = (): void => {
+    if (closed) return;
+    closed = true;
+    deliver(null);
+    try { socket.close(1000, 'consumer done'); } catch { /* already closed */ }
+  };
+
+  socket.addEventListener('message', event => {
+    if (closed) return;
+    const raw = (event as MessageEvent).data as string | ArrayBuffer | Uint8Array;
+    let bytes: Uint8Array;
+    if (typeof raw === 'string') bytes = new TextEncoder().encode(raw);
+    else if (raw instanceof Uint8Array) bytes = raw;
+    else bytes = new Uint8Array(raw);
+    deliver(bytes);
+  });
+  socket.addEventListener('close', () => { closed = true; deliver(null); });
+  socket.addEventListener('error', () => {
+    if (pendingError === null) pendingError = new Error('DurableHttpSession body socket error');
+    closed = true;
+    deliver(null);
+  });
+
+  if (signal) {
+    if (signal.aborted) end();
+    else signal.addEventListener('abort', end, { once: true });
+  }
+
+  return new ReadableStream<Uint8Array>({
+    async pull(controller): Promise<void> {
+      if (queueHead < queue.length) {
+        controller.enqueue(queue[queueHead]!);
+        queueHead += 1;
+        if (queueHead >= queue.length) {
+          queue.length = 0;
+          queueHead = 0;
+        }
+        return;
+      }
+      if (closed) {
+        if (pendingError) { controller.error(pendingError); return; }
+        controller.close();
+        return;
+      }
+      // Credit-based backpressure: request exactly one chunk, then wait for it.
+      // With highWaterMark 0 the stream only pulls on an active read, so a
+      // consumer that pauses (transport parked at exec_mcp) sends no credit and
+      // the DO holds subsequent chunks in its buffer for the next acquire.
+      try { socket.send(CREDIT_BYTE); } catch { controller.close(); return; }
+      const result = await new Promise<{ value?: Uint8Array; done: boolean }>(resolve => { pull = resolve; });
+      if (result.done) {
+        if (pendingError) controller.error(pendingError);
+        else controller.close();
+        return;
+      }
+      if (result.value) controller.enqueue(result.value);
+    },
+    cancel(): void { end(); },
+  }, { highWaterMark: 0 });
+};