diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts index 826958f26c7..b706bd7963a 100644 --- a/src/main/runtime/orca-runtime.test.ts +++ b/src/main/runtime/orca-runtime.test.ts @@ -34942,6 +34942,114 @@ describe('OrcaRuntimeService', () => { expect(replacementCleanup).toHaveBeenCalledTimes(1) }) + it('releases an owned subscription only while its registration still owns the id', async () => { + const runtime = createRuntime() + const oldCleanup = vi.fn() + const replacementCleanup = vi.fn() + + const oldRegistration = runtime.registerOwnedSubscriptionCleanup( + 'terminal:owned', + oldCleanup, + 'conn-old' + ) + + runtime.registerOwnedSubscriptionCleanup('terminal:owned', replacementCleanup, 'conn-new') + expect(oldCleanup).toHaveBeenCalledTimes(1) + + // The stale registration must not reach the replacement that now owns the id. + oldRegistration.releaseIfCurrent() + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(replacementCleanup).not.toHaveBeenCalled() + }) + + it('releases an owned subscription when the registration is still current', async () => { + const runtime = createRuntime() + const cleanup = vi.fn() + + const registration = runtime.registerOwnedSubscriptionCleanup('terminal:live', cleanup, 'conn') + registration.releaseIfCurrent() + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(cleanup).toHaveBeenCalledTimes(1) + // A second release is a no-op: the registration no longer owns the id. + registration.releaseIfCurrent() + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(cleanup).toHaveBeenCalledTimes(1) + }) + + it('refuses an unsubscribe from a connection that no longer owns the subscription', async () => { + const runtime = createRuntime() + const oldCleanup = vi.fn() + const replacementCleanup = vi.fn() + + runtime.registerSubscriptionCleanup('terminal:unsub', oldCleanup, 'conn-old') + runtime.registerSubscriptionCleanup('terminal:unsub', replacementCleanup, 'conn-new') + + expect(runtime.cleanupSubscriptionIfOwnedByConnection('terminal:unsub', 'conn-old')).toBe(false) + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(replacementCleanup).not.toHaveBeenCalled() + + expect(runtime.cleanupSubscriptionIfOwnedByConnection('terminal:unsub', 'conn-new')).toBe(true) + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(replacementCleanup).toHaveBeenCalledTimes(1) + }) + + it('reports an unregistered subscription as gone rather than refused', async () => { + const runtime = createRuntime() + + // Why it matters: a client retrying on `false` would otherwise chase a dead id. + expect(runtime.cleanupSubscriptionIfOwnedByConnection('terminal:missing', 'conn-a')).toBe(true) + }) + + it('reports a refusal even when a sibling id was merely absent', async () => { + const runtime = createRuntime() + const bareCleanup = vi.fn() + const compositeCleanup = vi.fn() + + // A clientless stream registers under the bare id; a client-scoped one under the composite. + runtime.registerSubscriptionCleanup('terminal-1', bareCleanup, 'conn-a') + runtime.registerSubscriptionCleanup('terminal-1:phone-1', compositeCleanup, 'conn-b') + + // conn-a owns the bare id but not the composite: one genuine teardown, one refusal. + expect(runtime.cleanupSubscriptionIfOwnedByConnection('terminal-1', 'conn-a')).toBe(true) + expect(runtime.cleanupSubscriptionIfOwnedByConnection('terminal-1:phone-1', 'conn-a')).toBe( + false + ) + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(bareCleanup).toHaveBeenCalledTimes(1) + expect(compositeCleanup).not.toHaveBeenCalled() + }) + + // Why: the lease-only branch's unguarded compensating handleMobileUnsubscribe is only + // safe while a viewport-less subscribe cannot yield to the macrotask queue. Pin it so + // adding an await to that path fails here instead of silently killing a live lease. + it('settles a viewport-less mobile subscribe without leaving the microtask queue', async () => { + const runtime = createRuntime() + let settled = false + + void runtime.handleMobileSubscribe('pty-lease', 'phone-1', undefined).then(() => { + settled = true + }) + // Drain microtasks only: any real await on this path leaves this unsettled. + for (let i = 0; i < 50; i += 1) { + await Promise.resolve() + } + + expect(settled).toBe(true) + }) + + it('tears down unconditionally for in-process callers that have no connection', async () => { + const runtime = createRuntime() + const cleanup = vi.fn() + + runtime.registerSubscriptionCleanup('terminal:inproc', cleanup, 'conn-owner') + expect(runtime.cleanupSubscriptionIfOwnedByConnection('terminal:inproc', undefined)).toBe(true) + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(cleanup).toHaveBeenCalledTimes(1) + }) + it('does not deliver or accept browser screencast frames before ready', async () => { const runtime = createRuntime() const done = deferred() diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 534eeaf0f66..39293ead80d 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -1754,6 +1754,13 @@ export type RuntimePtyDataAdmission = Readonly<{ completion: Promise }> +// Why: a subscription id is stable across reconnects, so holding the string is not +// proof of ownership. This handle is the only safe way to tear down a registration. +export type SubscriptionRegistration = Readonly<{ + /** Tears down only if this registration still owns the id; otherwise a no-op. */ + releaseIfCurrent: () => void +}> + export type RuntimeTerminalDataMeta = Readonly<{ seq?: number rawLength?: number @@ -13936,7 +13943,9 @@ export class OrcaRuntimeService { // Why: the stable id is about to belong to a newer connection; detach // the old owner before its asynchronous cleanup can overlap the rebind. this.removeSubscriptionConnectionIndex(subscriptionId) - this.cleanupSubscription(subscriptionId) + // Why: evict by the owner we captured, never by the key — a keyed evict + // would resolve to whoever holds the id at call time. + this.cleanupOwnedSubscription(subscriptionId, existing) } this.subscriptionCleanups.set(subscriptionId, cleanup) if (connectionId) { @@ -13950,12 +13959,67 @@ export class OrcaRuntimeService { } } + // Why: teardown keyed only by a string tears down whoever owns that key *now*. + // A mobile reconnect rebinds the stable `${terminal}:${clientId}` id, so a late + // teardown from the dead connection would kill the replacement stream (STA-4510). + // Callers that own a registration must go through this handle instead. + registerOwnedSubscriptionCleanup( + subscriptionId: string, + cleanup: () => void | Promise, + connectionId?: string + ): SubscriptionRegistration { + this.registerSubscriptionCleanup(subscriptionId, cleanup, connectionId) + return { + releaseIfCurrent: () => this.cleanupOwnedSubscription(subscriptionId, cleanup) + } + } + cleanupSubscription(subscriptionId: string): void { void this.cleanupSubscriptionAndWait(subscriptionId).catch((error) => { console.error(`[runtime] subscription cleanup failed for ${subscriptionId}:`, error) }) } + // Why: a client-supplied unsubscribe names a stable id it may no longer own — a + // reconnect or make-before-break migration rebinds that id to a newer connection, + // and honoring the stale message would kill the live stream (STA-4510). + // Returns whether the subscription was actually torn down. + cleanupSubscriptionIfOwnedByConnection( + subscriptionId: string, + connectionId: string | undefined + ): boolean { + // Why: an absent connectionId means a connection-less caller — the local + // unix-socket dispatch path, gated by the 0o600 metadata token. That tier keeps + // unconditional teardown authority; this guard scopes socket clients only. + if (!connectionId) { + this.cleanupSubscription(subscriptionId) + return true + } + // Why: an id with no registration is already gone, not a refusal. Reporting + // false there would tell a retrying client to keep chasing a dead id. + if (!this.subscriptionCleanups.has(subscriptionId)) { + return true + } + if (this.subscriptionConnectionByEntry.get(subscriptionId) !== connectionId) { + return false + } + this.cleanupSubscription(subscriptionId) + return true + } + + private cleanupOwnedSubscription( + subscriptionId: string, + expectedCleanup: () => void | Promise + ): void { + // Why: the ownership check is synchronous and cleanupSubscriptionAndWait re-reads + // the map before its first await, so nothing can rebind in between. Delegating + // preserves the in-flight join, the retain-on-failure, and the retry contract. + if (this.subscriptionCleanups.get(subscriptionId) !== expectedCleanup) { + return + } + this.cleanupSubscription(subscriptionId) + } + retrySubscriptionCleanupAfter( subscriptionId: string, cleanupOwner: () => void | Promise, diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts index 1b88b4afc73..9ef1426dbcb 100644 --- a/src/main/runtime/rpc/methods/terminal.ts +++ b/src/main/runtime/rpc/methods/terminal.ts @@ -2960,7 +2960,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ }) const subscriptionId = `${params.terminal}:${clientId}` // Why: chat needs the input-floor ack without registering a view subscriber or transporting duplicate PTY output. - runtime.registerSubscriptionCleanup( + const registration = runtime.registerOwnedSubscriptionCleanup( subscriptionId, () => { closed = true @@ -2972,23 +2972,28 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ ) void runtime .waitForTerminal(params.terminal, { condition: 'exit', signal }) - .then(() => runtime.cleanupSubscription(subscriptionId)) - .catch(() => runtime.cleanupSubscription(subscriptionId)) + .then(() => registration.releaseIfCurrent()) + .catch(() => registration.releaseIfCurrent()) try { // Why: a lease-only subscriber has no terminal view, so its cached viewport must never phone-fit the PTY. await runtime.handleMobileSubscribe(ptyId, clientId, undefined) if (closed || signal?.aborted) { // Why: a disconnect can win the awaited subscribe and resurrect mobile presence after cleanup already released it. + // Unguarded on purpose: this must still fire when our own cleanup already ran. + // Safe only because lease-only passes no viewport and both !viewport paths in + // handleMobileSubscribeInternal return with no await, so no rebind can land + // first. Adding an await there — or passing a viewport here — makes a + // superseded handler delete the replacement's (ptyId, clientId) presence. runtime.handleMobileUnsubscribe(ptyId, clientId) if (!closed) { - runtime.cleanupSubscription(subscriptionId) + registration.releaseIfCurrent() } return } emit({ type: 'subscribed', streamId: null, lines: [], truncated: false }) await streamClosed } catch (error) { - runtime.cleanupSubscription(subscriptionId) + registration.releaseIfCurrent() throw error } return @@ -3008,7 +3013,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ resolveStream = resolve }) // Why: register before viewport/snapshot awaits so a socket close can't orphan the stream listeners or its remote-desktop width floor. - runtime.registerSubscriptionCleanup( + const registration = runtime.registerOwnedSubscriptionCleanup( subscriptionId, () => { closed = true @@ -3039,13 +3044,13 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ ) } if (closed || signal?.aborted) { - runtime.cleanupSubscription(subscriptionId) + registration.releaseIfCurrent() return } const read = await runtime.readTerminal(params.terminal) const serialized = await serializeBudgetedMobileSnapshot(runtime, ptyId, false) if (closed || signal?.aborted) { - runtime.cleanupSubscription(subscriptionId) + registration.releaseIfCurrent() return } const size = runtime.getTerminalSize(ptyId) @@ -3094,11 +3099,11 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ // Why: bind the exit-waiter to the connection signal so socket close/error removes it instead of leaking until real exit. void runtime .waitForTerminal(params.terminal, { condition: 'exit', signal }) - .then(() => runtime.cleanupSubscription(subscriptionId)) - .catch(() => runtime.cleanupSubscription(subscriptionId)) + .then(() => registration.releaseIfCurrent()) + .catch(() => registration.releaseIfCurrent()) await streamClosed } catch (error) { - runtime.cleanupSubscription(subscriptionId) + registration.releaseIfCurrent() throw error } return @@ -3134,7 +3139,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ }) // Why: register cleanup before any await so a mid-subscribe disconnect still removes mobile presence; client-scoped ids also allow parallel desktop subscribers. const subscriptionId = clientId ? `${params.terminal}:${clientId}` : params.terminal - runtime.registerSubscriptionCleanup( + const registration = runtime.registerOwnedSubscriptionCleanup( subscriptionId, () => { outputBatcher?.flush() @@ -3156,10 +3161,12 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ connectionId ) // Why: bind the exit-waiter to the connection signal so socket close/error removes it instead of leaking until real exit. + // Why releaseIfCurrent: the signal aborts per socket, so after a reconnect rebinds + // this id a keyed teardown here would kill the replacement stream (STA-4510). void runtime .waitForTerminal(params.terminal, { condition: 'exit', signal }) - .then(() => runtime.cleanupSubscription(subscriptionId)) - .catch(() => runtime.cleanupSubscription(subscriptionId)) + .then(() => registration.releaseIfCurrent()) + .catch(() => registration.releaseIfCurrent()) const sendFrame = ( opcode: TerminalStreamOpcode, payload: Uint8Array = new Uint8Array(), @@ -3717,7 +3724,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ }) : () => {} } catch (error) { - runtime.cleanupSubscription(subscriptionId) + registration.releaseIfCurrent() throw error } @@ -3727,13 +3734,25 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [ defineMethod({ name: 'terminal.unsubscribe', params: TerminalUnsubscribe, - handler: async (params, { runtime }) => { + handler: async (params, { runtime, connectionId }) => { + // Why: only the connection that owns the subscription may retire it — a stale + // unsubscribe from the pre-reconnect socket names an id the replacement now owns. + let unsubscribed = runtime.cleanupSubscriptionIfOwnedByConnection( + params.subscriptionId, + connectionId + ) // Why: older builds send a bare-handle subscriptionId, so also try the reconstructed `${terminal}:${clientId}` composite key. - runtime.cleanupSubscription(params.subscriptionId) + // Why AND over the calls that ran: a clientless stream registers under the bare id + // and a client-scoped one under the composite, so either call can be the real + // teardown. Reporting false needs one genuine refusal, not merely a missing id. if (params.client && !params.subscriptionId.includes(':')) { - runtime.cleanupSubscription(`${params.subscriptionId}:${params.client.id}`) + unsubscribed = + runtime.cleanupSubscriptionIfOwnedByConnection( + `${params.subscriptionId}:${params.client.id}`, + connectionId + ) && unsubscribed } - return { unsubscribed: true } + return { unsubscribed } } }), defineMethod({ 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 { @@ -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 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((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/subscription-registry-test-double.ts b/src/main/runtime/rpc/subscription-registry-test-double.ts new file mode 100644 index 00000000000..b2618f6ddc3 --- /dev/null +++ b/src/main/runtime/rpc/subscription-registry-test-double.ts @@ -0,0 +1,168 @@ +import type { SubscriptionRegistration } from '../orca-runtime' + +type Cleanup = () => void | Promise + +export type SubscriptionRegistryDouble = { + registerSubscriptionCleanup: (id: string, cleanup: Cleanup, connectionId?: string) => void + registerOwnedSubscriptionCleanup: ( + id: string, + cleanup: Cleanup, + connectionId?: string + ) => SubscriptionRegistration + cleanupSubscription: (id: string) => void + cleanupSubscriptionIfOwnedByConnection: (id: string, connectionId: string | undefined) => boolean + cleanupSubscriptionsForConnection: (connectionId: string) => void + /** Test-only inspection; the runtime deliberately exposes no such accessor. */ + peekCleanup: (id: string) => Cleanup | undefined +} + +/** + * Faithful double of the runtime subscription registry (`OrcaRuntimeService`, + * `registerSubscriptionCleanup` through `cleanupSubscriptionsForConnection`). + * + * This mirrors production line-for-line, so it can drift. If you change + * `registerSubscriptionCleanup`, `cleanupSubscriptionAndWait`, + * `cleanupOwnedSubscription`, `cleanupSubscriptionIfOwnedByConnection`, or + * `cleanupSubscriptionsForConnection`, change this too — the runtime-level tests in + * `orca-runtime.test.ts` are what pin the real behavior; these doubles only pin routing. + * + * Why this exists: the ad-hoc `Map` stubs these tests used to carry never evicted the + * prior generation and had no connection index, so no test could observe a + * cross-generation teardown — which is how STA-4510 shipped. Anything exercising + * subscribe/teardown must use this instead of a bare Map. + */ +export function createSubscriptionRegistryDouble(): SubscriptionRegistryDouble { + const cleanups = new Map() + const inFlight = new Map }>() + const byConnection = new Map>() + const connectionByEntry = new Map() + + const removeIndex = (id: string): void => { + const connectionId = connectionByEntry.get(id) + if (!connectionId) { + return + } + connectionByEntry.delete(id) + const set = byConnection.get(connectionId) + if (!set) { + return + } + set.delete(id) + if (set.size === 0) { + byConnection.delete(connectionId) + } + } + + const cleanupAndWait = (id: string): Promise => { + const cleanup = cleanups.get(id) + if (!cleanup) { + return Promise.resolve() + } + // Mirrors cleanupSubscriptionAndWait: join an in-flight attempt for this exact owner. + const existing = inFlight.get(id) + if (existing?.cleanup === cleanup) { + return existing.promise + } + let result: void | Promise + try { + result = cleanup() + } catch (error) { + result = Promise.reject(error) + } + const promise = Promise.resolve(result) + .then(() => { + // Only the generation that registered this callback may retire it. + if (cleanups.get(id) !== cleanup) { + return + } + cleanups.delete(id) + removeIndex(id) + }) + .finally(() => { + if (inFlight.get(id)?.promise === promise) { + inFlight.delete(id) + } + }) + inFlight.set(id, { cleanup, promise }) + return promise + } + + // Failure retains the registration so it stays retryable, matching the runtime. + const cleanupSubscription = (id: string): void => { + void cleanupAndWait(id).catch(() => undefined) + } + + const cleanupOwned = (id: string, expected: Cleanup): void => { + if (cleanups.get(id) !== expected) { + return + } + cleanupSubscription(id) + } + + const registerSubscriptionCleanup = ( + id: string, + cleanup: Cleanup, + connectionId?: string + ): void => { + const existing = cleanups.get(id) + if (existing) { + removeIndex(id) + cleanupOwned(id, existing) + } + cleanups.set(id, cleanup) + if (!connectionId) { + return + } + let set = byConnection.get(connectionId) + if (!set) { + set = new Set() + byConnection.set(connectionId, set) + } + set.add(id) + connectionByEntry.set(id, connectionId) + } + + return { + registerSubscriptionCleanup, + registerOwnedSubscriptionCleanup: (id, cleanup, connectionId) => { + registerSubscriptionCleanup(id, cleanup, connectionId) + return { + releaseIfCurrent: () => cleanupOwned(id, cleanup) + } + }, + cleanupSubscription, + cleanupSubscriptionIfOwnedByConnection: (id, connectionId) => { + if (!connectionId) { + cleanupSubscription(id) + return true + } + // Mirrors the production early-out: an unregistered id is already gone, not refused. + if (!cleanups.has(id)) { + return true + } + if (connectionByEntry.get(id) !== connectionId) { + return false + } + cleanupSubscription(id) + return true + }, + cleanupSubscriptionsForConnection: (connectionId) => { + const set = byConnection.get(connectionId) + if (!set) { + return + } + for (const id of Array.from(set)) { + // A rebound id now belongs to another connection; skip it. + if (connectionByEntry.get(id) !== connectionId) { + set.delete(id) + continue + } + cleanupSubscription(id) + } + if (set.size === 0) { + byConnection.delete(connectionId) + } + }, + peekCleanup: (id) => cleanups.get(id) + } +} diff --git a/src/main/runtime/rpc/terminal-multiplex-ack-output-budget.test.ts b/src/main/runtime/rpc/terminal-multiplex-ack-output-budget.test.ts index 6aa18e5f245..3dc1f6b2918 100644 --- a/src/main/runtime/rpc/terminal-multiplex-ack-output-budget.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex-ack-output-budget.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from 'vitest' import { RpcDispatcher } from './dispatcher' import { TERMINAL_METHODS } from './methods/terminal' +import { createSubscriptionRegistryDouble } from './subscription-registry-test-double' import type { RuntimeTerminalWait } from '../../../shared/runtime-types' import { TerminalStreamOpcode, @@ -22,7 +23,7 @@ describe('terminal multiplex RPC', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void } = {} @@ -51,9 +52,8 @@ describe('terminal multiplex RPC', () => { subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()), getTerminalFitOverride: vi.fn().mockReturnValue(null), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { - cleanups.set(id, cleanup) - }), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) @@ -122,7 +122,7 @@ describe('terminal multiplex RPC', () => { expect(encodeSpy).not.toHaveBeenCalledWith(multibyteOutput) encodeSpy.mockRestore() - cleanups.get('terminal-multiplex:conn-multibyte-output-batch')?.() + registry.cleanupSubscription('terminal-multiplex:conn-multibyte-output-batch') await dispatchPromise } finally { vi.useRealTimers() @@ -136,7 +136,7 @@ describe('terminal multiplex RPC', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void } = {} @@ -165,14 +165,9 @@ describe('terminal multiplex RPC', () => { subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()), getTerminalFitOverride: vi.fn().mockReturnValue(null), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) @@ -280,7 +275,7 @@ describe('terminal multiplex RPC', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListeners = new Map< string, (data: string, meta?: { seq?: number; rawLength?: number }) => void @@ -315,14 +310,9 @@ describe('terminal multiplex RPC', () => { subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()), getTerminalFitOverride: vi.fn().mockReturnValue(null), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) @@ -514,7 +504,7 @@ describe('terminal multiplex RPC', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string) => void } = {} const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), @@ -534,14 +524,9 @@ describe('terminal multiplex RPC', () => { subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) diff --git a/src/main/runtime/rpc/terminal-multiplex-ack-overflow-recovery.test.ts b/src/main/runtime/rpc/terminal-multiplex-ack-overflow-recovery.test.ts index abd42cc7cfc..03327859c62 100644 --- a/src/main/runtime/rpc/terminal-multiplex-ack-overflow-recovery.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex-ack-overflow-recovery.test.ts @@ -99,7 +99,7 @@ describe('terminal multiplex RPC', () => { [sourceRange(flooded.length, flooded.length + trailing.length)] ) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -160,7 +160,7 @@ describe('terminal multiplex RPC', () => { expect(harness.commit).toHaveBeenCalledOnce() expect(harness.rollback).toHaveBeenCalledOnce() expect(harness.lifecycle).toEqual(['reserve', 'commit', 'rollback', 'cancel']) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -205,7 +205,7 @@ describe('terminal multiplex RPC', () => { expect(harness.commit).not.toHaveBeenCalled() expect(harness.rollback).toHaveBeenCalledOnce() expect(harness.lifecycle).toEqual(['reserve', 'rollback', 'cancel']) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -260,7 +260,7 @@ describe('terminal multiplex RPC', () => { expect(harness.commit).not.toHaveBeenCalled() expect(harness.rollback).toHaveBeenCalledOnce() expect(harness.lifecycle.slice(0, 3)).toEqual(['reserve', 'rollback', 'cancel']) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -298,7 +298,7 @@ describe('terminal multiplex RPC', () => { 'ack-pending-overflow' ) ).toBe(false) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -551,7 +551,7 @@ describe('terminal multiplex RPC', () => { await Promise.resolve() expect(serializeTerminalBuffer).toHaveBeenCalledTimes(2) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) diff --git a/src/main/runtime/rpc/terminal-multiplex-initial-snapshot-buffering.test.ts b/src/main/runtime/rpc/terminal-multiplex-initial-snapshot-buffering.test.ts index 68d69885af3..38117a7c762 100644 --- a/src/main/runtime/rpc/terminal-multiplex-initial-snapshot-buffering.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex-initial-snapshot-buffering.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from 'vitest' import { RpcDispatcher } from './dispatcher' import type { RuntimeTerminalDataMeta } from '../orca-runtime' import { TERMINAL_METHODS } from './methods/terminal' +import { createSubscriptionRegistryDouble } from './subscription-registry-test-double' import type { RuntimeTerminalWait } from '../../../shared/runtime-types' import { TerminalStreamOpcode, @@ -65,7 +66,7 @@ describe('terminal multiplex RPC', () => { ).toBe(true) ) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -74,7 +75,7 @@ describe('terminal multiplex RPC', () => { try { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void } = {} @@ -98,14 +99,9 @@ describe('terminal multiplex RPC', () => { subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) @@ -161,7 +157,7 @@ describe('terminal multiplex RPC', () => { try { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void } = {} @@ -195,14 +191,9 @@ describe('terminal multiplex RPC', () => { subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) @@ -268,7 +259,7 @@ describe('terminal multiplex RPC', () => { try { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void } = {} @@ -302,14 +293,9 @@ describe('terminal multiplex RPC', () => { subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) diff --git a/src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts b/src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts index 2cf20021ca0..b7fb20b2f73 100644 --- a/src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex-input-write-rejection.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from 'vitest' import { RpcDispatcher } from './dispatcher' 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' import { TerminalStreamOpcode, @@ -154,7 +155,7 @@ describe('terminal multiplex RPC', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const runtime = stubRuntime({ resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), @@ -172,9 +173,8 @@ describe('terminal multiplex RPC', () => { rows: 20 }), getDriver: vi.fn().mockReturnValue({ kind: 'mobile', clientId: 'phone-1' }), - registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { - cleanups.set(id, cleanup) - }), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) @@ -247,7 +247,7 @@ describe('terminal multiplex RPC', () => { ) expect(runtime.sendTerminal).not.toHaveBeenCalled() - cleanups.get('terminal-multiplex:conn-locked')?.() + registry.cleanupSubscription('terminal-multiplex:conn-locked') await dispatchPromise }) @@ -257,7 +257,7 @@ describe('terminal multiplex RPC', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const runtime = stubRuntime({ resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), @@ -271,14 +271,9 @@ describe('terminal multiplex RPC', () => { subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()), getTerminalFitOverride: vi.fn().mockReturnValue(null), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) @@ -349,7 +344,7 @@ describe('terminal multiplex RPC', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), @@ -361,14 +356,9 @@ describe('terminal multiplex RPC', () => { subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateDesktopViewport: vi.fn().mockResolvedValue(true) @@ -431,7 +421,7 @@ describe('terminal multiplex RPC', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const sendTerminal = vi.fn().mockRejectedValue(new Error('terminal_not_writable')) const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), @@ -444,10 +434,9 @@ describe('terminal multiplex RPC', () => { subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { - cleanups.set(id, cleanup) - }), - cleanupSubscription: vi.fn((id: string) => cleanups.get(id)?.()), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: sendTerminal as unknown as OrcaRuntimeService['sendTerminal'], updateDesktopViewport: vi.fn().mockResolvedValue(true) @@ -506,7 +495,7 @@ describe('terminal multiplex RPC', () => { // nothing past 12, so an unsolicited 17 is an unknown opcode on that wire. // A capable desktop subscriber shares the runtime and is driven second, so // its frame proves the rejection had already been processed for both. - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const sendTerminal = vi.fn().mockRejectedValue(new Error('terminal_not_writable')) const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), @@ -519,10 +508,9 @@ describe('terminal multiplex RPC', () => { subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { - cleanups.set(id, cleanup) - }), - cleanupSubscription: vi.fn((id: string) => cleanups.get(id)?.()), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: sendTerminal as unknown as OrcaRuntimeService['sendTerminal'], updateDesktopViewport: vi.fn().mockResolvedValue(true), diff --git a/src/main/runtime/rpc/terminal-multiplex-output-pause-and-viewport.test.ts b/src/main/runtime/rpc/terminal-multiplex-output-pause-and-viewport.test.ts index c26aa4a78b4..ef6087923bb 100644 --- a/src/main/runtime/rpc/terminal-multiplex-output-pause-and-viewport.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex-output-pause-and-viewport.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it, vi } from 'vitest' import { RpcDispatcher } from './dispatcher' import type { OrcaRuntimeService, RuntimeTerminalDataMeta } from '../orca-runtime' import { TERMINAL_METHODS } from './methods/terminal' +import { createSubscriptionRegistryDouble } from './subscription-registry-test-double' import type { RuntimeTerminalWait } from '../../../shared/runtime-types' import { TerminalStreamOpcode, @@ -114,7 +115,7 @@ describe('terminal multiplex RPC', () => { .join('') ).toContain('VISIBLE_MARKER') - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise expect(harness.handlers.size).toBe(0) }) @@ -180,7 +181,7 @@ describe('terminal multiplex RPC', () => { .join('') ).toContain('LEGACY_VISIBLE') - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -191,7 +192,7 @@ describe('terminal multiplex RPC', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() let resizeListener: | ((event: { cols: number @@ -232,9 +233,8 @@ describe('terminal multiplex RPC', () => { subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()), getTerminalFitOverride: vi.fn().mockReturnValue(null), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { - cleanups.set(id, cleanup) - }), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockImplementation(async (_handle, _action, options) => { options.reserveWrite('pty-1') @@ -356,13 +356,13 @@ describe('terminal multiplex RPC', () => { .map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : '')) expect(snapshotData).toEqual(['newer']) - cleanups.get('terminal-multiplex:conn-stale-multiplex-resize')?.() + registry.cleanupSubscription('terminal-multiplex:conn-stale-multiplex-resize') await dispatchPromise }) it('owns and releases a viewport floor for legacy JSON desktop streams', async () => { const messages: string[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), @@ -372,14 +372,9 @@ describe('terminal multiplex RPC', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) }) const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) @@ -443,7 +438,7 @@ describe('terminal multiplex RPC', () => { await vi.waitFor(() => expect(trace).toContain('driver-changed')) expect(trace.lastIndexOf('snapshot')).toBeLessThan(trace.indexOf('fit-override-changed')) expect(trace.lastIndexOf('snapshot')).toBeLessThan(trace.indexOf('driver-changed')) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) }) diff --git a/src/main/runtime/rpc/terminal-multiplex-pty-wait-capacity.test.ts b/src/main/runtime/rpc/terminal-multiplex-pty-wait-capacity.test.ts index 2c9743ae447..778cd022068 100644 --- a/src/main/runtime/rpc/terminal-multiplex-pty-wait-capacity.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex-pty-wait-capacity.test.ts @@ -143,7 +143,7 @@ describe('terminal multiplex RPC', () => { .filter((frame) => frame?.opcode === TerminalStreamOpcode.Error) .map((frame) => (frame ? decodeTerminalStreamText(frame.payload) : '')) ).toEqual([]) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -198,7 +198,7 @@ describe('terminal multiplex RPC', () => { expect(registerRemoteTerminalViewSubscriber).not.toHaveBeenCalled() expect(harness.handlers.has(7)).toBe(false) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -232,7 +232,7 @@ describe('terminal multiplex RPC', () => { expect(waitSignals[0]?.aborted).toBe(true) expect(waitSignals[1]?.aborted).toBe(false) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await vi.waitFor(() => expect(waitSignals[1]?.aborted).toBe(true)) await harness.dispatchPromise }) @@ -349,7 +349,7 @@ describe('terminal multiplex RPC', () => { expect(dataSubscriberCount).toBe(TERMINAL_MULTIPLEX_MAX_ACTIVE_STREAMS_PER_CONNECTION) expect(viewSubscriberCount).toBe(TERMINAL_MULTIPLEX_MAX_ACTIVE_STREAMS_PER_CONNECTION) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise expect(dataSubscriberCount).toBe(0) expect(viewSubscriberCount).toBe(0) @@ -447,7 +447,7 @@ describe('terminal multiplex RPC', () => { ).toHaveLength(1) }) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -470,7 +470,7 @@ describe('terminal multiplex RPC', () => { .map((frame) => decodeTerminalStreamFrame(frame)) .find((frame) => frame?.opcode === TerminalStreamOpcode.Error) expect(errorFrame && decodeTerminalStreamText(errorFrame.payload)).toBe('no_connected_pty') - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -491,7 +491,7 @@ describe('terminal multiplex RPC', () => { ) expect(runtime.waitForLeafPtyId).not.toHaveBeenCalled() expect(runtime.requestRendererTerminalTabMount).not.toHaveBeenCalled() - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) diff --git a/src/main/runtime/rpc/terminal-multiplex-snapshot-serialization.test.ts b/src/main/runtime/rpc/terminal-multiplex-snapshot-serialization.test.ts index 932dc51196a..4ce604ba27d 100644 --- a/src/main/runtime/rpc/terminal-multiplex-snapshot-serialization.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex-snapshot-serialization.test.ts @@ -161,7 +161,7 @@ describe('terminal multiplex RPC', () => { .join('') ).toBe('authoritative current screen\r\n') - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) diff --git a/src/main/runtime/rpc/terminal-multiplex-source-range-admission.test.ts b/src/main/runtime/rpc/terminal-multiplex-source-range-admission.test.ts index 12ffbcf8358..09cfa301d16 100644 --- a/src/main/runtime/rpc/terminal-multiplex-source-range-admission.test.ts +++ b/src/main/runtime/rpc/terminal-multiplex-source-range-admission.test.ts @@ -69,7 +69,7 @@ describe('terminal multiplex RPC', () => { (bytes) => decodeTerminalStreamFrame(bytes)?.opcode === TerminalStreamOpcode.SnapshotEnd ) expect(snapshotEndIndex).toBeGreaterThanOrEqual(0) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise } ) @@ -237,7 +237,7 @@ describe('terminal multiplex RPC', () => { ptyIdentities: 0 }) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise intake.dispose() } @@ -312,7 +312,7 @@ describe('terminal multiplex RPC', () => { expect(attach).not.toHaveBeenCalled() expect(reserve).not.toHaveBeenCalled() - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) @@ -568,7 +568,7 @@ describe('terminal multiplex RPC', () => { expect.objectContaining({ deliveryToken: 'token-2', sourceStartSu: 100 }) ]) - harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.() + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') await harness.dispatchPromise }) diff --git a/src/main/runtime/rpc/terminal-multiplex-test-harness.ts b/src/main/runtime/rpc/terminal-multiplex-test-harness.ts index 85ab630d5a0..39b3e3e2f66 100644 --- a/src/main/runtime/rpc/terminal-multiplex-test-harness.ts +++ b/src/main/runtime/rpc/terminal-multiplex-test-harness.ts @@ -4,6 +4,7 @@ import type { RpcRequest } from './core' import type { OrcaRuntimeService } from '../orca-runtime' import { TERMINAL_METHODS } from './methods/terminal' import type { RuntimeTerminalWait } from '../../../shared/runtime-types' +import { createSubscriptionRegistryDouble } from './subscription-registry-test-double' import { TerminalStreamOpcode, decodeTerminalStreamFrame, @@ -57,7 +58,7 @@ export function startDesktopMultiplexSubscribe( number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const runtime = stubRuntime({ readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), serializeTerminalBuffer: vi.fn().mockResolvedValue({ data: 'snapshot', cols: 120, rows: 40 }), @@ -70,12 +71,11 @@ export function startDesktopMultiplexSubscribe( subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()), getTerminalFitOverride: vi.fn().mockReturnValue(null), getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), - registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => { - cleanups.set(id, cleanup) - }), - cleanupSubscription: vi.fn((id: string) => { - cleanups.get(id)?.() - }), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), + cleanupSubscriptionIfOwnedByConnection: vi.fn(registry.cleanupSubscriptionIfOwnedByConnection), + cleanupSubscriptionsForConnection: vi.fn(registry.cleanupSubscriptionsForConnection), ...overrides, waitForTerminal: overrides.waitForTerminal ?? vi.fn(() => new Promise(() => {})) @@ -118,7 +118,7 @@ export function startDesktopMultiplexSubscribe( } } ) - return { messages, binaryFrames, handlers, cleanups, runtime, dispatchPromise } + return { messages, binaryFrames, handlers, registry, runtime, dispatchPromise } } export function sendDesktopMultiplexSubscribe( diff --git a/src/main/runtime/rpc/terminal-output-batching.test.ts b/src/main/runtime/rpc/terminal-output-batching.test.ts index c1ee755ee83..5493fee0fd2 100644 --- a/src/main/runtime/rpc/terminal-output-batching.test.ts +++ b/src/main/runtime/rpc/terminal-output-batching.test.ts @@ -3,6 +3,7 @@ import { RpcDispatcher } from './dispatcher' import 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' import { TerminalStreamOpcode, @@ -31,7 +32,7 @@ describe('terminal output batching', () => { vi.useFakeTimers() try { const messages: string[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string) => void } = {} const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), @@ -45,14 +46,9 @@ describe('terminal output batching', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) }) const dispatcher = new RpcDispatcher({ @@ -100,7 +96,7 @@ describe('terminal output batching', () => { try { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string) => void } = {} const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), @@ -119,14 +115,9 @@ describe('terminal output batching', () => { }), 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) => { - const cleanup = cleanups.get(id) - cleanups.delete(id) - cleanup?.() - }), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) @@ -192,7 +183,7 @@ describe('terminal output batching', () => { try { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string) => void } = {} let captureOutputFrames = false let firstOutputEncodeCount: number | undefined @@ -210,14 +201,9 @@ describe('terminal output batching', () => { }), 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) => { - const cleanup = cleanups.get(id) - cleanups.delete(id) - cleanup?.() - }), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) @@ -282,7 +268,7 @@ describe('terminal output batching', () => { number, (frame: NonNullable>) => void >() - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const write = vi.fn() const commit = vi.fn().mockResolvedValue(undefined) const rollback = vi.fn() @@ -304,13 +290,9 @@ describe('terminal output batching', () => { getDriver: vi.fn().mockReturnValue({ kind: 'idle' }), handleMobileSubscribe: vi.fn().mockResolvedValue(undefined), handleMobileUnsubscribe: 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockImplementation(async (_handle, _action, options) => { options.reserveWrite('pty-1') 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[] = [] - const cleanups = new Map 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(() => {})) } 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[] = [] - const cleanups = new Map 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(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue({ updated: true, applied: true }) diff --git a/src/main/runtime/rpc/terminal-subscribe-blank-mount.test.ts b/src/main/runtime/rpc/terminal-subscribe-blank-mount.test.ts index 89a46d36073..87fedfd51e9 100644 --- a/src/main/runtime/rpc/terminal-subscribe-blank-mount.test.ts +++ b/src/main/runtime/rpc/terminal-subscribe-blank-mount.test.ts @@ -4,6 +4,7 @@ import { RpcDispatcher } from './dispatcher' import 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 { @@ -64,7 +65,7 @@ describe('terminal.subscribe blank-tab background mount', () => { it('requests a renderer tab mount when a mobile subscribe has no headless model', async () => { // Why: stale preview text must not hide the missing live model/attachment. - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const callOrder: string[] = [] const unsubscribeData = vi.fn() const requestRendererTerminalTabMount = vi.fn(() => { @@ -91,13 +92,9 @@ describe('terminal.subscribe blank-tab background mount', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) }) const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) @@ -127,7 +124,7 @@ describe('terminal.subscribe blank-tab background mount', () => { }) it('does not request a renderer tab mount when an attached terminal is legitimately blank', async () => { - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const requestRendererTerminalTabMount = vi.fn(() => true) const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), @@ -143,13 +140,9 @@ describe('terminal.subscribe blank-tab background mount', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) }) const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) @@ -176,7 +169,7 @@ describe('terminal.subscribe blank-tab background mount', () => { }) it('does not wait for a remount when the current snapshot came from the renderer', async () => { - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const requestRendererTerminalTabMount = vi.fn(() => true) const waitForRendererTerminalSerializer = vi.fn() const runtime = stubRuntime({ @@ -200,13 +193,9 @@ describe('terminal.subscribe blank-tab background mount', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) }) const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) diff --git a/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts b/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts index 8265e713fb3..ad29d8a2643 100644 --- a/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts +++ b/src/main/runtime/rpc/terminal-subscribe-buffer.test.ts @@ -3,6 +3,7 @@ import { RpcDispatcher } from './dispatcher' import 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' import { TerminalStreamOpcode, @@ -82,7 +83,7 @@ describe('terminal subscribe buffering', () => { it('captures live queries before awaiting mobile fit and delivers them after the snapshot', async () => { const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() let dataListener: | ((data: string, meta?: { seq?: number; rawLength?: number }) => void) | undefined @@ -112,13 +113,9 @@ describe('terminal subscribe buffering', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) }) const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) @@ -199,7 +196,7 @@ describe('terminal subscribe buffering', () => { it('marks legacy scrollback previews truncated when the uncursored read is limited', async () => { const messages: string[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), readTerminal: vi.fn().mockResolvedValue({ @@ -213,13 +210,9 @@ describe('terminal subscribe buffering', () => { 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) => { - cleanups.get(id)?.() - cleanups.delete(id) - }), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), waitForTerminal: vi.fn(() => new Promise(() => {})) }) const dispatcher = new RpcDispatcher({ @@ -256,6 +249,7 @@ describe('terminal subscribe buffering', () => { try { const messages: string[] = [] const controller = new AbortController() + const registry = createSubscriptionRegistryDouble() let resolveSnapshot: (value: { data: string; cols: number; rows: number }) => void = () => {} const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), @@ -271,8 +265,9 @@ describe('terminal subscribe buffering', () => { getLayout: vi.fn().mockReturnValue({ seq: 1 }), subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()), subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), - registerSubscriptionCleanup: vi.fn(), - cleanupSubscription: vi.fn(), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), waitForTerminal: vi.fn(() => new Promise(() => {})) }) const dispatcher = new RpcDispatcher({ @@ -303,14 +298,15 @@ describe('terminal subscribe buffering', () => { expect(runtime.subscribeToFitOverrideChanges).not.toHaveBeenCalled() // Cleanup must exist before snapshot work so an abort cannot orphan a // desktop width floor, but the abort must consume it before listeners start. - expect(runtime.registerSubscriptionCleanup).toHaveBeenCalledWith( + expect(runtime.registerOwnedSubscriptionCleanup).toHaveBeenCalledWith( 'terminal-1:desktop-1', expect.any(Function), 'conn-legacy-json' ) - expect(runtime.cleanupSubscription).toHaveBeenCalledWith('terminal-1:desktop-1') + expect(registry.peekCleanup('terminal-1:desktop-1')).toBeUndefined() expect(runtime.waitForTerminal).not.toHaveBeenCalled() - expect(messages).toEqual([]) + // The consumed cleanup emits the terminating frame; no snapshot/data may leak past it. + expect(messages.map((msg) => JSON.parse(msg).result?.type)).toEqual(['end']) } finally { vi.useRealTimers() } @@ -319,7 +315,7 @@ describe('terminal subscribe buffering', () => { it('keeps a limited retained-tail fallback usable for binary first paint', async () => { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), readTerminal: vi.fn().mockResolvedValue({ @@ -334,13 +330,9 @@ describe('terminal subscribe buffering', () => { subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()), 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) @@ -390,7 +382,7 @@ describe('terminal subscribe buffering', () => { it('does not mark binary snapshot frames truncated from an overflowed read when serialized data is available', async () => { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const runtime = stubRuntime({ resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), readTerminal: vi.fn().mockResolvedValue({ @@ -409,13 +401,9 @@ describe('terminal subscribe buffering', () => { subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()), 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) @@ -462,7 +450,7 @@ describe('terminal subscribe buffering', () => { try { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void } = {} @@ -497,14 +485,9 @@ describe('terminal subscribe buffering', () => { }), 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) => { - const cleanup = cleanups.get(id) - cleanups.delete(id) - cleanup?.() - }), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) @@ -591,7 +574,7 @@ describe('terminal subscribe buffering', () => { it('drops stale mobile resize re-stream completions for legacy binary streams', async () => { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() let resizeListener: | ((event: { cols: number @@ -647,14 +630,9 @@ describe('terminal subscribe buffering', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue({ updated: true, applied: true }) @@ -742,7 +720,7 @@ describe('terminal subscribe buffering', () => { try { const messages: string[] = [] const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const dataListenerRef: { current?: (data: string, meta?: { seq?: number; rawLength?: number }) => void } = {} @@ -777,14 +755,9 @@ describe('terminal subscribe buffering', () => { }), 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) => { - const cleanup = cleanups.get(id) - cleanups.delete(id) - cleanup?.() - }), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), waitForTerminal: vi.fn(() => new Promise(() => {})), sendTerminal: vi.fn().mockResolvedValue({ accepted: true }), updateMobileViewport: vi.fn().mockResolvedValue(false) diff --git a/src/main/runtime/rpc/terminal-subscribe-lease-only.test.ts b/src/main/runtime/rpc/terminal-subscribe-lease-only.test.ts index 3aba4b58200..31425defeb9 100644 --- a/src/main/runtime/rpc/terminal-subscribe-lease-only.test.ts +++ b/src/main/runtime/rpc/terminal-subscribe-lease-only.test.ts @@ -4,6 +4,7 @@ import type { OrcaRuntimeService } from '../orca-runtime' import type { RpcRequest } from './core' import { RpcDispatcher } from './dispatcher' import { TERMINAL_METHODS } from './methods/terminal' +import { createSubscriptionRegistryDouble } from './subscription-registry-test-double' const request: RpcRequest = { id: 'req-1', @@ -20,7 +21,7 @@ const request: RpcRequest = { describe('terminal lease-only subscription', () => { it('keeps mobile input ownership without viewport resize or output delivery', async () => { const messages: string[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() const runtime = { getRuntimeId: () => 'test-runtime', resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), @@ -32,13 +33,9 @@ describe('terminal lease-only subscription', () => { serializeTerminalBuffer: vi.fn(), subscribeToTerminalResize: vi.fn(), subscribeToFitOverrideChanges: 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), waitForTerminal: vi.fn(() => new Promise(() => {})) } as unknown as OrcaRuntimeService const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) diff --git a/src/main/runtime/rpc/terminal-subscribe-mount-replay.test.ts b/src/main/runtime/rpc/terminal-subscribe-mount-replay.test.ts index 3a7614587f1..7533881f903 100644 --- a/src/main/runtime/rpc/terminal-subscribe-mount-replay.test.ts +++ b/src/main/runtime/rpc/terminal-subscribe-mount-replay.test.ts @@ -9,6 +9,7 @@ import type { OrcaRuntimeService } from '../orca-runtime' import type { RpcRequest } from './core' import { RpcDispatcher } from './dispatcher' import { TERMINAL_METHODS } from './methods/terminal' +import { createSubscriptionRegistryDouble } from './subscription-registry-test-double' const request: RpcRequest = { id: 'req-1', @@ -24,7 +25,7 @@ const request: RpcRequest = { describe('terminal subscribe mount replay', () => { it('includes the restored idle screen when a missing model is background-mounted', async () => { const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() let mounted = false const runtime = { getRuntimeId: () => 'test-runtime', @@ -59,13 +60,9 @@ describe('terminal subscribe mount replay', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) } as unknown as OrcaRuntimeService const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) @@ -95,7 +92,7 @@ describe('terminal subscribe mount replay', () => { it('prefers restored history when phone-fit creates suffix-only headless state', async () => { const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() let generation = 1 let headlessPresent = false let serializeCalls = 0 @@ -139,13 +136,9 @@ describe('terminal subscribe mount replay', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) } as unknown as OrcaRuntimeService const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) @@ -188,7 +181,7 @@ describe('terminal subscribe mount replay', () => { it('replays a late recovery when readiness lands after the bounded initial response', async () => { vi.useFakeTimers() const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() let mounted = false let signalWaitStarted!: () => void const waitStarted = new Promise((resolve) => { @@ -235,13 +228,9 @@ describe('terminal subscribe mount replay', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) } as unknown as OrcaRuntimeService const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) @@ -291,7 +280,7 @@ describe('terminal subscribe mount replay', () => { it('recovers from the pre-mount generation when suffix state appears during the PTY wait', async () => { const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() let generation = 0 let headlessPresent = false const requestRendererTerminalTabMount = vi.fn(() => { @@ -338,13 +327,9 @@ describe('terminal subscribe mount replay', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) } as unknown as OrcaRuntimeService const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) @@ -379,7 +364,7 @@ describe('terminal subscribe mount replay', () => { }) it('cancels the mount-ready wait when the mobile subscription closes', async () => { - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() let waitSignal: AbortSignal | undefined const runtime = { getRuntimeId: () => 'test-runtime', @@ -411,13 +396,9 @@ describe('terminal subscribe mount replay', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) } as unknown as OrcaRuntimeService const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) diff --git a/src/main/runtime/rpc/terminal-subscribe-ownership.test.ts b/src/main/runtime/rpc/terminal-subscribe-ownership.test.ts new file mode 100644 index 00000000000..cba7b181f2a --- /dev/null +++ b/src/main/runtime/rpc/terminal-subscribe-ownership.test.ts @@ -0,0 +1,256 @@ +import { describe, expect, it, vi } from 'vitest' +import type { RuntimeTerminalWait } from '../../../shared/runtime-types' +import type { OrcaRuntimeService } from '../orca-runtime' +import type { RpcRequest } from './core' +import { RpcDispatcher } from './dispatcher' +import { TERMINAL_METHODS } from './methods/terminal' +import { createSubscriptionRegistryDouble } from './subscription-registry-test-double' + +const SUBSCRIPTION_ID = 'terminal-1:phone-1' + +type Waiter = { resolve: (value: RuntimeTerminalWait) => void; reject: (reason: unknown) => void } + +function stubRuntime( + registry: ReturnType, + waiters: Waiter[], + overrides: Record = {} +): OrcaRuntimeService { + return { + getRuntimeId: () => 'test-runtime', + registerRemoteTerminalViewSubscriber: () => () => {}, + requestRendererTerminalTabMount: () => false, + resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), + handleMobileSubscribe: vi.fn().mockResolvedValue(true), + handleMobileUnsubscribe: vi.fn(), + subscribeToTerminalData: vi.fn(() => vi.fn()), + readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), + serializeTerminalBuffer: vi + .fn() + .mockResolvedValue({ data: 'snapshot', cols: 80, rows: 24, seq: 4 }), + getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }), + getMobileDisplayMode: vi.fn().mockReturnValue('auto'), + getLayout: vi.fn().mockReturnValue({ seq: 1 }), + isTerminalAlternateScreen: vi.fn().mockReturnValue(false), + subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), + subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), + cleanupSubscriptionIfOwnedByConnection: vi.fn(registry.cleanupSubscriptionIfOwnedByConnection), + // Mirrors bindTerminalWaiterAbort: abort rejects with request_aborted. + waitForTerminal: vi.fn( + (_handle: string, options?: { signal?: AbortSignal }) => + new Promise((resolve, reject) => { + waiters.push({ resolve, reject }) + options?.signal?.addEventListener('abort', () => reject(new Error('request_aborted')), { + once: true + }) + }) + ), + ...overrides + } as unknown as OrcaRuntimeService +} + +const makeRequest = (params: unknown): RpcRequest => ({ + id: 'req-1', + authToken: 'tok', + method: 'terminal.subscribe', + params +}) + +const binaryParams = { + terminal: 'terminal-1', + client: { id: 'phone-1', type: 'mobile' }, + capabilities: { terminalBinaryStream: 1 } +} + +const leaseOnlyParams = { + terminal: 'terminal-1', + client: { id: 'phone-1', type: 'mobile' }, + capabilities: { terminalBinaryStream: 1, mobileInputLeaseOnly: 1 } +} + +// No viewport: keeps the branch off the remote-desktop registration path so the test +// isolates the post-rebind snapshot continuation. +const legacyParams = { + terminal: 'terminal-1', + client: { id: 'phone-1', type: 'desktop' } +} + +const streamOptions = (connectionId: string, signal?: AbortSignal) => ({ + signal, + connectionId, + sendBinary: vi.fn(), + registerBinaryStreamHandler: vi.fn(() => vi.fn()) +}) + +const flush = (ms = 20): Promise => new Promise((resolve) => setTimeout(resolve, ms)) + +describe('terminal.subscribe teardown ownership', () => { + // T4(b): the anti-leak guard — an abort that is still the current owner must tear down. + it('tears down the current owner when its own connection aborts', async () => { + const registry = createSubscriptionRegistryDouble() + const waiters: Waiter[] = [] + const runtime = stubRuntime(registry, waiters) + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + const conn = new AbortController() + + void dispatcher.dispatchStreaming(makeRequest(binaryParams), vi.fn(), { + ...streamOptions('conn-a', conn.signal) + }) + await vi.waitFor(() => expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBeDefined()) + + conn.abort() + await flush() + + expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBeUndefined() + expect(runtime.handleMobileUnsubscribe).toHaveBeenCalledWith('pty-1', 'phone-1') + }) + + // T4(c): a genuine terminal-gone rejection still tears down. + it('tears down the current owner when the terminal handle goes stale', async () => { + const registry = createSubscriptionRegistryDouble() + const waiters: Waiter[] = [] + const runtime = stubRuntime(registry, waiters) + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + + void dispatcher.dispatchStreaming(makeRequest(binaryParams), vi.fn(), streamOptions('conn-a')) + await vi.waitFor(() => expect(waiters).toHaveLength(1)) + + waiters[0]!.reject(new Error('terminal_handle_stale')) + await flush() + + expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBeUndefined() + }) + + // T8: the exit-waiter is only half the story — stale async continuations must be owned too. + it('does not let a lease-only subscribe rejecting after a rebind retire the replacement', async () => { + const registry = createSubscriptionRegistryDouble() + const waiters: Waiter[] = [] + let failFirstSubscribe = (): void => {} + const handleMobileSubscribe = vi + .fn() + .mockImplementationOnce( + () => + new Promise((_resolve, reject) => { + failFirstSubscribe = () => reject(new Error('subscribe_failed')) + }) + ) + .mockResolvedValue(true) + const runtime = stubRuntime(registry, waiters, { handleMobileSubscribe }) + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + + void dispatcher + .dispatchStreaming(makeRequest(leaseOnlyParams), vi.fn(), streamOptions('conn-a')) + .catch(() => undefined) + await vi.waitFor(() => expect(handleMobileSubscribe).toHaveBeenCalledTimes(1)) + + void dispatcher.dispatchStreaming( + makeRequest(leaseOnlyParams), + vi.fn(), + streamOptions('conn-b') + ) + await vi.waitFor(() => expect(handleMobileSubscribe).toHaveBeenCalledTimes(2)) + const live = registry.peekCleanup(SUBSCRIPTION_ID) + + failFirstSubscribe() + await flush() + + expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBe(live) + }) + + it('does not let a legacy JSON snapshot resuming after a rebind retire the replacement', async () => { + const registry = createSubscriptionRegistryDouble() + const waiters: Waiter[] = [] + let releaseFirstRead = (): void => {} + const readTerminal = vi + .fn() + .mockImplementationOnce( + () => + new Promise((resolve) => { + releaseFirstRead = () => resolve({ tail: [], truncated: false }) + }) + ) + .mockResolvedValue({ tail: [], truncated: false }) + const runtime = stubRuntime(registry, waiters, { readTerminal }) + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + + void dispatcher.dispatchStreaming(makeRequest(legacyParams), vi.fn(), { + connectionId: 'conn-a' + }) + await vi.waitFor(() => expect(readTerminal).toHaveBeenCalledTimes(1)) + + void dispatcher.dispatchStreaming(makeRequest(legacyParams), vi.fn(), { + connectionId: 'conn-b' + }) + await vi.waitFor(() => expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBeDefined()) + await vi.waitFor(() => expect(readTerminal).toHaveBeenCalledTimes(2)) + const live = registry.peekCleanup(SUBSCRIPTION_ID) + + releaseFirstRead() + await flush() + + expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBe(live) + }) +}) + +describe('terminal.unsubscribe connection ownership', () => { + const unsubscribeRequest = (subscriptionId: string): RpcRequest => ({ + id: 'req-unsub', + authToken: 'tok', + method: 'terminal.unsubscribe', + params: { subscriptionId, client: { id: 'phone-1' } } + }) + + // T5 stale-connection: the make-before-break migration case. + it('ignores an unsubscribe from a connection that no longer owns the subscription', async () => { + const registry = createSubscriptionRegistryDouble() + const waiters: Waiter[] = [] + const runtime = stubRuntime(registry, waiters) + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + + void dispatcher.dispatchStreaming(makeRequest(binaryParams), vi.fn(), streamOptions('conn-a')) + await vi.waitFor(() => expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBeDefined()) + void dispatcher.dispatchStreaming(makeRequest(binaryParams), vi.fn(), streamOptions('conn-b')) + await flush() + const live = registry.peekCleanup(SUBSCRIPTION_ID) + expect(live).toBeDefined() + + const replies: string[] = [] + // Why dispatchStreaming: websocket requests route through it even for + // non-streaming methods, and it is the only path that carries connectionId. + await dispatcher.dispatchStreaming( + unsubscribeRequest(SUBSCRIPTION_ID), + (msg) => replies.push(msg), + { connectionId: 'conn-a' } + ) + await flush() + + expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBe(live) + expect(JSON.parse(replies[0]!).result).toEqual({ unsubscribed: false }) + }) + + // T5 same-connection: preservation — the owning connection may still unsubscribe. + it('honors an unsubscribe from the connection that owns the subscription', async () => { + const registry = createSubscriptionRegistryDouble() + const waiters: Waiter[] = [] + const runtime = stubRuntime(registry, waiters) + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + + void dispatcher.dispatchStreaming(makeRequest(binaryParams), vi.fn(), streamOptions('conn-a')) + await vi.waitFor(() => expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBeDefined()) + + const replies: string[] = [] + // Why dispatchStreaming: websocket requests route through it even for + // non-streaming methods, and it is the only path that carries connectionId. + await dispatcher.dispatchStreaming( + unsubscribeRequest(SUBSCRIPTION_ID), + (msg) => replies.push(msg), + { connectionId: 'conn-a' } + ) + await flush() + + expect(registry.peekCleanup(SUBSCRIPTION_ID)).toBeUndefined() + expect(JSON.parse(replies[0]!).result).toEqual({ unsubscribed: true }) + }) +}) diff --git a/src/main/runtime/rpc/terminal-subscribe-reconnect-rebind.test.ts b/src/main/runtime/rpc/terminal-subscribe-reconnect-rebind.test.ts new file mode 100644 index 00000000000..1d0e3e7456d --- /dev/null +++ b/src/main/runtime/rpc/terminal-subscribe-reconnect-rebind.test.ts @@ -0,0 +1,121 @@ +import { describe, expect, it, vi } from 'vitest' +import type { RuntimeTerminalWait } from '../../../shared/runtime-types' +import type { OrcaRuntimeService } from '../orca-runtime' +import type { RpcRequest } from './core' +import { RpcDispatcher } from './dispatcher' +import { TERMINAL_METHODS } from './methods/terminal' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame +} from '../../../shared/terminal-stream-protocol' +import { createSubscriptionRegistryDouble } from './subscription-registry-test-double' + +const makeRequest = (params: unknown): RpcRequest => ({ + id: 'req-1', + authToken: 'tok', + method: 'terminal.subscribe', + params +}) + +const subscribeParams = { + terminal: 'terminal-1', + client: { id: 'phone-1', type: 'mobile' }, + capabilities: { terminalBinaryStream: 1 } +} + +describe('terminal.subscribe reconnect rebind (STA-4510)', () => { + it('keeps the rebound live stream when the pre-reconnect connection aborts', async () => { + const registry = createSubscriptionRegistryDouble() + const dataListeners: ((data: string, meta?: { seq?: number; rawLength?: number }) => void)[] = + [] + + const runtime = { + getRuntimeId: () => 'test-runtime', + registerRemoteTerminalViewSubscriber: () => () => {}, + requestRendererTerminalTabMount: () => false, + resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }), + handleMobileSubscribe: vi.fn().mockResolvedValue(true), + handleMobileUnsubscribe: vi.fn(), + subscribeToTerminalData: vi.fn((_ptyId: string, listener: (typeof dataListeners)[number]) => { + dataListeners.push(listener) + return vi.fn() + }), + readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }), + serializeTerminalBuffer: vi + .fn() + .mockResolvedValue({ data: 'snapshot', cols: 80, rows: 24, seq: 4 }), + getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }), + getMobileDisplayMode: vi.fn().mockReturnValue('auto'), + getLayout: vi.fn().mockReturnValue({ seq: 1 }), + isTerminalAlternateScreen: vi.fn().mockReturnValue(false), + subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()), + subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()), + registerSubscriptionCleanup: vi.fn(registry.registerSubscriptionCleanup), + cleanupSubscription: vi.fn(registry.cleanupSubscription), + registerOwnedSubscriptionCleanup: vi.fn(registry.registerOwnedSubscriptionCleanup), + // Mirrors bindTerminalWaiterAbort (orca-runtime.ts:34183): abort rejects. + waitForTerminal: vi.fn( + (_handle: string, options?: { signal?: AbortSignal }) => + new Promise((_resolve, reject) => { + options?.signal?.addEventListener('abort', () => reject(new Error('request_aborted')), { + once: true + }) + }) + ) + } as unknown as OrcaRuntimeService + const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS }) + + // --- connection A: the original mobile socket ------------------------- + const connA = new AbortController() + void dispatcher.dispatchStreaming(makeRequest(subscribeParams), vi.fn(), { + signal: connA.signal, + connectionId: 'conn-a', + sendBinary: vi.fn(), + registerBinaryStreamHandler: vi.fn(() => vi.fn()) + }) + await vi.waitFor(() => expect(dataListeners).toHaveLength(1)) + + // --- connection B: the reconnect re-sends subscribe, no unsubscribe ---- + const connB = new AbortController() + const framesB: Uint8Array[] = [] + void dispatcher.dispatchStreaming(makeRequest(subscribeParams), vi.fn(), { + signal: connB.signal, + connectionId: 'conn-b', + sendBinary: (bytes) => { + framesB.push(bytes) + }, + registerBinaryStreamHandler: vi.fn(() => vi.fn()) + }) + await vi.waitFor(() => expect(dataListeners).toHaveLength(2)) + const liveListener = dataListeners[1] + + // Control: the rebound stream delivers output before the old socket is reaped. + await new Promise((resolve) => setTimeout(resolve, 50)) + framesB.length = 0 + liveListener('before\r\n', { seq: 1, rawLength: 8 }) + await new Promise((resolve) => setTimeout(resolve, 50)) + expect( + framesB.some((b) => decodeTerminalStreamFrame(b)?.opcode === TerminalStreamOpcode.Output) + ).toBe(true) + + // A's own cleanup already ran at rebind time; nothing more may fire after. + const unsubscribesAfterRebind = vi.mocked(runtime.handleMobileUnsubscribe).mock.calls.length + + // --- the half-open socket A is finally detected and closed ------------ + // runtime-rpc.ts:1339-1341 aborts dispatches first, then sweeps the conn. + connA.abort() + registry.cleanupSubscriptionsForConnection('conn-a') + await new Promise((resolve) => setTimeout(resolve, 0)) + + // --- the live (rebound) stream must survive --------------------------- + framesB.length = 0 + liveListener('after\r\n', { seq: 2, rawLength: 7 }) + await new Promise((resolve) => setTimeout(resolve, 20)) + expect( + framesB.some((b) => decodeTerminalStreamFrame(b)?.opcode === TerminalStreamOpcode.Output) + ).toBe(true) + expect(vi.mocked(runtime.handleMobileUnsubscribe).mock.calls.length).toBe( + unsubscribesAfterRebind + ) + }) +}) diff --git a/src/main/runtime/rpc/terminal-subscribe-renderer-recovery-output.test.ts b/src/main/runtime/rpc/terminal-subscribe-renderer-recovery-output.test.ts index 91cd571e0f4..550a95b5c36 100644 --- a/src/main/runtime/rpc/terminal-subscribe-renderer-recovery-output.test.ts +++ b/src/main/runtime/rpc/terminal-subscribe-renderer-recovery-output.test.ts @@ -9,6 +9,7 @@ import type { OrcaRuntimeService } from '../orca-runtime' import type { RpcRequest } from './core' import { RpcDispatcher } from './dispatcher' import { TERMINAL_METHODS } from './methods/terminal' +import { createSubscriptionRegistryDouble } from './subscription-registry-test-double' const request: RpcRequest = { id: 'req-output-race', @@ -24,7 +25,7 @@ const request: RpcRequest = { describe('terminal subscribe renderer recovery output ordering', () => { it('replays bytes absent from the renderer snapshot using its exact sequence boundary', async () => { const binaryFrames: Uint8Array[] = [] - const cleanups = new Map void>() + const registry = createSubscriptionRegistryDouble() let outputSequence = 0 let rendererSerializeCalls = 0 let onData: @@ -67,13 +68,9 @@ describe('terminal subscribe renderer recovery output ordering', () => { 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), waitForTerminal: vi.fn(() => new Promise(() => {})) } as unknown as OrcaRuntimeService const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })