From 8cd338357ee35e7348cb08d3a0522fee161b2e28 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 17 Aug 2026 16:13:45 -0700 Subject: [PATCH] fix(runtime): tear down terminal subscriptions only through their owning registration (#14992) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(runtime): tear down terminal subscriptions only through their owning registration A terminal.subscribe teardown was keyed on `${terminal}:${clientId}`, which is stable across reconnects. cleanupSubscription invokes whichever cleanup currently owns that key, so after a mobile reconnect rebound the id, a late teardown from the dead connection killed the replacement stream and the terminal froze (STA-4510). Add registerOwnedSubscriptionCleanup, returning a registration handle whose releaseIfCurrent no-ops once the id has been rebound, and route all 12 teardown call sites in the three terminal.subscribe branches through it. Register-time eviction now also targets the owner it captured rather than re-resolving the key. terminal.unsubscribe gains the same ownership rule via connectionId, matching the runtime.clientEvents.unsubscribe precedent: make-before-break migration sends the unsubscribe over the old session after the new one has already rebound the id. The teardown-by-key pattern predates the bug; #7490 made it reachable by binding the exit-waiter to the per-socket abort signal, so every socket close now runs it. Tests use a faithful subscription-registry double; the ad-hoc Map stubs they replace never evicted the prior generation, which is why no test caught this. * test(runtime): migrate the remaining subscription stubs to the faithful registry streaming.test.ts and terminal-provider-snapshot-sequence.test.ts still stubbed registerSubscriptionCleanup only, so terminal.subscribe's owned registration was undefined and teardown never fired. Assert the registration is retired rather than spying on cleanupSubscription, which the owned path now calls internally. * fix(runtime): guard the lease-only presence release and use the shared registry double Review findings on this PR: - The lease-only branch's compensating handleMobileUnsubscribe ran unconditionally after a rebind. Presence is keyed (ptyId, clientId) with no refcount and `closed` is exactly the post-rebind state, so a superseded handler deleted the replacement subscriber's presence. Gate it on the registration still being current. This also gives SubscriptionRegistration.isCurrent its production caller. - The STA-4510 regression test hand-rolled a second registry copy that diverged from the shared double (no try/catch around cleanup, no in-flight join). Import the shared double instead, so the test proving the bug uses the same fidelity as the rest of the suite. - cleanupSubscriptionIfOwnedByConnection treats an absent connectionId as authority. That is the connection-less local unix-socket path, not an oversight; say so. * fix(runtime): drop the lease-only presence guard; it disabled a real compensation Review pass 2 showed the guard added in the previous commit was wrong. registration.isCurrent() is always false whenever `closed` is true: either our own cleanup ran, in which case cleanupSubscriptionAndWait already deleted the map entry, or a rebind replaced it. So the guard did not distinguish the two cases — it made the compensating handleMobileUnsubscribe unreachable, which is precisely the resurrect-after-cleanup case that line exists to handle. The scenario the guard was meant to fix is also not reachable: the lease-only call passes no viewport, and both !viewport paths in handleMobileSubscribeInternal return with no await, so the subscribe resolves in microtasks and a socket close cannot win it. Revert the guard, and drop SubscriptionRegistration.isCurrent with it — it had no remaining production caller and shipping unused runtime API invites exactly this. Also from review: - cleanupSubscriptionIfOwnedByConnection reported false for an id with no registration, conflating 'refused, another connection owns it' with 'already gone'. Report true. - Note on the registry double that it mirrors production and can drift; the runtime tests pin real behavior, the doubles only pin routing. * fix(runtime): make the unsubscribe refusal observable and restore test-double parity Review pass 3: - The registry test double omitted production's 'unregistered id is already gone' early-out, so it returned refused where production returns gone. The legacy terminal.unsubscribe path reaches that branch with a never-registered bare id, so a routing test would have locked in the inverse of production. - terminal.unsubscribe ORed the bare-id and composite results. Registrations always use the composite, so the bare id reported 'already gone' and masked a real ownership refusal: a stale connection was correctly refused but told unsubscribed: true. The composite answer is authoritative when we try it. - Pin why the lease-only compensating handleMobileUnsubscribe is deliberately unguarded: it is safe only because that call passes no viewport and returns with no await. - Cover the no-registration branch, which had no test. * fix(runtime): report an unsubscribe refusal without masking a real teardown Review pass 4 disproved the previous commit's rationale. A clientless legacy-JSON stream registers under the bare terminal id, not the composite, so either call in terminal.unsubscribe can be the real teardown. Overwriting reported false after a destructive success; the earlier OR reported true after a refusal. AND over the calls that actually ran is the honest aggregate: false needs a genuine refusal. Tests: - cover the bare-id path, which the ownership tests never exercised - pin the microtask invariant the unguarded lease-only compensation depends on. The first version of that test was vacuous: it raced two setTimeout(0) timers, and the earlier-registered one always won, so it passed with an await injected. It now drains microtasks only and goes red under that mutation. --- src/main/runtime/orca-runtime.test.ts | 108 ++++++++ src/main/runtime/orca-runtime.ts | 66 ++++- src/main/runtime/rpc/methods/terminal.ts | 57 ++-- src/main/runtime/rpc/streaming.test.ts | 21 +- .../rpc/subscription-registry-test-double.ts | 168 ++++++++++++ ...rminal-multiplex-ack-output-budget.test.ts | 49 ++-- ...al-multiplex-ack-overflow-recovery.test.ts | 12 +- ...ltiplex-initial-snapshot-buffering.test.ts | 42 +-- ...al-multiplex-input-write-rejection.test.ts | 54 ++-- ...ultiplex-output-pause-and-viewport.test.ts | 29 +- ...rminal-multiplex-pty-wait-capacity.test.ts | 14 +- ...l-multiplex-snapshot-serialization.test.ts | 2 +- ...l-multiplex-source-range-admission.test.ts | 8 +- .../rpc/terminal-multiplex-test-harness.ts | 16 +- .../rpc/terminal-output-batching.test.ts | 52 ++-- ...erminal-provider-snapshot-sequence.test.ts | 28 +- .../terminal-subscribe-blank-mount.test.ts | 37 +-- .../rpc/terminal-subscribe-buffer.test.ts | 101 +++---- .../rpc/terminal-subscribe-lease-only.test.ts | 13 +- .../terminal-subscribe-mount-replay.test.ts | 61 ++--- .../rpc/terminal-subscribe-ownership.test.ts | 256 ++++++++++++++++++ ...erminal-subscribe-reconnect-rebind.test.ts | 121 +++++++++ ...subscribe-renderer-recovery-output.test.ts | 13 +- 23 files changed, 965 insertions(+), 363 deletions(-) create mode 100644 src/main/runtime/rpc/subscription-registry-test-double.ts create mode 100644 src/main/runtime/rpc/terminal-subscribe-ownership.test.ts create mode 100644 src/main/runtime/rpc/terminal-subscribe-reconnect-rebind.test.ts 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 })