fix(memory): close app-service leak paths behind the per-task memory ramp

Prod app tasks climb from ~3.0 GB to 9+ GB average (20.6 GB worst task)
with uptime and reset only on deploy. The growth lives in the main Next.js
process — the isolated-vm worker is a separate, bounded child and its
disposal already runs in finally on every path. Four fixes at the sites
that can actually accumulate there:

- copilot stream teardown (lib/copilot/request/lifecycle/start.ts): the
  activeStreams registration, the 250ms Redis abort poller, and the SSE
  keepalive were acquired outside the try whose finally releases them, so
  a throw before the lifecycle started (e.g. resetBuffer on a Redis blip)
  or a throw inside the ordered teardown orphaned two immortal intervals
  and the registration. Add an idempotent backstop in the orchestration's
  outer finally; the ordered teardown sets a flag so the normal path pays
  nothing.

- large-value cache (lib/execution/payloads/cache.ts): expiry was enforced
  only inside later cache calls, so on a quieting instance the last
  entries — parsed object graphs worth a multiple of their JSON-byte
  accounting — sat indefinitely instead of for the 15-minute TTL. Add a
  self-retiring, unref'd sweep interval, and export occupancy stats.

- memory telemetry (lib/monitoring/memory-telemetry.ts): add the
  large-value cache occupancy and detached-context count to the periodic
  snapshot so the JSON-bytes-vs-heap amplification and context retention
  are readable from the same log line as heapUsedMB.

- BYOK rotation cursors (lib/api-key/byok.ts): the tenant-keyed cursor Map
  had no delete, TTL, or ceiling. Bound it with an LRU; evicting an idle
  pool's cursor just restarts its rotation at index 0.

- collab-doc converter (lib/collab-doc/converter.ts): the DOM guard read
  the bundled module's `window` binding while the install wrote
  `globalThis.window` — the same bundler mismatch documented for TipTap in
  next.config.ts — so a runtime where the two disagree re-allocated a
  multi-MB jsdom window on every conversion. Guard and install now go
  through globalThis, and the jsdom window is a module singleton either
  way.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: icecrasher321

apps/sim/lib/api-key/byok.ts

@@ -2,6 +2,7 @@ import { db } from '@sim/db'
 import { organizationBYOKKeys, workspace, workspaceBYOKKeys } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, asc, eq, notExists } from 'drizzle-orm'
+import { LRUCache } from 'lru-cache'
 import { isOrganizationBYOKEntitledCached } from '@/lib/api-key/byok-entitlement'
 import { getRotatingApiKey } from '@/lib/core/config/api-keys'
 import { env } from '@/lib/core/config/env'
