diff --git a/src/web/server.ts b/src/web/server.ts index 9180dab..8ece6cc 100644 --- a/src/web/server.ts +++ b/src/web/server.ts @@ -57,6 +57,11 @@ import { import { stripeConfig, verifyStripeSignature, classifyStripeEvent, createCheckoutSession, sessionIdForPaymentIntent, STRIPE_PACKS } from "../billing/stripe.js"; import { ACCOUNT_HTML } from "./account-page.js"; import { handleGateway } from "./gateway.js"; +import { + TenantRuntimeRegistry, + tenantRuntimeOptions, + type TenantRuntimeLease, +} from "./tenant-runtime.js"; import { readCappedText, BodyTooLargeError, @@ -420,21 +425,6 @@ export async function startWebServer(opts: WebServerOptions): Promise(); - promptCache.set(lisaGlobalHome(), { fp: initialFingerprint, text: snapshot.text }); - const rebuildPrompt = async (): Promise<{ text: string; fingerprint: string }> => { - const key = lisaHome(); - const fp = await getPromptFingerprint(); - const cached = promptCache.get(key); - if (cached && fp === cached.fp) return { text: cached.text, fingerprint: fp }; - const next = await buildSystemPromptSnapshot(); - promptCache.set(key, { fp, text: next.text }); - return { text: next.text, fingerprint: fp }; - }; // Resume previous chat across restarts. Three-tier fallback: // 1. ~/.lisa/active-web-session.txt (set on every web startup) // 2. Most recent session on disk in this cwd (catches the case where the @@ -476,21 +466,38 @@ export async function startWebServer(opts: WebServerOptions): Promise; + prompt?: { fp: string; text: string }; } - const globalChat: ChatCtx = { session, history, chain: Promise.resolve() }; - const uidChats = new Map(); - const ctxForRequest = async (): Promise => { + const globalChat: ChatCtx = { + session, + history, + chain: Promise.resolve(), + prompt: { fp: initialFingerprint, text: snapshot.text }, + }; + const tenantRuntimes = new TenantRuntimeRegistry(tenantRuntimeOptions()); + const ctxForRequest = async (): Promise> => { const scoped = homeScope.getStore(); - if (!scoped) return globalChat; - let ctx = uidChats.get(scoped); - if (!ctx) { + if (!scoped) return { value: globalChat, release: () => {} }; + return tenantRuntimes.acquire(scoped, async () => { const s = await resumeOrCreateWebSession(opts.model); await writeActiveWebSession(s.id); const { messages } = await s.readMessagePage(0, 9999); - ctx = { session: s, history: [...messages], chain: Promise.resolve() }; - uidChats.set(scoped, ctx); + return { session: s, history: [...messages], chain: Promise.resolve() }; + }); + }; + // Per-tenant prompt cache now shares the runtime's TTL/LRU lifecycle instead + // of growing independently forever. + const rebuildPrompt = async (): Promise<{ text: string; fingerprint: string }> => { + const scoped = homeScope.getStore(); + const runtime = scoped ? tenantRuntimes.peek(scoped) : globalChat; + if (!runtime) throw new Error("tenant runtime is not acquired"); + const fp = await getPromptFingerprint(); + if (runtime.prompt && fp === runtime.prompt.fp) { + return { text: runtime.prompt.text, fingerprint: fp }; } - return ctx; + const next = await buildSystemPromptSnapshot(); + runtime.prompt = { fp, text: next.text }; + return { text: next.text, fingerprint: fp }; }; // Lazy per-user birth (B2, hub'd in S3): a signed-in user's first request @@ -515,7 +522,8 @@ export async function startWebServer(opts: WebServerOptions): Promise birth({ model: opts.model, onStep: emit })).promise; - promptCache.delete(lisaHome()); // pick the newborn soul up next turn + const runtime = tenantRuntimes.peek(lisaHome()); + if (runtime) runtime.prompt = undefined; // pick the newborn soul up next turn console.error(`[accounts] soul born for ${uid}`); } catch (e) { console.error(`[accounts] birth failed for ${uid}: ${(e as Error).message}`); @@ -1805,8 +1813,7 @@ export async function startWebServer(opts: WebServerOptions): Promise { } if (req.method === "GET" && url === "/session") { - const chat = await ctxForRequest(); - res.writeHead(200, { "content-type": "application/json" }); - res.end(JSON.stringify({ id: chat.session.id, model: opts.model })); + const runtime = await ctxForRequest(); + try { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ id: runtime.value.session.id, model: opts.model })); + } finally { + runtime.release(); + } return; } if (req.method === "GET" && url === "/events") { - const chat = await ctxForRequest(); + const runtime = await ctxForRequest(); + const sessionId = runtime.value.session.id; + runtime.release(); res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-cache", connection: "keep-alive", }); - res.write(`data: ${JSON.stringify({ type: "hello", session: chat.session.id })}\n\n`); + res.write(`data: ${JSON.stringify({ type: "hello", session: sessionId })}\n\n`); // Send current mood right away res.write(`data: ${JSON.stringify({ type: "mood", slug: moodBus.current() })}\n\n`); // Pin this subscriber to the account it authenticated as (B2). null on the @@ -2882,9 +2895,15 @@ self.addEventListener('fetch', (event) => { const qs = new URL(url, "http://localhost").searchParams; const page = Math.max(0, parseInt(qs.get("page") ?? "0", 10)); const pageSize = 20; - const { messages, hasMore } = await (await ctxForRequest()).session.readMessagePage(page, pageSize); - res.writeHead(200, { "content-type": "application/json" }); - res.end(JSON.stringify({ messages, hasMore, page })); + const runtime = await ctxForRequest(); + try { + const { messages, hasMore } = + await runtime.value.session.readMessagePage(page, pageSize); + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ messages, hasMore, page })); + } finally { + runtime.release(); + } return; } @@ -3253,7 +3272,8 @@ self.addEventListener('fetch', (event) => { run.listeners.add(listener); try { await run.promise; - promptCache.delete(lisaHome()); // pick the newborn soul up next turn + const runtime = tenantRuntimes.peek(lisaHome()); + if (runtime) runtime.prompt = undefined; // pick the newborn soul up next turn send({ kind: "done", message: "she is alive" }); } catch (err) { send({ kind: "error", message: (err as Error).message }); @@ -3348,7 +3368,6 @@ self.addEventListener('fetch', (event) => { } if (req.method === "POST" && url === "/chat") { - const chat = await ctxForRequest(); const body = await readRequestText(req, res, RICH_BODY_LIMIT); if (body === null) return; let message: string; @@ -3396,6 +3415,14 @@ self.addEventListener('fetch', (event) => { res.end(JSON.stringify({ error: "billing_state_unavailable" })); return; } + let runtimeLease: TenantRuntimeLease; + try { + runtimeLease = await ctxForRequest(); + } catch (err) { + await inferencePermit?.release(); + throw err; + } + const chat = runtimeLease.value; // User just talked — reset the idle watcher + stamp focus freshness. try { getIdleWatcher(60 * 60_000).tick(); } catch {} lastUserMessageAt = Date.now(); @@ -3568,12 +3595,14 @@ self.addEventListener('fetch', (event) => { await job; } finally { await inferencePermit?.release(); + runtimeLease.release(); } return; } if (req.method === "POST" && url === "/reflect") { - const chat = await ctxForRequest(); + const runtime = await ctxForRequest(); + const chat = runtime.value; try { const r = await reflectOnSession({ history: chat.history, @@ -3585,6 +3614,8 @@ self.addEventListener('fetch', (event) => { } catch (err) { res.writeHead(500); res.end((err as Error).message); + } finally { + runtime.release(); } return; } diff --git a/src/web/tenant-runtime.test.ts b/src/web/tenant-runtime.test.ts new file mode 100644 index 0000000..ee73c29 --- /dev/null +++ b/src/web/tenant-runtime.test.ts @@ -0,0 +1,124 @@ +import { describe, test } from "node:test"; +import assert from "node:assert/strict"; + +const { TenantRuntimeRegistry, tenantRuntimeOptions } = await import("./tenant-runtime.js"); + +describe("TenantRuntimeRegistry", () => { + test("creates one runtime for overlapping requests to the same tenant", async () => { + const registry = new TenantRuntimeRegistry({ + ttlMs: 1000, + maxEntries: 10, + }); + let creates = 0; + const create = async () => { + creates++; + await new Promise((resolve) => setTimeout(resolve, 10)); + return "runtime"; + }; + const [a, b] = await Promise.all([ + registry.acquire("u-1", create), + registry.acquire("u-1", create), + ]); + assert.equal(creates, 1); + assert.equal(a.value, b.value); + assert.equal(registry.stats().pinned, 1); + a.release(); + b.release(); + assert.equal(registry.stats().pinned, 0); + }); + + test("evicts idle entries by LRU while preserving pinned runtimes", async () => { + let now = 0; + const registry = new TenantRuntimeRegistry({ + ttlMs: 10_000, + maxEntries: 2, + now: () => now, + }); + const a = await registry.acquire("a", async () => "a"); + now = 1; + const b = await registry.acquire("b", async () => "b"); + b.release(); + now = 2; + const c = await registry.acquire("c", async () => "c"); + c.release(); + + assert.equal(registry.peek("a"), "a"); + assert.equal(registry.peek("b"), undefined); + assert.equal(registry.peek("c"), "c"); + assert.equal(registry.stats().entries, 2); + a.release(); + }); + + test("expires idle runtimes but not a runtime in active use", async () => { + let now = 0; + const registry = new TenantRuntimeRegistry({ + ttlMs: 100, + maxEntries: 10, + now: () => now, + }); + const active = await registry.acquire("active", async () => "active"); + const idle = await registry.acquire("idle", async () => "idle"); + idle.release(); + now = 101; + registry.sweep(); + assert.equal(registry.peek("idle"), undefined); + assert.equal(registry.peek("active"), "active"); + active.release(); + }); + + test("release is idempotent and failed creation is retryable", async () => { + const registry = new TenantRuntimeRegistry({ + ttlMs: 1000, + maxEntries: 10, + }); + await assert.rejects( + () => registry.acquire("u", async () => { + throw new Error("boom"); + }), + /boom/, + ); + const lease = await registry.acquire("u", async () => "ok"); + lease.release(); + lease.release(); + assert.equal(registry.stats().pinned, 0); + }); + + test("deleting during creation prevents a runtime from being resurrected", async () => { + const registry = new TenantRuntimeRegistry({ + ttlMs: 1000, + maxEntries: 10, + }); + let finish!: (value: string) => void; + const creating = registry.acquire( + "u", + () => new Promise((resolve) => { + finish = resolve; + }), + ); + await Promise.resolve(); + assert.equal(registry.delete("u"), true); + finish("stale"); + await assert.rejects(creating, /invalidated during creation/); + assert.equal(registry.peek("u"), undefined); + + const retry = await registry.acquire("u", async () => "fresh"); + assert.equal(retry.value, "fresh"); + retry.release(); + }); +}); + +describe("tenantRuntimeOptions", () => { + test("uses bounded defaults and accepts positive overrides", () => { + assert.deepEqual(tenantRuntimeOptions({}), { + ttlMs: 30 * 60_000, + maxEntries: 100, + }); + assert.deepEqual( + tenantRuntimeOptions({ + LISA_TENANT_RUNTIME_TTL_MIN: "5", + LISA_TENANT_RUNTIME_MAX: "20", + }), + { ttlMs: 5 * 60_000, maxEntries: 20 }, + ); + }); +}); diff --git a/src/web/tenant-runtime.ts b/src/web/tenant-runtime.ts new file mode 100644 index 0000000..d23733d --- /dev/null +++ b/src/web/tenant-runtime.ts @@ -0,0 +1,152 @@ +/** + * Bounded lifecycle registry for per-tenant in-memory runtime state. + * + * Entries are created single-flight, pinned while a request uses them, and + * evicted by idle TTL then LRU. Pinned entries may temporarily exceed the + * configured capacity; the next release/sweep brings the registry back under + * the limit without allowing two live runtimes for one tenant. + */ +export interface TenantRuntimeOptions { + ttlMs: number; + maxEntries: number; + now?: () => number; +} + +export interface TenantRuntimeLease { + value: T; + release(): void; +} + +interface Entry { + value: T; + lastAccessAt: number; + pins: number; +} + +interface PendingCreation { + promise: Promise; + invalidated: boolean; +} + +export interface TenantRuntimeStats { + entries: number; + pinned: number; + creating: number; + evictions: number; +} + +export function tenantRuntimeOptions( + env: Record = process.env, +): TenantRuntimeOptions { + const ttlMinutes = Number(env.LISA_TENANT_RUNTIME_TTL_MIN); + const maxEntries = Number(env.LISA_TENANT_RUNTIME_MAX); + return { + ttlMs: (Number.isFinite(ttlMinutes) && ttlMinutes > 0 ? ttlMinutes : 30) * 60_000, + maxEntries: + Number.isInteger(maxEntries) && maxEntries > 0 + ? maxEntries + : 100, + }; +} + +export class TenantRuntimeRegistry { + private readonly entries = new Map>(); + private readonly creating = new Map>(); + private readonly now: () => number; + private evictions = 0; + + constructor(private readonly options: TenantRuntimeOptions) { + if (!Number.isFinite(options.ttlMs) || options.ttlMs <= 0) { + throw new Error("tenant runtime ttlMs must be positive"); + } + if (!Number.isInteger(options.maxEntries) || options.maxEntries <= 0) { + throw new Error("tenant runtime maxEntries must be a positive integer"); + } + this.now = options.now ?? Date.now; + } + + async acquire(key: string, create: () => Promise): Promise> { + this.sweep(); + let entry = this.entries.get(key); + if (!entry) { + let pending = this.creating.get(key); + if (!pending) { + pending = { promise: create(), invalidated: false }; + this.creating.set(key, pending); + } + try { + const value = await pending.promise; + if (pending.invalidated) { + throw new Error("tenant runtime invalidated during creation"); + } + entry = this.entries.get(key); + if (!entry) { + entry = { value, lastAccessAt: this.now(), pins: 0 }; + this.entries.set(key, entry); + } + } finally { + if (this.creating.get(key) === pending) this.creating.delete(key); + } + } + + entry.pins++; + entry.lastAccessAt = this.now(); + let released = false; + return { + value: entry.value, + release: () => { + if (released) return; + released = true; + entry!.pins = Math.max(0, entry!.pins - 1); + entry!.lastAccessAt = this.now(); + this.sweep(); + }, + }; + } + + peek(key: string): T | undefined { + const entry = this.entries.get(key); + if (!entry) return undefined; + entry.lastAccessAt = this.now(); + return entry.value; + } + + delete(key: string): boolean { + const pending = this.creating.get(key); + if (pending) { + pending.invalidated = true; + this.creating.delete(key); + } + return this.entries.delete(key) || pending !== undefined; + } + + sweep(): void { + const now = this.now(); + for (const [key, entry] of this.entries) { + if (entry.pins === 0 && now - entry.lastAccessAt >= this.options.ttlMs) { + this.entries.delete(key); + this.evictions++; + } + } + if (this.entries.size <= this.options.maxEntries) return; + const idle = [...this.entries.entries()] + .filter(([, entry]) => entry.pins === 0) + .sort((a, b) => a[1].lastAccessAt - b[1].lastAccessAt); + for (const [key] of idle) { + if (this.entries.size <= this.options.maxEntries) break; + this.entries.delete(key); + this.evictions++; + } + } + + stats(): TenantRuntimeStats { + let pinned = 0; + for (const entry of this.entries.values()) pinned += entry.pins > 0 ? 1 : 0; + return { + entries: this.entries.size, + pinned, + creating: this.creating.size, + evictions: this.evictions, + }; + } +}