From 08889bb085b4a4eaef2a5d785d7bcc2d70f17c1b Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 11 Oct 2026 18:28:34 +0000 Subject: [PATCH] test: Durable Object harness for SessionDO (#31) A bare-node harness that drives a real SessionDO inside npm test: ctx.storage.sql over in-memory node:sqlite (multi-statement exec and ArrayBuffer BLOBs reproduced), explicit alarm firing, restart over retained storage, and AI via fakeAiRunner behind selectBudgetedAiRunner with call counting and interception. Four tests: hydrate after restart, the expanded:false round trip through real SQL, failed pole embed kicking the pump, and axisCreationsInFlight blocking a second concurrent create at the cap. Each fails when the code it guards is removed. fakeAiRunner takes an optional latencyMs (default unchanged at 250). Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011jTqcyCetskFRRENsms9wZ --- CLAUDE.md | 15 ++ src/dev-fake-ai.ts | 7 +- test/helpers/cloudflare-workers-stub.ts | 12 ++ test/helpers/do-harness.ts | 257 ++++++++++++++++++++++++ test/session-do.test.ts | 133 ++++++++++++ 5 files changed, 422 insertions(+), 2 deletions(-) create mode 100644 test/helpers/cloudflare-workers-stub.ts create mode 100644 test/helpers/do-harness.ts create mode 100644 test/session-do.test.ts diff --git a/CLAUDE.md b/CLAUDE.md index fcfe08f..3b168d7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -32,6 +32,21 @@ is pure and heavily tested, [src/session-do.ts](src/session-do.ts) is the thin stateful shell around it. New pool/scoring/projection logic belongs in the former. +For what genuinely lives in the shell (SQL round trips, alarm kicks, guards +that span awaits), [test/helpers/do-harness.ts](test/helpers/do-harness.ts) +drives a real `SessionDO` inside the same bare-node `npm test`. Storage is a +real in-memory SQLite (`node:sqlite`, Node 22), so the DO's own SQL runs as +written; alarms only fire when a test calls `fireAlarm()`/`drainAlarms()`; +`restart()` re-constructs over the same storage to simulate eviction; AI is +`fakeAiRunner` behind the production `selectBudgetedAiRunner` seam, with +`h.ai.calls` to count calls and `h.ai.intercept` to fail or hold them. Each +test file stubs the runtime module itself (vi.mock is per file): +`vi.mock("cloudflare:workers", () => import("./helpers/cloudflare-workers-stub"))`. +See [test/session-do.test.ts](test/session-do.test.ts). Stored embeddings come +back at float32 precision; compare against `Float32Array`-rounded values. +`@cloudflare/vitest-pool-workers` was passed over: a second pool inside +workerd, and the `remote: true` AI binding drags in the Access-gated proxy. + ## Reaching Workers AI locally Two independent traps, and they stack. Probe before debugging app code — diff --git a/src/dev-fake-ai.ts b/src/dev-fake-ai.ts index 455f4bd..1e38576 100644 --- a/src/dev-fake-ai.ts +++ b/src/dev-fake-ai.ts @@ -132,11 +132,14 @@ function fakeRewrite(fragment: string, n: number): string { return `${stem}${marker}`; } -export function fakeAiRunner(): AiRunner { +/** `latencyMs` is pretend latency per call. The dev server keeps the default so + * the field drips at a believable pace; the Durable Object harness + * (test/helpers/do-harness.ts) passes 0 so tests do not sleep. */ +export function fakeAiRunner({ latencyMs = 250 }: { latencyMs?: number } = {}): AiRunner { const counters = new Map(); return { async run(_model, inputs) { - await new Promise((r) => setTimeout(r, 250)); // a little pretend latency + if (latencyMs > 0) await new Promise((r) => setTimeout(r, latencyMs)); if (Array.isArray(inputs.text)) { return { data: (inputs.text as string[]).map(pseudoEmbedding) }; } diff --git a/test/helpers/cloudflare-workers-stub.ts b/test/helpers/cloudflare-workers-stub.ts new file mode 100644 index 0000000..3f537c9 --- /dev/null +++ b/test/helpers/cloudflare-workers-stub.ts @@ -0,0 +1,12 @@ +// Stand-in for the `cloudflare:workers` module, which does not resolve under +// plain vitest. Unlike the bare `class {}` other test files mock it with, this +// base class stores `ctx` and `env` the way the runtime's does, so a DO can be +// constructed normally and its constructor (blockConcurrencyWhile, migrate, +// hydrate) actually runs. Used with test/helpers/do-harness.ts. + +export class DurableObject { + constructor( + protected readonly ctx: unknown, + protected readonly env: Env, + ) {} +} diff --git a/test/helpers/do-harness.ts b/test/helpers/do-harness.ts new file mode 100644 index 0000000..f68b769 --- /dev/null +++ b/test/helpers/do-harness.ts @@ -0,0 +1,257 @@ +// A Durable Object test harness that runs inside the ordinary `npm test`: bare +// node, no Workers runtime, no wrangler, no network, no credentials (issue #31). +// +// Why not @cloudflare/vitest-pool-workers: it runs tests inside workerd, which +// means a second vitest pool next to the bare-node one every other test uses, +// and this project's `ai` binding is `remote: true`, which the pool resolves +// through the same remote-binding proxy that dies without Cloudflare Access +// credentials (CLAUDE.md). The parts of the runtime SessionDO actually leans on +// are small, so this fakes exactly those: +// +// - `ctx.storage.sql` — backed by a REAL SQLite (node:sqlite, in memory), so +// the DO's own CREATE/INSERT/SELECT statements run as written. Two runtime +// behaviours are reproduced on purpose: multi-statement `exec` (migrate() +// relies on it) and BLOB columns reading back as ArrayBuffer, not node's +// Uint8Array — `new Float32Array(uint8)` would silently convert element by +// element and corrupt every embedding. +// - `ctx.storage.getAlarm/setAlarm/deleteAlarm` — a single recorded time. Tests +// fire it explicitly with `fireAlarm()`; nothing runs on a timer. +// - `ctx.blockConcurrencyWhile` — runs the callback and keeps its promise, so +// `ready()` can await it. +// - Restart — `restart()` builds a fresh instance over the same database, which +// is what isolate eviction looks like to a DO: memory gone, storage kept. +// +// AI goes through the production seam (`selectBudgetedAiRunner`) with the +// Workers AI branch pointed at `fakeAiRunner` from src/dev-fake-ai.ts, wrapped +// so a test can count calls and make specific ones fail or wait. +// +// Test files must stub the runtime module themselves, because vi.mock is +// hoisted per file: +// +// vi.mock("cloudflare:workers", () => import("./helpers/cloudflare-workers-stub")); + +import { DatabaseSync, type SQLInputValue } from "node:sqlite"; +import { fakeAiRunner } from "../../src/dev-fake-ai"; + +type Row = Record; + +/** The subset of the runtime's SqlStorageCursor that SessionDO uses. */ +class FakeCursor { + constructor(private readonly rows: T[]) {} + toArray(): T[] { + return this.rows; + } + one(): T { + if (this.rows.length !== 1) throw new Error(`Expected exactly one result from SQL query, got ${this.rows.length}`); + return this.rows[0]!; + } + [Symbol.iterator](): Iterator { + return this.rows[Symbol.iterator](); + } +} + +function toBinding(value: unknown): SQLInputValue { + if (value instanceof ArrayBuffer) return new Uint8Array(value); + return value as SQLInputValue; +} + +/** The runtime hands BLOBs back as ArrayBuffer; node:sqlite hands back + * Uint8Array. Normalise so the code under test sees what production sees. */ +function fromColumn(value: unknown): unknown { + if (value instanceof Uint8Array) return value.buffer.slice(value.byteOffset, value.byteOffset + value.byteLength); + return value; +} + +export class FakeSqlStorage { + constructor(readonly db: DatabaseSync) {} + + exec(query: string, ...bindings: unknown[]): FakeCursor { + // The runtime accepts several statements in one exec when nothing is + // bound; prepare() would compile only the first and silently drop the rest. + if (bindings.length === 0 && /;\s*\S/.test(query.trim().replace(/;\s*$/, ""))) { + this.db.exec(query); + return new FakeCursor([]); + } + const stmt = this.db.prepare(query); + const params = bindings.map(toBinding); + if (!/^\s*(SELECT|WITH|PRAGMA)\b/i.test(query) && !/\bRETURNING\b/i.test(query)) { + stmt.run(...params); + return new FakeCursor([]); + } + const rows = stmt.all(...params).map((r) => { + const out: Row = {}; + for (const [k, v] of Object.entries(r)) out[k] = fromColumn(v); + return out as T; + }); + return new FakeCursor(rows); + } +} + +export class FakeStorage { + readonly sql: FakeSqlStorage; + alarmAt: number | null = null; + + constructor(db: DatabaseSync) { + this.sql = new FakeSqlStorage(db); + } + + async getAlarm(): Promise { + return this.alarmAt; + } + async setAlarm(time: number | Date): Promise { + this.alarmAt = typeof time === "number" ? time : time.getTime(); + } + async deleteAlarm(): Promise { + this.alarmAt = null; + } +} + +export interface FakeCtx { + id: { toString(): string; name?: string }; + storage: FakeStorage; + blockConcurrencyWhile(fn: () => Promise): Promise; + waitUntil(p: Promise): void; + /** Resolves once every blockConcurrencyWhile callback has settled. */ + ready(): Promise; +} + +function fakeCtx(id: string, storage: FakeStorage): FakeCtx { + const blocking: Promise[] = []; + return { + id: { toString: () => id }, + storage, + blockConcurrencyWhile(fn) { + const p = fn(); + blocking.push(p); + return p; + }, + waitUntil: () => {}, + async ready() { + await Promise.all(blocking); + }, + }; +} + +/** One AI call as the fake binding saw it. `kind` is decided from the payload + * shape the same way dev-fake-ai dispatches: embeddings carry `text`. */ +export interface AiCall { + model: string; + kind: "embed" | "chat"; + inputs: Record; +} + +/** Runs before the fake answers. Throw to fail the call; return a promise to + * hold it in flight. */ +export type AiInterceptor = (call: AiCall) => void | Promise; + +export interface FakeAi { + calls: AiCall[]; + intercept: AiInterceptor | null; + binding: { run(model: string, inputs: Record): Promise }; +} + +function fakeAi(): FakeAi { + const fake = fakeAiRunner({ latencyMs: 0 }); + const ai: FakeAi = { + calls: [], + intercept: null, + binding: { + async run(model, inputs) { + const call: AiCall = { model, kind: inputs.text !== undefined ? "embed" : "chat", inputs }; + ai.calls.push(call); + if (ai.intercept) await ai.intercept(call); + return fake.run(model, inputs); + }, + }, + }; + return ai; +} + +/** The Env a DO sees. AI resolves through the real `selectBudgetedAiRunner`: + * DEV_FAKE_AI is left unset so the Workers AI branch is taken and lands on + * the instrumented binding above, and the account budget always grants. */ +function fakeEnv(ai: FakeAi): Record { + let permits = 0; + return { + AI: ai.binding, + GEN_MODEL: "@cf/test/gen", + EMBED_MODEL: "@cf/test/embed", + ACCOUNT_BUDGET: { + getByName: () => ({ + acquireAi: async () => ({ granted: true, permitId: `permit-${++permits}` }), + releaseAi: async () => {}, + }), + }, + }; +} + +type DoClass = new (ctx: never, env: never) => T; + +export interface DoHarness { + /** The live instance. Replaced by `restart()`. */ + instance: T; + ctx: FakeCtx; + storage: FakeStorage; + db: DatabaseSync; + ai: FakeAi; + /** Evict and re-construct over the same storage: in-memory state is lost, + * SQLite and the alarm survive, and the constructor's hydrate runs again. */ + restart(): Promise; + /** Fire the pending alarm the way the runtime does: cleared first, then + * `alarm()` invoked. Returns false when no alarm was set. */ + fireAlarm(): Promise; + /** Fire alarms until none is pending — a pump that has filled its pool. + * Throws past `max` so a pump that never settles fails loudly. */ + drainAlarms(max?: number): Promise; + /** Raw rows, for asserting on what was actually written. */ + rows(query: string, ...bindings: unknown[]): R[]; +} + +export async function createDoHarness }>( + Class: DoClass, + { id = "test-do" }: { id?: string } = {}, +): Promise> { + const db = new DatabaseSync(":memory:"); + const storage = new FakeStorage(db); + const ai = fakeAi(); + const env = fakeEnv(ai); + + const construct = async () => { + const ctx = fakeCtx(id, storage); + const instance = new Class(ctx as never, env as never); + await ctx.ready(); + return { ctx, instance }; + }; + + const first = await construct(); + const harness: DoHarness = { + instance: first.instance, + ctx: first.ctx, + storage, + db, + ai, + async restart() { + const next = await construct(); + harness.instance = next.instance; + harness.ctx = next.ctx; + return next.instance; + }, + async fireAlarm() { + if (storage.alarmAt === null) return false; + storage.alarmAt = null; + await harness.instance.alarm(); + return true; + }, + async drainAlarms(max = 200) { + let fired = 0; + while (await harness.fireAlarm()) { + if (++fired > max) throw new Error(`alarm still pending after ${max} fires — the pump never settled`); + } + return fired; + }, + rows(query, ...bindings) { + return storage.sql.exec(query, ...bindings).toArray() as never; + }, + }; + return harness; +} diff --git a/test/session-do.test.ts b/test/session-do.test.ts new file mode 100644 index 0000000..dbef9cb --- /dev/null +++ b/test/session-do.test.ts @@ -0,0 +1,133 @@ +// SessionDO driven for real: its own SQL against a real (in-memory) SQLite, +// its own alarm pump, restarts that re-run hydrate(). The harness and why it +// is shaped this way are in test/helpers/do-harness.ts (issue #31). + +import { describe, expect, it, vi } from "vitest"; +import { SessionDO } from "../src/session-do"; +import type { Axis, SerializedAxis } from "../src/types"; +import { createDoHarness, type DoHarness } from "./helpers/do-harness"; + +vi.mock("cloudflare:workers", () => import("./helpers/cloudflare-workers-stub")); + +/** The in-memory axes, which carry `expanded` per pole. SerializedAxis folds + * it into `degraded`, so the round trip is asserted on the real field. */ +function coreAxes(session: SessionDO): Axis[] { + return (session as unknown as { core: { axes(): Axis[] } }).core.axes(); +} + +/** A session past init with its pump drained: the pool is full and no alarm + * is pending, so any alarm set afterwards was set by the code under test. + * That is the scenario the failed-embed kick exists for — against a full + * pool, nothing else would ever wake the pump. */ +async function settledSession(): Promise> { + const h = await createDoHarness(SessionDO); + await h.instance.init("s1", "security awareness"); + await h.drainAlarms(); + expect(h.storage.alarmAt).toBeNull(); + return h; +} + +function axisNamed(axes: SerializedAxis[] | null, negTerm: string): SerializedAxis { + const axis = axes?.find((a) => a.neg.term === negTerm); + if (!axis) throw new Error(`no axis with neg term ${negTerm}`); + return axis; +} + +describe("SessionDO harness", () => { + it("constructs, persists to SQLite, and hydrates the same session after a restart", async () => { + const h = await createDoHarness(SessionDO); + expect(await h.instance.getInfo()).toBeNull(); + + await h.instance.init("s1", " security awareness "); + const meta = h.rows<{ value: string }>("SELECT value FROM meta WHERE key = 'meta'"); + expect(JSON.parse(meta[0]!.value)).toMatchObject({ id: "s1", seed: "security awareness" }); + expect(h.storage.alarmAt).not.toBeNull(); // init kicks the first pass + + await h.drainAlarms(); + expect(h.rows("SELECT text FROM pool").length).toBeGreaterThan(0); + + const before = await h.instance.getInfo(); + const after = await (await h.restart()).getInfo(); + expect(after).toEqual(before); + }); +}); + +describe("SessionDO axes", () => { + it("an axis survives create → persist → restart → hydrate with expanded: false intact", async () => { + const h = await settledSession(); + // Fail only the "frozen" expansion, so this axis carries one unexpanded + // pole and one expanded one — both values of the boolean in one row. + h.ai.intercept = (call) => { + if (call.kind === "chat" && JSON.stringify(call.inputs).includes("frozen")) throw new Error("model down"); + }; + + const result = await h.instance.createAxis("frozen", "molten"); + expect(result).toMatchObject({ created: true }); + const [created] = coreAxes(h.instance); + expect(created!.neg).toMatchObject({ term: "frozen", phrase: "frozen", expanded: false }); + expect(created!.pos.expanded).toBe(true); + + const [row] = h.rows<{ neg_expanded: number; pos_expanded: number }>("SELECT neg_expanded, pos_expanded FROM axes"); + expect(row).toEqual({ neg_expanded: 0, pos_expanded: 1 }); + + const restarted = await h.restart(); + const [hydrated] = coreAxes(restarted); + expect(hydrated!.neg.expanded).toBe(false); + expect(hydrated!.pos.expanded).toBe(true); + // The embeddings came back through the BLOB column intact (at the float32 + // precision the column stores), so the axis is still serving coordinates — + // and still honestly reports its bare pole. + expect(hydrated!.neg.embedding).toEqual(Array.from(new Float32Array(created!.neg.embedding!))); + expect(axisNamed(await restarted.listAxes(), "frozen")).toMatchObject({ ready: true, degraded: true }); + }); + + it("a failed pole embed kicks the pump, and a later pass fills the embedding", async () => { + const h = await settledSession(); + let failed = false; + h.ai.intercept = (call) => { + if (call.kind === "embed" && !failed && JSON.stringify(call.inputs).includes("calm")) { + failed = true; + throw new Error("embed down"); + } + }; + + const result = await h.instance.createAxis("calm", "storm"); + expect(result).toMatchObject({ created: true }); + expect(failed).toBe(true); + expect(axisNamed(await h.instance.listAxes(), "calm").ready).toBe(false); + // Persisted unembedded, not dropped. + expect(h.rows("SELECT id FROM axes WHERE neg_embedding IS NULL AND pos_embedding IS NULL")).toHaveLength(1); + // The pool was full, so this alarm exists only because createAxis set it. + expect(h.storage.alarmAt).not.toBeNull(); + + await h.drainAlarms(); + expect(axisNamed(await h.instance.listAxes(), "calm").ready).toBe(true); + expect(h.rows("SELECT id FROM axes WHERE neg_embedding IS NULL OR pos_embedding IS NULL")).toHaveLength(0); + expect(axisNamed(await (await h.restart()).listAxes(), "calm").ready).toBe(true); + }); + + it("axisCreationsInFlight stops a second concurrent create at the cap from spending AI calls", async () => { + const h = await settledSession(); + await h.instance.createAxis("cold", "hot"); + await h.instance.createAxis("quiet", "loud"); + expect(await h.instance.listAxes()).toHaveLength(2); + + // Hold the first create's pole expansions in flight, so the second + // arrives while the axis count still reads 2 of 3. + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + h.ai.intercept = (call) => (call.kind === "chat" ? gate : undefined); + const callsBefore = h.ai.calls.length; + + const first = h.instance.createAxis("small", "large"); + const second = h.instance.createAxis("dull", "sharp"); + await new Promise((r) => setTimeout(r, 0)); // let both reach their awaits + // Only the first create's two expansions — the second paid for nothing. + expect(h.ai.calls.slice(callsBefore).filter((c) => c.kind === "chat")).toHaveLength(2); + + release(); + expect(await second).toMatchObject({ created: false, reason: "cap" }); + expect(await first).toMatchObject({ created: true }); + expect(await h.instance.listAxes()).toHaveLength(3); + }); +});