@@ -27,7 +28,13 @@ export interface BYOKKeyResult {
 
 export type BYOKKeyScopeName = 'workspace' | 'organization'
 
-const rotationCounters = new Map<string, number>()
+/**
+ * Bounded so tenant-keyed cursors cannot accumulate for the life of the
+ * process (one entry per workspace/organization × provider that ever rotated).
+ * Evicting an idle pool's cursor just restarts its rotation at index 0, which
+ * the per-instance, approximate-rotation contract already tolerates.
+ */
+const rotationCounters = new LRUCache<string, number>({ max: 10_000 })
 
 interface EncryptedBYOKKey {
   id: string

apps/sim/lib/collab-doc/converter.ts

@@ -45,6 +45,8 @@ function markdownSchema(): Schema {
   return cachedSchema
 }
 
+let cachedJsdomWindow: import('jsdom').DOMWindow | null = null
+
 /**
  * Ensure a DOM exists for the TipTap editor the markdown engine constructs. In a `jsdom`/browser
  * environment `window` + `document` already exist and this is a no-op; in a plain Node server it
@@ -56,20 +58,28 @@ function markdownSchema(): Schema {
  * `document`-only guard (plus a sticky flag) skipped this setup — leaving TipTap to throw "there is no
  * window object available". Re-checking the globals every call means a partial stub can never wedge it.
  * When `window` is missing we install a coherent jsdom window+document pair, overwriting any such stub.
+ *
+ * Both the guard and the install go through `globalThis` explicitly, and the jsdom window itself is a
+ * module-level singleton. The server bundler can give a bundled module a `window` binding that does
+ * NOT read `globalThis` (the documented reason TipTap/Yjs sit in `serverExternalPackages` — see
+ * `next.config.ts`); a bare-`window` guard paired with a `globalThis.window` install can therefore
+ * disagree forever, re-entering the install on every call. Reading and writing the same object makes
+ * the guard self-consistent, and the singleton caps this module at ONE jsdom window (megabytes each)
+ * per process even if some runtime still defeats the guard.
  */
 function ensureDomForTipTap(): void {
-  if (typeof window !== 'undefined' && typeof document !== 'undefined') return
-  // Lazy require so the client bundle never pulls jsdom in. Bind to `jsdomWindow`, NOT `window` — a
-  // local `const window` would shadow the global and put the `typeof window` guard above in its
-  // temporal dead zone ("Cannot access 'window' before initialization").
-  const { JSDOM } = require('jsdom') as typeof import('jsdom')
-  const { window: jsdomWindow } = new JSDOM('<!doctype html><html><body></body></html>')
+  if (typeof globalThis.window !== 'undefined' && typeof globalThis.document !== 'undefined') return
+  if (!cachedJsdomWindow) {
+    // Lazy require so the client bundle never pulls jsdom in.
+    const { JSDOM } = require('jsdom') as typeof import('jsdom')
+    cachedJsdomWindow = new JSDOM('<!doctype html><html><body></body></html>').window
+  }
   // double-cast-allowed: assigning the jsdom shims onto the global needs an
   // index-signature view of `globalThis`, whose declared type has none.
   const g = globalThis as unknown as Record<string, unknown>
-  g.window = jsdomWindow
-  g.document = jsdomWindow.document
-  g.navigator ??= jsdomWindow.navigator
+  g.window = cachedJsdomWindow
+  g.document = cachedJsdomWindow.document
+  g.navigator ??= cachedJsdomWindow.navigator
 }
 
 /** Convert a file's markdown to a fresh collaborative {@link Y.Doc} (cold-start seed). */

apps/sim/lib/copilot/request/lifecycle/start.test.ts

@@ -25,7 +25,9 @@ const {
   appendEvent,
   cleanupAbortMarker,
   hasAbortMarker,
+  registerActiveStream,
   releasePendingChatStream,
+  unregisterActiveStream,
   fetchGo,
 } = vi.hoisted(() => ({
   runCopilotLifecycle: vi.fn(),
@@ -39,7 +41,9 @@ const {
   appendEvent: vi.fn(),
   cleanupAbortMarker: vi.fn(),
   hasAbortMarker: vi.fn(),
+  registerActiveStream: vi.fn(),
   releasePendingChatStream: vi.fn(),
+  unregisterActiveStream: vi.fn(),
   fetchGo: vi.fn(),
 }))
 
@@ -77,8 +81,8 @@ vi.mock('@/lib/copilot/request/session', () => ({
   cleanupAbortMarker,
   hasAbortMarker,
   releasePendingChatStream,
-  registerActiveStream: vi.fn(),
-  unregisterActiveStream: vi.fn(),
+  registerActiveStream,
+  unregisterActiveStream,
   startAbortPoller: vi.fn().mockReturnValue(setInterval(() => {}, 999999)),
   isExplicitStopReason: vi.fn().mockReturnValue(false),
   SSE_RESPONSE_HEADERS: {},
@@ -325,6 +329,32 @@ describe('createSSEStream terminal error handling', () => {
     expect(lifecycleTraceparent).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-0[0-9a-f]$/)
   })
 
+  it('releases the stream registration and pollers when the session reset fails before the lifecycle starts', async () => {
+    resetBuffer.mockRejectedValue(new Error('redis down'))
+
+    const stream = createSSEStream({
+      requestPayload: { message: 'hello' },
+      userId: 'user-1',
+      streamId: 'stream-leak',
+      executionId: 'exec-leak',
+      runId: 'run-leak',
+      chatId: 'chat-leak',
+      currentChat: null,
+      isNewChat: false,
+      message: 'hello',
+      titleModel: 'gpt-5.4',
+      requestId: 'req-leak',
+      orchestrateOptions: {},
+    })
+
+    await expect(drainStream(stream)).rejects.toThrow('redis down')
+
+    expect(runCopilotLifecycle).not.toHaveBeenCalled()
+    expect(registerActiveStream).toHaveBeenCalledWith('stream-leak', expect.any(AbortController))
+    expect(unregisterActiveStream).toHaveBeenCalledWith('stream-leak')
+    expect(releasePendingChatStream).toHaveBeenCalledWith('chat-leak', 'stream-leak')
+  })
+
   it('does not scan manually authored title input against unrelated active secrets', async () => {
     runCopilotLifecycle.mockResolvedValue({
       success: true,

apps/sim/lib/copilot/request/lifecycle/start.ts

@@ -122,6 +122,17 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
 
   const publisher = new StreamWriter({ streamId, chatId, requestId })
 
+  // Declared at function scope (same rationale as `cancelReason` below) so the
+  // leak backstop in the orchestration's outer finally can always reach them:
+  // the stream registration above, the abort poller, and the keepalive are
+  // process-held resources, and a throw that bypasses the inner finally's
+  // ordered teardown (e.g. `resetBuffer` failing on a Redis blip before the
+  // lifecycle starts) previously orphaned them — the poller and keepalive
+  // intervals then ran, and the activeStreams entry sat, for the life of the
+  // process.
+  let abortPoller: ReturnType<typeof startAbortPoller> | undefined
+  let processResourcesReleased = false
+
   // Classify cancel: signal.reason (explicit-stop set) wins, then
   // clientDisconnected, else Unknown (latent contract bug — log it).
   const recordCancelled = (errorMessage?: string): CopilotRequestCancelReasonValue => {
@@ -220,7 +231,7 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
             })
           }
 
-          const abortPoller = startAbortPoller(streamId, abortController, {
+          abortPoller = startAbortPoller(streamId, abortController, {
             requestId,
             chatId,
           })
@@ -347,6 +358,7 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
             if (chatId) {
               await releasePendingChatStream(chatId, streamId)
             }
+            processResourcesReleased = true
             await scheduleBufferCleanup(streamId)
             await scheduleFilePreviewSessionCleanup(streamId)
             await cleanupAbortMarker(streamId)
@@ -371,6 +383,24 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
           rootError = error
           throw error
         } finally {
+          // Leak backstop for throws that bypassed the inner finally's
+          // ordered teardown (a session reset failing before the lifecycle
+          // started, or the teardown itself throwing before its release
+          // lines). Every step is idempotent — clearInterval and
+          // stopKeepalive no-op when already stopped, unregister is a keyed
+          // delete, and the chat-stream release is ownership-guarded against
+          // a successor stream — and none of them throw, so the otel finish
+          // below always still runs. On the normal path the flag set by the
+          // ordered teardown skips this entirely.
+          if (!processResourcesReleased) {
+            processResourcesReleased = true
+            clearInterval(abortPoller)
+            publisher.stopKeepalive()
+            unregisterActiveStream(streamId)
+            if (chatId) {
+              await releasePendingChatStream(chatId, streamId)
+            }
+          }
           // `finish` is idempotent, so it's safe whether the POST
           // handler started the root (and may also call finish on an
           // error path before the stream ran) or we did. The cancel

apps/sim/lib/execution/payloads/cache.test.ts

@@ -0,0 +1,51 @@
+/**
+ * @vitest-environment node
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  cacheLargeValue,
+  clearLargeValueCacheForTests,
+  getLargeValueCacheStats,
+} from '@/lib/execution/payloads/cache'
+
+describe('large value cache sweep', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+    clearLargeValueCacheForTests()
+  })
+
+  afterEach(() => {
+    clearLargeValueCacheForTests()
+    vi.useRealTimers()
+  })
+
+  it('drains expired entries without further cache traffic', () => {
+    expect(
+      cacheLargeValue('lv_sweep', { data: 'x'.repeat(64) }, 64, { executionId: 'exec-1' })
+    ).toBe(true)
+    expect(getLargeValueCacheStats()).toEqual({ entries: 1, trackedBytes: 64 })
+
+    vi.advanceTimersByTime(16 * 60 * 1000)
+
+    expect(getLargeValueCacheStats()).toEqual({ entries: 0, trackedBytes: 0 })
+  })
+
+  it('retires the sweep timer once the cache drains and re-arms on the next insert', () => {
+    cacheLargeValue('lv_a', { data: 1 }, 8, { executionId: 'exec-1' })
+    expect(vi.getTimerCount()).toBe(1)
+
+    vi.advanceTimersByTime(16 * 60 * 1000)
+    expect(vi.getTimerCount()).toBe(0)
+
+    cacheLargeValue('lv_b', { data: 2 }, 8, { executionId: 'exec-1' })
+    expect(vi.getTimerCount()).toBe(1)
+  })
+
+  it('keeps unexpired entries readable across sweep ticks', () => {
+    cacheLargeValue('lv_live', { data: 'live' }, 16, { executionId: 'exec-1' })
+
+    vi.advanceTimersByTime(5 * 60 * 1000)
+
+    expect(getLargeValueCacheStats()).toEqual({ entries: 1, trackedBytes: 16 })
+  })
+})

apps/sim/lib/execution/payloads/cache.ts

@@ -6,6 +6,7 @@ import {
 
 const FALLBACK_TTL_MS = 15 * 60 * 1000
 const MAX_IN_MEMORY_BYTES = 256 * 1024 * 1024
+const SWEEP_INTERVAL_MS = 60 * 1000
 
 interface LargeValueCacheScope {
   workspaceId?: string
@@ -28,9 +29,26 @@ const inMemoryValues = new Map<
 >()
 let inMemoryBytes = 0
 
+let sweepTimer: ReturnType<typeof setInterval> | null = null
+
 export function clearLargeValueCacheForTests(): void {
   inMemoryValues.clear()
   inMemoryBytes = 0
+  if (sweepTimer) {
+    clearInterval(sweepTimer)
+    sweepTimer = null
+  }
+}
+
+/**
+ * Point-in-time occupancy of the in-memory large-value cache, for the periodic
+ * memory-telemetry snapshot. `trackedBytes` is the JSON-serialized accounting
+ * the admission/eviction budget uses — the retained heap of the parsed values
+ * is a multiple of it, which is exactly what comparing this line against
+ * `heapUsedMB` in the same snapshot is meant to expose.
+ */
+export function getLargeValueCacheStats(): { entries: number; trackedBytes: number } {
+  return { entries: inMemoryValues.size, trackedBytes: inMemoryBytes }
 }
 
 function cleanupExpiredValues(now = Date.now()): void {
@@ -42,6 +60,27 @@ function cleanupExpiredValues(now = Date.now()): void {
   }
 }
 
+/**
+ * Keeps the fallback TTL honest on quiet instances. Expiry was previously
+ * enforced only inside `cacheLargeValue`/`materializeLargeValueRefSync`, so on
+ * an instance that stopped storing large values the last entries — up to the
+ * full budget, held as parsed object graphs — sat in memory indefinitely
+ * instead of for `FALLBACK_TTL_MS`. The timer is unref'd so it never holds the
+ * process open, and retires itself once the cache drains (the next insert
+ * restarts it), so an idle process carries no interval at all.
+ */
+function ensureSweepTimer(): void {
+  if (sweepTimer) return
+  sweepTimer = setInterval(() => {
+    cleanupExpiredValues()
+    if (inMemoryValues.size === 0 && sweepTimer) {
+      clearInterval(sweepTimer)
+      sweepTimer = null
+    }
+  }, SWEEP_INTERVAL_MS)
+  sweepTimer.unref()
+}
+
 export function cacheLargeValue(
   id: string,
   value: unknown,
@@ -87,6 +126,7 @@ export function cacheLargeValue(
     expiresAt: Date.now() + FALLBACK_TTL_MS,
   })
   inMemoryBytes += size
+  ensureSweepTimer()
   return true
 }
 

apps/sim/lib/monitoring/memory-telemetry.ts

@@ -5,6 +5,7 @@
 
 import v8 from 'node:v8'
 import { createLogger } from '@sim/logger'
+import { getLargeValueCacheStats } from '@/lib/execution/payloads/cache'
 
 const logger = createLogger('MemoryTelemetry', { logLevel: 'INFO' })
 
@@ -19,6 +20,7 @@ export function startMemoryTelemetry(intervalMs = 60_000) {
   const timer = setInterval(() => {
     const mem = process.memoryUsage()
     const heap = v8.getHeapStatistics()
+    const largeValueCache = getLargeValueCacheStats()
 
     logger.info('Memory snapshot', {
       heapUsedMB: Math.round(mem.heapUsed / MB),
@@ -28,10 +30,13 @@ export function startMemoryTelemetry(intervalMs = 60_000) {
       arrayBuffersMB: Math.round(mem.arrayBuffers / MB),
       heapSizeLimitMB: Math.round(heap.heap_size_limit / MB),
       nativeContexts: heap.number_of_native_contexts,
+      detachedContexts: heap.number_of_detached_contexts,
       activeResources:
         typeof process.getActiveResourcesInfo === 'function'
           ? process.getActiveResourcesInfo().length
           : -1,
+      largeValueCacheEntries: largeValueCache.entries,
+      largeValueCacheTrackedMB: Math.round(largeValueCache.trackedBytes / MB),
       uptimeMin: Math.round(process.uptime() / 60),
     })
   }, intervalMs)