diff --git a/src/main/runtime/rpc/streaming.test.ts b/src/main/runtime/rpc/streaming.test.ts
index 382353de803..502614f3e87 100644
--- a/src/main/runtime/rpc/streaming.test.ts
+++ b/src/main/runtime/rpc/streaming.test.ts
@@ -4,6 +4,7 @@ import { RpcDispatcher } from './dispatcher'
 import { defineMethod, defineStreamingMethod, type RpcRequest } from './core'
 import type { OrcaRuntimeService } from '../orca-runtime'
 import { TERMINAL_METHODS } from './methods/terminal'
+import { createSubscriptionRegistryDouble } from './subscription-registry-test-double'
 import type { RuntimeTerminalWait } from '../../../shared/runtime-types'
 
 function stubRuntime(overrides: Partial<OrcaRuntimeService> = {}): OrcaRuntimeService {
@@ -296,7 +297,7 @@ describe('RpcDispatcher streaming', () => {
   it('ends terminal.subscribe when the backing terminal exits', async () => {
     const messages: string[] = []
     let resolveExit!: () => void
-    const cleanups = new Map<string, () => void>()
+    const registry = createSubscriptionRegistryDouble()
     const runtime = stubRuntime({
       resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
       readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
@@ -306,14 +307,12 @@ describe('RpcDispatcher streaming', () => {
       getLayout: vi.fn().mockReturnValue({ seq: 1 }),
       subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()),
       subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
-      registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
-        cleanups.set(id, cleanup)
-      }),
-      cleanupSubscription: vi.fn((id: string) => {
-        const cleanup = cleanups.get(id)
-        cleanups.delete(id)
-        cleanup?.()
-      }),
+      registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup),
+      registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup),
+      cleanupSubscription: vi.fn(registry.cleanupSubscription),
+      cleanupSubscriptionIfOwnedByConnection: vi.fn(
+        registry.cleanupSubscriptionIfOwnedByConnection
+      ),
       waitForTerminal: vi.fn(
         () =>
           new Promise<RuntimeTerminalWait>((resolve) => {
@@ -338,7 +337,7 @@ describe('RpcDispatcher streaming', () => {
       (msg) => messages.push(msg)
     )
 
-    await vi.waitFor(() => expect(cleanups.has('terminal-1:desktop-1')).toBe(true))
+    await vi.waitFor(() => expect(registry.peekCleanup('terminal-1:desktop-1')).toBeDefined())
     // Cleanup now registers before snapshot work so a disconnect cannot orphan
     // a desktop width floor; wait for the actual exit waiter before resolving it.
     await vi.waitFor(() => expect(runtime.waitForTerminal).toHaveBeenCalled())
@@ -346,6 +345,6 @@ describe('RpcDispatcher streaming', () => {
     await dispatchPromise
 
     expect(messages.some((msg) => JSON.parse(msg).result?.type === 'end')).toBe(true)
-    expect(runtime.cleanupSubscription).toHaveBeenCalledWith('terminal-1:desktop-1')
+    expect(registry.peekCleanup('terminal-1:desktop-1')).toBeUndefined()
   })
 })
diff --git a/src/main/runtime/rpc/terminal-provider-snapshot-sequence.test.ts b/src/main/runtime/rpc/terminal-provider-snapshot-sequence.test.ts
index 5d04ef24389..be0df0a1775 100644
--- a/src/main/runtime/rpc/terminal-provider-snapshot-sequence.test.ts
+++ b/src/main/runtime/rpc/terminal-provider-snapshot-sequence.test.ts
@@ -2,6 +2,7 @@ import { expect, it, vi } from 'vitest'
 import { RpcDispatcher } from './dispatcher'
 import type { RpcRequest } from './core'
 import { TERMINAL_METHODS } from './methods/terminal'
+import { createSubscriptionRegistryDouble } from './subscription-registry-test-double'
 import type { OrcaRuntimeService } from '../orca-runtime'
 import type { RuntimeTerminalWait } from '../../../shared/runtime-types'
 import {
@@ -23,7 +24,7 @@ const request: RpcRequest = {
 
 it('replays post-capture output in the provider snapshot sequence domain', async () => {
   const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
-  const cleanups = new Map<string, () => void>()
+  const registry = createSubscriptionRegistryDouble()
   let dataListener:
     | ((data: string, meta?: { seq?: number; rawLength?: number }) => void)
     | undefined
@@ -53,13 +54,10 @@ it('replays post-capture output in the provider snapshot sequence domain', async
     isTerminalAlternateScreen: vi.fn().mockReturnValue(false),
     subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
     subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
-    registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
-      cleanups.set(id, cleanup)
-    }),
-    cleanupSubscription: vi.fn((id: string) => {
-      cleanups.get(id)?.()
-      cleanups.delete(id)
-    }),
+    registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup),
+    registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup),
+    cleanupSubscription: vi.fn(registry.cleanupSubscription),
+    cleanupSubscriptionIfOwnedByConnection: vi.fn(registry.cleanupSubscriptionIfOwnedByConnection),
     waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
   } as unknown as OrcaRuntimeService
   const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
@@ -95,7 +93,7 @@ it('replays post-capture output in the provider snapshot sequence domain', async
 
 it('keeps provider-backed alternate-screen resizes geometry-only', async () => {
   const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
-  const cleanups = new Map<string, () => void>()
+  const registry = createSubscriptionRegistryDouble()
   let resizeListener:
     | ((event: {
         cols: number
@@ -126,14 +124,10 @@ it('keeps provider-backed alternate-screen resizes geometry-only', async () => {
       return vi.fn()
     }),
     subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
-    registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
-      cleanups.set(id, cleanup)
-    }),
-    cleanupSubscription: vi.fn((id: string) => {
-      const cleanup = cleanups.get(id)
-      cleanups.delete(id)
-      cleanup?.()
-    }),
+    registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup),
+    registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup),
+    cleanupSubscription: vi.fn(registry.cleanupSubscription),
+    cleanupSubscriptionIfOwnedByConnection: vi.fn(registry.cleanupSubscriptionIfOwnedByConnection),
     waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
     sendTerminal: vi.fn().mockResolvedValue({ accepted: true }),
     updateMobileViewport: vi.fn().mockResolvedValue({ updated: true, applied: true })