test(runtime): migrate the remaining subscription stubs to the faithful registry

streaming.test.ts and terminal-provider-snapshot-sequence.test.ts still stubbed
registerSubscriptionCleanup only, so terminal.subscribe's owned registration was
undefined and teardown never fired. Assert the registration is retired rather than
spying on cleanupSubscription, which the owned path now calls internally.

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()
   })
 })

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 })