From bdd7a5431b6ee58a37771cfa458dc6ec2948e5da Mon Sep 17 00:00:00 2001 From: eitsupi Date: Mon, 12 Oct 2026 03:17:40 +0000 Subject: [PATCH 1/6] feat: delegate manual attachment to external terminals --- CHANGELOG.md | 1 + src/api.d.ts | 28 + src/extension.ts | 1 + src/session.ts | 155 +++++- src/test/suite/session.test.ts | 4 + src/test/suite/sessionTerminalBinding.test.ts | 1 + .../suite/sessionTerminalIntegration.test.ts | 499 ++++++++++++++++++ .../suite/sessionTerminalLifecycle.test.ts | 3 + src/test/suite/workspaceViewer.test.ts | 1 + 9 files changed, 687 insertions(+), 6 deletions(-) create mode 100644 src/test/suite/sessionTerminalIntegration.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index c35e2d313..c74617bea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ - Added the extension API's `session.prepareAttachment()` for extensions that own an R process and need vscode-R's managed `sess` setup. R errors and interrupts during manual attachment are now reported as messages while leaving the R prompt usable. - Added connection change events and live connection references for sess attachments, with activation controls for consumer-owned terminals. +- Added `session.registerTerminalIntegration()` so external terminals can recover sess connections without falling back to a native attach command. - The httpgd plot backend now requires the `httpgd` R package version 2.0.1 or later. Older versions are treated as unavailable when choosing the plot backend. (#1855) ## 3.0.1 - 2026-09-28 diff --git a/src/api.d.ts b/src/api.d.ts index 3590e9af0..3dae0890e 100644 --- a/src/api.d.ts +++ b/src/api.d.ts @@ -40,6 +40,16 @@ export interface RSessionActivationOptions { select?: boolean; } +/** Consumer-owned terminal hook for restoring a sess connection without bypassing its proxy. */ +export interface RExternalTerminalIntegration { + /** + * Reconnect through the consumer's transport and bind this terminal before resolving true. + * A resolved false is a consumer-reported decline/busy result; a rejection is shown by + * vscode-R. Cancellation can stop pending work, but cannot retract R code already submitted. + */ + attach(token: vscode.CancellationToken): Promise; +} + export interface RSessionPreparationOptions { /** * A local sess IPC endpoint on the current R host, such as a proxy socket. @@ -123,6 +133,24 @@ export interface RSessionApi { * is still current after activation completes; a later disconnect can still occur. */ activate(sessionId: string, options?: RSessionActivationOptions): Promise; + /** + * Register one attach hook for an open terminal. Duplicate or closed-terminal registrations + * throw. The hook remains registered across session disconnects and replacements until its + * Disposable is disposed, the terminal closes, or the extension host shuts down. It is + * invoked before native PID discovery; false, rejection, busy, or cancellation never falls + * back to a native attach command. On true, vscode-R requires a current live session binding + * for this terminal, then focuses that session. Consumers should bind with + * `activate(..., { terminal, connectionId, select: false })` before resolving true. If the + * terminal already has a current live owner, that owner is focused and the hook is skipped. + * While a hook is in flight, repeated activations of that terminal do not invoke it again. + * Registered terminals are excluded from native PID association, so consumers must bind + * their session explicitly. Registration during host shutdown throws. + * If the user selects another session while the hook is pending, vscode-R leaves that selection in + * place and retains the terminal binding. Cancellation is requested when focus leaves the + * terminal, it closes, registration is disposed, or the host shuts down; submitted R code + * cannot be recalled. + */ + registerTerminalIntegration(terminal: vscode.Terminal, integration: RExternalTerminalIntegration): vscode.Disposable; } export type HelpSubMenu = 'doc' | 'pkgList' | 'refresh' | '?' | '??'; diff --git a/src/extension.ts b/src/extension.ts index 48ad774d1..ce861eac7 100644 --- a/src/extension.ts +++ b/src/extension.ts @@ -66,6 +66,7 @@ export async function activate(context: vscode.ExtensionContext): Promise( target => sessions.get(target.sessionId) === target && isConnectedSession(target), terminal => !terminal.exitStatus && window.terminals.includes(terminal), ); +interface TerminalIntegrationRun { + readonly cancellation: vscode.CancellationTokenSource; + cancel(): void; +} +interface TerminalIntegrationRegistration { + readonly terminal: vscode.Terminal; + readonly integration: RExternalTerminalIntegration; + readonly closeSubscription: vscode.Disposable; + active?: TerminalIntegrationRun; +} +const terminalIntegrations = new Map(); const documentSessions = new Map(); const sessionDocumentBound = new vscode.EventEmitter(); export const onDidBindSessionDocument = sessionDocumentBound.event; @@ -714,6 +726,68 @@ function isLiveTerminal(terminal: vscode.Terminal): boolean { /** Called for every terminal close, including extension-owned pseudoterminals. */ export function cleanupTerminalBinding(terminal: vscode.Terminal): void { terminalRegistry.closeTerminal(terminal); + removeTerminalIntegration(terminal); +} + +function removeTerminalIntegration(terminal: vscode.Terminal): void { + removeTerminalIntegrationRegistration(terminalIntegrations.get(terminal)); +} + +function removeTerminalIntegrationRegistration(registration: TerminalIntegrationRegistration | undefined): void { + if (!registration) { return; } + if (terminalIntegrations.get(registration.terminal) !== registration) { return; } + // Remove ownership before canceling: cancellation listeners may synchronously + // attempt a new registration or start another attach operation. + terminalIntegrations.delete(registration.terminal); + registration.closeSubscription.dispose(); + if (window.activeTerminal === registration.terminal) { sessionSelectionRevision++; } + registration.active?.cancel(); +} + +function revokeTerminalIntegrationSnapshot(registrations: readonly TerminalIntegrationRegistration[]): void { + for (const registration of registrations) { + if (terminalIntegrations.get(registration.terminal) === registration) { + terminalIntegrations.delete(registration.terminal); + } + } + for (const registration of registrations) { + registration.closeSubscription.dispose(); + } + if (registrations.some(registration => window.activeTerminal === registration.terminal)) { + sessionSelectionRevision++; + } + for (const registration of registrations) { + registration.active?.cancel(); + } +} + +export function registerTerminalIntegration( + terminal: vscode.Terminal, + integration: RExternalTerminalIntegration, +): vscode.Disposable { + if (attachmentLifecycleStopping) { + throw new Error('Cannot register a terminal integration while the session watcher is shutting down.'); + } + if (!terminal || !terminalRegistry.isLive(terminal)) { + throw new Error('Cannot register an integration for a closed or unavailable terminal.'); + } + if (terminalIntegrations.has(terminal)) { + throw new Error('A terminal integration is already registered for this terminal.'); + } + if (!integration || typeof integration.attach !== 'function') { + throw new TypeError('A terminal integration must provide an attach(token) function.'); + } + const registration: TerminalIntegrationRegistration = { + terminal, + integration, + closeSubscription: window.onDidCloseTerminal(changed => { + if (changed === terminal) { cleanupTerminalBinding(terminal); } + }), + }; + terminalIntegrations.set(terminal, registration); + // Invalidate an earlier native PID/command path that may still be awaiting I/O. + if (window.activeTerminal === terminal) { sessionSelectionRevision++; } + return { dispose: () => removeTerminalIntegrationRegistration(registration) }; } /** @@ -1362,6 +1436,7 @@ async function removePathIfExists(pathLike: string): Promise { export async function shutdownSessionWatcher(): Promise { attachmentLifecycleGeneration++; attachmentLifecycleStopping = true; + const terminalIntegrationSnapshot = [...terminalIntegrations.values()]; const shutdownGeneration = attachmentLifecycleGeneration; const consentStartup = attachConsentStartup; const consentService = attachConsentService; @@ -1396,6 +1471,8 @@ export async function shutdownSessionWatcher(): Promise { }); attachScriptPublication = manualCleanup; // Snapshot and revoke before notifying; listeners may redeploy synchronously. + sessionSelectionRevision++; + revokeTerminalIntegrationSnapshot(terminalIntegrationSnapshot); enqueueSessionConnectionChanges(shutdownChanges); await Promise.allSettled(pendingOperations); await manualCleanup.catch(() => undefined); @@ -1455,9 +1532,6 @@ export async function activateRSession(): Promise { return; } const terminal = window.activeTerminal; - const selectionRevision = ++sessionSelectionRevision; - const isCurrent = () => selectionRevision === sessionSelectionRevision && terminal === window.activeTerminal - && (!terminal || (!isTerminalClosed(terminal) && !terminal.exitStatus)); const focusOwner = (): Promise | undefined => { const owner = terminalRegistry.ownerOf(terminal); if (!terminal || !owner) { return undefined; } @@ -1472,6 +1546,70 @@ export async function activateRSession(): Promise { const focused = focusOwner(); if (focused) { await focused; return; } + if (terminal) { + const registration = terminalIntegrations.get(terminal); + if (registration) { + if (registration.active) { + void window.showInformationMessage('An R session attach is already in progress for this terminal.'); + return; + } + if (!isLiveTerminal(terminal) || window.activeTerminal !== terminal) { return; } + // Supersede native attach operations that may still be waiting for PID or command I/O. + sessionSelectionRevision++; + const integrationSelectionRevision = sessionSelectionRevision; + const cancellation = new vscode.CancellationTokenSource(); + let focusSubscription: vscode.Disposable = { dispose: () => undefined }; + const run: TerminalIntegrationRun = { + cancellation, + cancel: () => { + focusSubscription.dispose(); + cancellation.cancel(); + }, + }; + registration.active = run; + focusSubscription = window.onDidChangeActiveTerminal(changed => { + if (changed !== terminal) { run.cancel(); } + }); + const isCurrentIntegration = () => terminalIntegrations.get(terminal) === registration + && registration.active === run + && !cancellation.token.isCancellationRequested + && isLiveTerminal(terminal) + && window.activeTerminal === terminal; + try { + if (!isCurrentIntegration()) { return; } + const attached = await registration.integration.attach(cancellation.token); + if (!isCurrentIntegration()) { return; } + if (attached !== true) { return; } + const owner = terminalRegistry.ownerOf(terminal); + if (!owner) { + void window.showErrorMessage('The external R terminal reported a successful attach but has no current session binding.'); + return; + } + if (sessionSelectionRevision !== integrationSelectionRevision && activeSession !== owner) { return; } + const activation = activeSession === owner ? Promise.resolve() : activateSession(owner); + const activationRevision = sessionSelectionRevision; + await activation; + if (sessionSelectionRevision !== activationRevision) { return; } + if (!isCurrentIntegration() || terminalRegistry.ownerOf(terminal) !== owner) { return; } + terminal.show(); + } catch (error) { + if (isCurrentIntegration()) { + const detail = error instanceof Error ? error.message : String(error); + void window.showErrorMessage(`Failed to attach the R session in this external terminal: ${detail}`); + } + } finally { + focusSubscription.dispose(); + cancellation.dispose(); + if (registration.active === run) { registration.active = undefined; } + } + return; + } + } + + const selectionRevision = ++sessionSelectionRevision; + const isCurrent = () => selectionRevision === sessionSelectionRevision && terminal === window.activeTerminal + && (!terminal || (!isTerminalClosed(terminal) && !terminal.exitStatus)); + let discoveryPath = terminal && terminalDiscoveryPath(terminal); if (terminal && !discoveryPath) { const terminalPid = await terminal.processId; @@ -1486,7 +1624,7 @@ export async function activateRSession(): Promise { if (!isCurrent()) { return; } const focused = focusOwner(); if (focused) { await focused; return; } - if (terminal && isLiveTerminal(terminal)) { + if (terminal && isLiveTerminal(terminal) && !terminalIntegrations.has(terminal)) { terminal.sendText(command, true); terminal.show(); } @@ -2635,10 +2773,15 @@ interface NativeTerminalCandidate { async function findLocalTerminal(rPid: string): Promise { const candidates = new Map(); for (const terminal of window.terminals) { + if (terminalIntegrations.has(terminal)) { continue; } const isCurrent = terminalRegistry.snapshot(terminal); const terminalPid = await terminal.processId; - if (terminalPid === undefined || !isCurrent()) { continue; } - const candidate = { terminal, pid: String(terminalPid), isCurrent }; + if (terminalPid === undefined || !isCurrent() || terminalIntegrations.has(terminal)) { continue; } + const candidate = { + terminal, + pid: String(terminalPid), + isCurrent: () => isCurrent() && !terminalIntegrations.has(terminal), + }; if (candidate.pid === rPid) { return candidate; } candidates.set(terminalPid, candidate); } diff --git a/src/test/suite/session.test.ts b/src/test/suite/session.test.ts index acd058853..5648dc743 100644 --- a/src/test/suite/session.test.ts +++ b/src/test/suite/session.test.ts @@ -176,6 +176,7 @@ suite('Session Communication', () => { prepareAttachment: session.prepareAttachment, onDidChangeSessionConnection: session.onDidChangeSessionConnection, getSessionReference: session.getSessionReference, + registerTerminalIntegration: session.registerTerminalIntegration, activate: session.activateSessionById, }; @@ -311,6 +312,7 @@ suite('Session Communication', () => { prepareAttachment: session.prepareAttachment, onDidChangeSessionConnection: session.onDidChangeSessionConnection, getSessionReference: session.getSessionReference, + registerTerminalIntegration: session.registerTerminalIntegration, activate: session.activateSessionById, }; assert.strictEqual(await api.getConnectionInfo(), undefined); @@ -343,6 +345,7 @@ suite('Session Communication', () => { prepareAttachment: session.prepareAttachment, onDidChangeSessionConnection: session.onDidChangeSessionConnection, getSessionReference: session.getSessionReference, + registerTerminalIntegration: session.registerTerminalIntegration, activate: session.activateSessionById, }; const endpoint = await waitFor(() => session.globalPipePath); @@ -722,6 +725,7 @@ suite('Session Communication', () => { prepareAttachment: session.prepareAttachment, onDidChangeSessionConnection: session.onDidChangeSessionConnection, getSessionReference: session.getSessionReference, + registerTerminalIntegration: session.registerTerminalIntegration, activate: session.activateSessionById, }; const endpoint = process.platform === 'win32' diff --git a/src/test/suite/sessionTerminalBinding.test.ts b/src/test/suite/sessionTerminalBinding.test.ts index ea35f4e34..57c0472be 100644 --- a/src/test/suite/sessionTerminalBinding.test.ts +++ b/src/test/suite/sessionTerminalBinding.test.ts @@ -27,6 +27,7 @@ suite('Session Terminal Binding', () => { const extensionRoot = path.join(__dirname, '..', '..', '..'); const api: RSessionApi = { getConnectionInfo: session.getConnectionInfo, prepareAttachment: session.prepareAttachment, onDidChangeSessionConnection: session.onDidChangeSessionConnection, getSessionReference: session.getSessionReference, + registerTerminalIntegration: session.registerTerminalIntegration, activate: session.activateSessionById }; setup(async () => { diff --git a/src/test/suite/sessionTerminalIntegration.test.ts b/src/test/suite/sessionTerminalIntegration.test.ts new file mode 100644 index 000000000..00e31f890 --- /dev/null +++ b/src/test/suite/sessionTerminalIntegration.test.ts @@ -0,0 +1,499 @@ +import * as assert from 'assert'; +import * as os from 'os'; +import * as path from 'path'; +import * as sinon from 'sinon'; +import * as vscode from 'vscode'; +import * as extension from '../../extension'; +import * as rTerminal from '../../rTerminal'; +import * as session from '../../session'; +import * as util from '../../util'; +import { mockExtensionContext } from '../common/mockvscode'; +import { deferred, SessionConnections, waitForValue } from '../common/sessionConnections'; + +suite('External Terminal Integration', () => { + let sandbox: sinon.SinonSandbox; + let connections: SessionConnections; + let terminals: sinon.SinonStub; + let activeTerminal: sinon.SinonStub; + let showError: sinon.SinonStub; + let showInformation: sinon.SinonStub; + let closeEvents: vscode.EventEmitter; + let activeEvents: vscode.EventEmitter; + let activeSubscriptionDisposals: sinon.SinonSpy[]; + const terminalSpies = new WeakMap(); + const extensionRoot = path.join(__dirname, '..', '..', '..'); + + function makeTerminal(processId: Promise = Promise.resolve(undefined)) { + const sendText = sandbox.stub(); + const show = sandbox.stub(); + const terminal = { + name: 'External R Console', processId, + sendText, show, dispose: sandbox.stub(), + } as unknown as vscode.Terminal; + terminalSpies.set(terminal, { sendText, show }); + return terminal; + } + + function setTerminals(values: vscode.Terminal[], active = values[0]) { + terminals.value(values); + activeTerminal.value(active); + } + + setup(async () => { + await session.shutdownSessionWatcher(); + sandbox = sinon.createSandbox(); + connections = new SessionConnections(); + mockExtensionContext(extensionRoot, sandbox); + sandbox.stub(extension, 'enableSessionWatcher').value(true); + sandbox.stub(extension, 'rWorkspace').value(undefined); + sandbox.stub(vscode.commands, 'executeCommand').resolves(); + terminals = sandbox.stub(vscode.window, 'terminals').value([]); + activeTerminal = sandbox.stub(vscode.window, 'activeTerminal').value(undefined); + closeEvents = new vscode.EventEmitter(); + activeEvents = new vscode.EventEmitter(); + activeSubscriptionDisposals = []; + sandbox.stub(vscode.window, 'onDidCloseTerminal').callsFake(closeEvents.event); + sandbox.stub(vscode.window, 'onDidChangeActiveTerminal').callsFake(listener => { + const subscription = activeEvents.event(listener); + const disposed = sandbox.spy(); + activeSubscriptionDisposals.push(disposed); + return { dispose: () => { disposed(); subscription.dispose(); } }; + }); + showError = sandbox.stub(vscode.window, 'showErrorMessage').resolves(undefined); + showInformation = sandbox.stub(vscode.window, 'showInformationMessage').resolves(undefined); + sandbox.stub(util, 'config').returns({ + get: (key: string) => ({ sessionWatcher: true, alwaysUseActiveTerminal: false, bracketedPaste: true })[key], + } as unknown as vscode.WorkspaceConfiguration); + session.deploySessionWatcher(extensionRoot); + await session.getGlobalPipePath(); + }); + + teardown(async () => { + await connections.dispose(); + await session.shutdownSessionWatcher(); + closeEvents.dispose(); + activeEvents.dispose(); + sandbox.restore(); + }); + + test('unresolved PID terminal uses the hook, then binds a real sess connection before success', async () => { + const terminal = makeTerminal(new Promise(() => undefined)); + setTerminals([terminal]); + let hookCalls = 0; + const registration = session.registerTerminalIntegration(terminal, { + attach: async token => { + hookCalls++; + assert.strictEqual(token.isCancellationRequested, false); + const socket = await connections.attach('integration-real-connection', 55101, 'external-host', { + attachmentId: 'integration-test', autoActivate: false, + }); + const reference = await waitForValue(() => session.getSessionReference('integration-real-connection')); + return !socket.destroyed && await session.activateSessionById(reference.sessionId, { + connectionId: reference.connectionId, terminal, select: false, + }); + }, + }); + try { + await session.activateRSession(); + assert.strictEqual(hookCalls, 1); + assert.strictEqual(session.activeSession?.sessionId, 'integration-real-connection'); + assert.strictEqual(session.getSessionReference('integration-real-connection')?.attachmentId, 'integration-test'); + await session.executeSessionCode(session.activeSession!, '1 + 1'); + sinon.assert.calledOnceWithExactly(terminalSpies.get(terminal)!.sendText, '1 + 1'); + sinon.assert.calledOnce(terminalSpies.get(terminal)!.show); + } finally { + registration.dispose(); + await session.cleanupSession('integration-real-connection'); + } + }); + + test('live owner is focused before the integration hook', async () => { + const terminal = makeTerminal(); + setTerminals([terminal]); + const owner = session.registerSessionTransport('integration-existing-owner', 'host', '/tmp', () => Promise.resolve({})); + await session.activateSessionById(owner.sessionId, { terminal, select: false }); + const attach = sandbox.stub().resolves(false); + const registration = session.registerTerminalIntegration(terminal, { attach }); + try { + await session.activateRSession(); + sinon.assert.notCalled(attach); + sinon.assert.calledOnce(terminalSpies.get(terminal)!.show); + assert.strictEqual(session.activeSession, owner); + } finally { + registration.dispose(); + session.unregisterSessionTransport(owner); + } + }); + + test('a registered unresolved PID terminal is excluded from native discovery scans', async () => { + const external = makeTerminal(new Promise(() => undefined)); + const native = makeTerminal(Promise.resolve(55108)); + setTerminals([external, native], native); + const registration = session.registerTerminalIntegration(external, { attach: () => Promise.resolve(false) }); + try { + const socket = await connections.attach('native-scan-with-integration', 55108, os.hostname()); + assert.strictEqual(socket._terminalPid, 55108); + assert.strictEqual(session.activeSession?.sessionId, 'native-scan-with-integration'); + } finally { + registration.dispose(); + await session.cleanupSession('native-scan-with-integration'); + } + }); + + test('registering the active terminal during native PID discovery cancels the native path', async () => { + let resolvePid!: (pid: number | undefined) => void; + let pidRead!: () => void; + const readStarted = new Promise(resolve => { pidRead = resolve; }); + const sendText = sandbox.stub(); + const show = sandbox.stub(); + const terminal = { + name: 'Native R', + get processId() { pidRead(); return new Promise(resolve => { resolvePid = resolve; }); }, + sendText, show, + } as unknown as vscode.Terminal; + terminalSpies.set(terminal, { sendText, show }); + setTerminals([terminal]); + const createRTerm = sandbox.stub(rTerminal, 'createRTerm').resolves(true); + const nativeOperation = session.activateRSession(); + await readStarted; + const registration = session.registerTerminalIntegration(terminal, { attach: () => Promise.resolve(false) }); + try { + await session.activateRSession(); + resolvePid(undefined); + await nativeOperation; + sinon.assert.notCalled(terminalSpies.get(terminal)!.sendText); + sinon.assert.notCalled(createRTerm); + } finally { + resolvePid(undefined); + registration.dispose(); + } + }); + + test('registering a background terminal does not invalidate an unrelated native operation', async () => { + let resolvePid!: (pid: number | undefined) => void; + let pidRead!: () => void; + const readStarted = new Promise(resolve => { pidRead = resolve; }); + const sendText = sandbox.stub(); + const show = sandbox.stub(); + const active = { + name: 'Native R', + get processId() { pidRead(); return new Promise(resolve => { resolvePid = resolve; }); }, + sendText, show, + } as unknown as vscode.Terminal; + terminalSpies.set(active, { sendText, show }); + const background = makeTerminal(); + setTerminals([active, background], active); + const createRTerm = sandbox.stub(rTerminal, 'createRTerm').resolves(true); + const nativeOperation = session.activateRSession(); + await readStarted; + const registration = session.registerTerminalIntegration(background, { attach: () => Promise.resolve(false) }); + try { + resolvePid(undefined); + await nativeOperation; + sinon.assert.calledOnce(createRTerm); + } finally { + resolvePid(undefined); + registration.dispose(); + } + }); + + test('false and thrown hooks never use native fallback; throws get a host diagnostic', async () => { + const terminal = makeTerminal(); + setTerminals([terminal]); + const createRTerm = sandbox.stub(rTerminal, 'createRTerm').resolves(true); + const decline = session.registerTerminalIntegration(terminal, { attach: () => Promise.resolve(false) }); + try { + await session.activateRSession(); + sinon.assert.notCalled(terminalSpies.get(terminal)!.sendText); + sinon.assert.notCalled(createRTerm); + } finally { decline.dispose(); } + + const failure = new Error('proxy setup failed'); + const rejecting = session.registerTerminalIntegration(terminal, { attach: () => Promise.reject(failure) }); + try { + await session.activateRSession(); + sinon.assert.calledOnceWithExactly(showError, + 'Failed to attach the R session in this external terminal: proxy setup failed'); + sinon.assert.notCalled(terminalSpies.get(terminal)!.sendText); + sinon.assert.notCalled(createRTerm); + } finally { rejecting.dispose(); } + }); + + test('true without a current terminal binding is diagnosed and does not fall back', async () => { + const terminal = makeTerminal(); + setTerminals([terminal]); + const createRTerm = sandbox.stub(rTerminal, 'createRTerm').resolves(true); + const registration = session.registerTerminalIntegration(terminal, { attach: () => Promise.resolve(true) }); + try { + await session.activateRSession(); + sinon.assert.calledOnceWithExactly(showError, + 'The external R terminal reported a successful attach but has no current session binding.'); + sinon.assert.notCalled(terminalSpies.get(terminal)!.sendText); + sinon.assert.notCalled(createRTerm); + } finally { registration.dispose(); } + }); + + test('busy clicks do not duplicate or invalidate the in-flight hook', async () => { + const terminal = makeTerminal(); + setTerminals([terminal]); + const owner = session.registerSessionTransport('integration-busy-owner', 'host', '/tmp', () => Promise.resolve({})); + const gate = deferred(); + const started = deferred(); + const attach = sandbox.stub().callsFake(() => { started.resolve(); return gate.promise; }); + const registration = session.registerTerminalIntegration(terminal, { attach }); + try { + const first = session.activateRSession(); + await started.promise; + await session.activateRSession(); + sinon.assert.calledOnce(attach); + sinon.assert.calledOnceWithExactly(showInformation, 'An R session attach is already in progress for this terminal.'); + assert.strictEqual(await session.activateSessionById(owner.sessionId, { terminal, select: false }), true); + gate.resolve(true); + await first; + assert.strictEqual(session.activeSession, owner, 'the successful hook should still select its live bound owner'); + sinon.assert.calledOnce(terminalSpies.get(terminal)!.show); + } finally { + gate.resolve(true); + registration.dispose(); + session.unregisterSessionTransport(owner); + } + }); + + test('focus away and back cancels the hook without reviving it or attaching natively', async () => { + const terminal = makeTerminal(); + const other = makeTerminal(); + setTerminals([terminal, other]); + const gate = deferred(); + const started = deferred(); + const registration = session.registerTerminalIntegration(terminal, { + attach: token => { started.resolve(token); return gate.promise; }, + }); + try { + const operation = session.activateRSession(); + const token = await started.promise; + activeTerminal.value(other); + activeEvents.fire(other); + activeTerminal.value(terminal); + activeEvents.fire(terminal); + assert.strictEqual(token.isCancellationRequested, true); + gate.resolve(true); + await operation; + sinon.assert.notCalled(terminalSpies.get(terminal)!.sendText); + sinon.assert.notCalled(showError); + } finally { + gate.resolve(false); + registration.dispose(); + } + }); + + test('dispose, close, and shutdown cancel active hooks and revoke registration first', async () => { + for (const end of ['dispose', 'close', 'shutdown'] as const) { + const terminal = makeTerminal(); + setTerminals([terminal]); + const gate = deferred(); + const started = deferred(); + const registration = session.registerTerminalIntegration(terminal, { + attach: token => { started.resolve(token); return gate.promise; }, + }); + const operation = session.activateRSession(); + const token = await started.promise; + const focusDispose = activeSubscriptionDisposals.at(-1)!; + if (end === 'dispose') { registration.dispose(); } + else if (end === 'close') { closeEvents.fire(terminal); } + else { await session.shutdownSessionWatcher(); } + assert.strictEqual(token.isCancellationRequested, true, `${end} should cancel the hook`); + sinon.assert.calledOnce(focusDispose); + gate.resolve(true); + await operation; + sinon.assert.notCalled(terminalSpies.get(terminal)!.sendText); + if (end === 'dispose') { + const replacement = session.registerTerminalIntegration(terminal, { attach: () => Promise.resolve(false) }); + replacement.dispose(); + } else { + assert.throws(() => session.registerTerminalIntegration(terminal, { attach: () => Promise.resolve(true) }), /closed|shutting down/); + } + if (end === 'close') { + session.cleanupTerminalBinding(terminal); + // A new lifecycle can register only after a fresh terminal object appears. + const freshTerminal = makeTerminal(); + setTerminals([freshTerminal]); + const fresh = session.registerTerminalIntegration(freshTerminal, { attach: () => Promise.resolve(false) }); + fresh.dispose(); + } + if (end === 'shutdown') { break; } + } + }); + + test('shutdown cancellation can redeploy without the old teardown revoking new registrations', async () => { + const terminal = makeTerminal(); + const freshTerminal = makeTerminal(); + setTerminals([terminal, freshTerminal]); + const gate = deferred(); + const started = deferred(); + let replacementRegistration: vscode.Disposable | undefined; + const replacementAttach = sandbox.stub().resolves(false); + const oldRegistration = session.registerTerminalIntegration(terminal, { + attach: token => { + token.onCancellationRequested(() => { + session.deploySessionWatcher(extensionRoot); + replacementRegistration = session.registerTerminalIntegration(freshTerminal, { attach: replacementAttach }); + }); + started.resolve(); + return gate.promise; + }, + }); + const operation = session.activateRSession(); + await started.promise; + try { + await session.shutdownSessionWatcher(); + assert.ok(replacementRegistration, 'cancellation reentrancy should register against the redeployed lifecycle'); + assert.strictEqual(session.isTerminalClosed(freshTerminal), false); + gate.resolve(false); + await operation; + activeTerminal.value(freshTerminal); + activeEvents.fire(freshTerminal); + await session.activateRSession(); + sinon.assert.calledOnce(replacementAttach); + assert.strictEqual(session.activeSession, undefined, 'the new registration should remain live and may decline attach'); + } finally { + gate.resolve(false); + oldRegistration.dispose(); + replacementRegistration?.dispose(); + await session.shutdownSessionWatcher(); + } + }); + + test('duplicate registrations reject and unregistered terminals retain the native fallback path', async () => { + const terminal = makeTerminal(); + setTerminals([terminal]); + const registration = session.registerTerminalIntegration(terminal, { attach: () => Promise.resolve(false) }); + assert.throws(() => session.registerTerminalIntegration(terminal, { attach: () => Promise.resolve(false) }), /already registered/); + registration.dispose(); + const createRTerm = sandbox.stub(rTerminal, 'createRTerm').resolves(true); + await session.activateRSession(); + sinon.assert.calledOnce(createRTerm); + }); + + test('a stale registration Disposable cannot remove its replacement', async () => { + const terminal = makeTerminal(); + setTerminals([terminal]); + const oldAttach = sandbox.stub().resolves(false); + const oldRegistration = session.registerTerminalIntegration(terminal, { attach: oldAttach }); + oldRegistration.dispose(); + const newAttach = sandbox.stub().resolves(false); + const newRegistration = session.registerTerminalIntegration(terminal, { attach: newAttach }); + try { + oldRegistration.dispose(); + await session.activateRSession(); + sinon.assert.notCalled(oldAttach); + sinon.assert.calledOnce(newAttach); + } finally { newRegistration.dispose(); } + }); + + test('an old pending handler completion cannot clear a replacement handler run', async () => { + const terminal = makeTerminal(); + setTerminals([terminal]); + const oldGate = deferred(); + const oldStarted = deferred(); + const oldRegistration = session.registerTerminalIntegration(terminal, { + attach: () => { oldStarted.resolve(); return oldGate.promise; }, + }); + const oldOperation = session.activateRSession(); + await oldStarted.promise; + oldRegistration.dispose(); + + const newGate = deferred(); + const newStarted = deferred(); + const newAttach = sandbox.stub().callsFake(() => { newStarted.resolve(); return newGate.promise; }); + const newRegistration = session.registerTerminalIntegration(terminal, { attach: newAttach }); + try { + const newOperation = session.activateRSession(); + await newStarted.promise; + oldGate.resolve(true); + await oldOperation; + await session.activateRSession(); + sinon.assert.calledOnce(newAttach); + newGate.resolve(false); + await newOperation; + } finally { + oldGate.resolve(false); + newGate.resolve(false); + newRegistration.dispose(); + } + }); + + test('another session selection during the hook is preserved while terminal binding remains usable', async () => { + const terminal = makeTerminal(); + setTerminals([terminal]); + const other = session.registerSessionTransport('integration-other-selection', 'host', '/tmp', () => Promise.resolve({})); + const gate = deferred(); + const started = deferred(); + const registration = session.registerTerminalIntegration(terminal, { + attach: async () => { + const socket = await connections.attach('integration-select-false', 55102, 'external-host', { + attachmentId: 'select-false-binding', autoActivate: false, + }); + const reference = await waitForValue(() => session.getSessionReference('integration-select-false')); + const success = await session.activateSessionById(reference.sessionId, { + connectionId: reference.connectionId, terminal, select: false, + }); + started.resolve(); + await gate.promise; + return success && !socket.destroyed; + }, + }); + try { + const operation = session.activateRSession(); + await started.promise; + await session.activateSessionById(other.sessionId); + gate.resolve(true); + await operation; + assert.strictEqual(session.activeSession, other, 'hook completion must not take back a newer session selection'); + const reference = session.getSessionReference('integration-select-false'); + assert.ok(reference); + assert.strictEqual(reference.attachmentId, 'select-false-binding'); + assert.strictEqual(await session.activateSessionById(reference.sessionId, { + connectionId: reference.connectionId, terminal, select: false, + }), true, 'the binding must remain available after a newer selection'); + assert.strictEqual(session.activeSession, other); + assert.strictEqual(await session.activateSessionById(reference.sessionId, { connectionId: reference.connectionId }), true); + await session.executeSessionCode(session.activeSession!, 'View(iris)'); + sinon.assert.calledOnceWithExactly(terminalSpies.get(terminal)!.sendText, 'View(iris)'); + } finally { + gate.resolve(true); + registration.dispose(); + session.unregisterSessionTransport(other); + await session.cleanupSession('integration-select-false'); + } + }); + + test('integration registration survives a disconnect and can attach a replacement connection', async () => { + const terminal = makeTerminal(); + setTerminals([terminal]); + let connectionsMade = 0; + const registration = session.registerTerminalIntegration(terminal, { + attach: async token => { + assert.strictEqual(token.isCancellationRequested, false); + connectionsMade++; + const id = 'integration-reconnect'; + const socket = await connections.attach(id, 55103 + connectionsMade, 'external-host', { + attachmentId: `reconnect-${connectionsMade}`, autoActivate: false, + }); + const reference = await waitForValue(() => session.getSessionReference(id)); + return !socket.destroyed && await session.activateSessionById(id, { + connectionId: reference.connectionId, terminal, select: false, + }); + }, + }); + try { + await session.activateRSession(); + await session.cleanupSession('integration-reconnect'); + await session.activateRSession(); + assert.strictEqual(connectionsMade, 2); + assert.strictEqual(session.getSessionReference('integration-reconnect')?.attachmentId, 'reconnect-2'); + } finally { + registration.dispose(); + await session.cleanupSession('integration-reconnect'); + } + }); +}); diff --git a/src/test/suite/sessionTerminalLifecycle.test.ts b/src/test/suite/sessionTerminalLifecycle.test.ts index d059897c8..bec23f2f2 100644 --- a/src/test/suite/sessionTerminalLifecycle.test.ts +++ b/src/test/suite/sessionTerminalLifecycle.test.ts @@ -260,6 +260,7 @@ suite('Session Terminal Lifecycle', () => { if (explicitBinding) { const api: RSessionApi = { getConnectionInfo: session.getConnectionInfo, prepareAttachment: session.prepareAttachment, onDidChangeSessionConnection: session.onDidChangeSessionConnection, getSessionReference: session.getSessionReference, + registerTerminalIntegration: session.registerTerminalIntegration, activate: session.activateSessionById }; assert.strictEqual(await api.activate(owner.sessionId, { terminal }), true); } @@ -290,6 +291,7 @@ suite('Session Terminal Lifecycle', () => { } as unknown as vscode.WorkspaceConfiguration); const api: RSessionApi = { getConnectionInfo: session.getConnectionInfo, prepareAttachment: session.prepareAttachment, onDidChangeSessionConnection: session.onDidChangeSessionConnection, getSessionReference: session.getSessionReference, + registerTerminalIntegration: session.registerTerminalIntegration, activate: session.activateSessionById }; await connections.attach('native-moved-to-explicit', 46280); const owner = session.activeSession; @@ -368,6 +370,7 @@ suite('Session Terminal Lifecycle', () => { } as unknown as vscode.WorkspaceConfiguration); const api: RSessionApi = { getConnectionInfo: session.getConnectionInfo, prepareAttachment: session.prepareAttachment, onDidChangeSessionConnection: session.onDidChangeSessionConnection, getSessionReference: session.getSessionReference, + registerTerminalIntegration: session.registerTerminalIntegration, activate: session.activateSessionById }; const old = await connections.attach('selected-reconnect', 46301); const previous = session.activeSession; diff --git a/src/test/suite/workspaceViewer.test.ts b/src/test/suite/workspaceViewer.test.ts index f074debc7..d9937509b 100644 --- a/src/test/suite/workspaceViewer.test.ts +++ b/src/test/suite/workspaceViewer.test.ts @@ -131,6 +131,7 @@ suite('Workspace Viewer', () => { } as unknown as vscode.WorkspaceConfiguration); const api: RSessionApi = { getConnectionInfo: session.getConnectionInfo, prepareAttachment: session.prepareAttachment, onDidChangeSessionConnection: session.onDidChangeSessionConnection, getSessionReference: session.getSessionReference, + registerTerminalIntegration: session.registerTerminalIntegration, activate: session.activateSessionById }; assert.strictEqual(await api.activate(first.sessionId, { terminal }), true); session.updateSessionWorkspace(first, data('table')); From d858844aa450df195f9062ccf83180758ca9b8cb Mon Sep 17 00:00:00 2001 From: eitsupi Date: Mon, 12 Oct 2026 03:33:59 +0000 Subject: [PATCH 2/6] test: isolate external terminal routing from paste and focus settings --- src/test/suite/sessionTerminalIntegration.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/test/suite/sessionTerminalIntegration.test.ts b/src/test/suite/sessionTerminalIntegration.test.ts index 00e31f890..b6057e1a5 100644 --- a/src/test/suite/sessionTerminalIntegration.test.ts +++ b/src/test/suite/sessionTerminalIntegration.test.ts @@ -62,7 +62,9 @@ suite('External Terminal Integration', () => { showError = sandbox.stub(vscode.window, 'showErrorMessage').resolves(undefined); showInformation = sandbox.stub(vscode.window, 'showInformationMessage').resolves(undefined); sandbox.stub(util, 'config').returns({ - get: (key: string) => ({ sessionWatcher: true, alwaysUseActiveTerminal: false, bracketedPaste: true })[key], + get: (key: string) => ({ + sessionWatcher: true, alwaysUseActiveTerminal: false, bracketedPaste: false, 'source.focus': 'none', + })[key], } as unknown as vscode.WorkspaceConfiguration); session.deploySessionWatcher(extensionRoot); await session.getGlobalPipePath(); From 31cfa5e3c71e111f5ec9d20a03e8d0be64e4c131 Mon Sep 17 00:00:00 2001 From: eitsupi Date: Mon, 12 Oct 2026 03:38:27 +0000 Subject: [PATCH 3/6] test: synchronize native httpgd WebSocket fixture --- src/test/node/httpgdClient.test.ts | 143 +++++++++++++++++++++++++---- 1 file changed, 123 insertions(+), 20 deletions(-) diff --git a/src/test/node/httpgdClient.test.ts b/src/test/node/httpgdClient.test.ts index 3e2a9dc07..e3d97395a 100644 --- a/src/test/node/httpgdClient.test.ts +++ b/src/test/node/httpgdClient.test.ts @@ -30,7 +30,6 @@ async function until(condition: () => boolean): Promise { suite('Internal httpgd client', () => { let server: Server; - let upgradedSocket: Duplex | undefined; let client: HttpgdClient; let origin: string; let remote: HttpgdState; @@ -44,7 +43,6 @@ suite('Internal httpgd client', () => { const binary = Buffer.from([0, 255, 128, 13, 10]); setup(async () => { - upgradedSocket = undefined; remote = { upid: 1, hsize: 1, active: true }; ids = ['1']; requests = []; @@ -80,7 +78,6 @@ suite('Internal httpgd client', () => { }); teardown(async () => { client.disconnect(); - upgradedSocket?.destroy(); if (websocketDescriptor) { Object.defineProperty(globalThis, 'WebSocket', websocketDescriptor); } else { Reflect.deleteProperty(globalThis, 'WebSocket'); } for (const response of held) { response.destroy(); } @@ -140,25 +137,131 @@ suite('Internal httpgd client', () => { await until(() => requests.filter(request => request.path === '/plots').length >= 5); }); test('receives notifications through an actual native WebSocket connection', async function () { - if (typeof nativeWebSocket !== 'function') { this.skip(); } - globalThis.WebSocket = nativeWebSocket; - server.on('upgrade', (request, socket) => { - assert.strictEqual(new URL(request.url ?? '/', origin).searchParams.get('token'), 'secret'); - const key = request.headers['sec-websocket-key']; - assert.strictEqual(typeof key, 'string'); - const accept = createHash('sha1').update(`${String(key)}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64'); - socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`); + if (typeof nativeWebSocket !== 'function') { this.skip(); return; } + + const handshakeTimeoutMs = 5000; + const testOrigin = origin; + const actualServer = server; + let upgradedSocket: Duplex | undefined; + let tearingDown = false; + let resolveClientFailure!: (error: Error) => void; + let resolveServerFailure!: (error: Error) => void; + const clientFailure = new Promise(resolve => { resolveClientFailure = resolve; }); + const serverFailure = new Promise(resolve => { resolveServerFailure = resolve; }); + const socketFailure = Promise.race([clientFailure, serverFailure]).then(error => { throw error; }); + let resolveClientOpen!: () => void; + let rejectClientOpen!: (error: Error) => void; + let clientOpenSettled = false; + let clientOpenTimer: NodeJS.Timeout | undefined; + const clientOpened = new Promise((resolve, reject) => { + resolveClientOpen = () => { + if (clientOpenSettled) { return; } + clientOpenSettled = true; + clearTimeout(clientOpenTimer); + resolve(); + }; + rejectClientOpen = error => { + if (clientOpenSettled) { return; } + clientOpenSettled = true; + clearTimeout(clientOpenTimer); + reject(error); + }; + }); + clientOpenTimer = setTimeout(() => rejectClientOpen(new Error(`Timed out waiting for native WebSocket open after ${handshakeTimeoutMs}ms`)), handshakeTimeoutMs); + void clientOpened.catch(() => undefined); + class CapturingWebSocket extends nativeWebSocket { + constructor(url: string | URL) { + super(url); + this.addEventListener('open', resolveClientOpen, { once: true }); + this.addEventListener('error', () => { + const error = new Error('Native WebSocket failed before or during the test connection.'); + rejectClientOpen(error); + if (!tearingDown) { resolveClientFailure(error); } + }); + this.addEventListener('close', () => rejectClientOpen(new Error('Native WebSocket closed before it opened.')), { once: true }); + } + } + + let upgradeTimer: NodeJS.Timeout | undefined; + let upgradeHandler: ((request: import('node:http').IncomingMessage, socket: Duplex) => void) | undefined; + let resolveUpgrade!: (socket: Duplex) => void; + let rejectUpgrade!: (error: Error) => void; + let upgradeSettled = false; + const upgraded = new Promise((resolve, reject) => { + resolveUpgrade = socket => { + if (upgradeSettled) { return; } + upgradeSettled = true; + clearTimeout(upgradeTimer); + resolve(socket); + }; + rejectUpgrade = error => { + if (upgradeSettled) { return; } + upgradeSettled = true; + clearTimeout(upgradeTimer); + reject(error); + }; + }); + void upgraded.catch(() => undefined); + upgradeHandler = (request, socket) => { upgradedSocket = socket; + socket.on('error', error => { + if (!tearingDown) { resolveServerFailure(error); } + }); + try { + assert.strictEqual(new URL(request.url ?? '/', testOrigin).searchParams.get('token'), 'secret'); + const key = request.headers['sec-websocket-key']; + assert.strictEqual(typeof key, 'string'); + const accept = createHash('sha1').update(`${String(key)}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`).digest('base64'); + socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`); + resolveUpgrade(socket); + } catch (error) { + rejectUpgrade(error instanceof Error ? error : new Error(String(error))); + } + }; + actualServer.on('upgrade', upgradeHandler); + upgradeTimer = setTimeout(() => rejectUpgrade(new Error(`Timed out waiting for HTTP WebSocket upgrade after ${handshakeTimeoutMs}ms`)), handshakeTimeoutMs); + + globalThis.WebSocket = CapturingWebSocket as typeof WebSocket; + client = new HttpgdClient(testOrigin, 'secret', { + pollIntervalMs: handshakeTimeoutMs + 1000, + retryIntervalMs: handshakeTimeoutMs + 1000, + // Exercise a real native handshake; short deadlines remain in fake timeout tests. + webSocketTimeoutMs: handshakeTimeoutMs, }); - await client.connect(); - await until(() => notifications.length === 2); - ids = ['2']; remote = { ...remote, upid: 2 }; - const message = Buffer.from(JSON.stringify(remote)); - // The fixture's state JSON is a short, unmasked server text frame. - assert.ok(message.length < 126); - assert.ok(upgradedSocket); - upgradedSocket.write(Buffer.concat([Buffer.from([0x81, message.length]), message])); - await until(() => client.getPlots()[0]?.id === '2'); + const actualClient = client; + const actualNotifications: string[][] = []; + actualClient.onPlotsChanged(value => actualNotifications.push(value.plots.map(plot => plot.id))); + + try { + await Promise.race([actualClient.connect(), socketFailure]); + const [socket] = await Promise.race([ + Promise.all([upgraded, clientOpened]), + socketFailure, + ]); + assert.ok(socket); + // Wait for the onopen resync to finish before injecting the state change. + await Promise.race([ + until(() => actualNotifications.length >= 2), + socketFailure, + ]); + ids = ['2']; remote = { ...remote, upid: 2 }; + const message = Buffer.from(JSON.stringify(remote)); + // The fixture's state JSON is a short, unmasked server text frame. + assert.ok(message.length < 126); + socket.write(Buffer.concat([Buffer.from([0x81, message.length]), message])); + await Promise.race([ + until(() => actualClient.getPlots()[0]?.id === '2' && actualNotifications.some(value => value.length === 1 && value[0] === '2')), + socketFailure, + ]); + assert.ok(actualNotifications.some(value => value.length === 1 && value[0] === '2')); + } finally { + tearingDown = true; + clearTimeout(upgradeTimer); + clearTimeout(clientOpenTimer); + if (upgradeHandler) { actualServer.removeListener('upgrade', upgradeHandler); } + actualClient.disconnect(); + upgradedSocket?.destroy(); + } }); test('falls back when WebSocket cannot open', async () => { globalThis.WebSocket = TestWebSocket as unknown as typeof WebSocket; From 5a9939434f7d170a77390eb472a8c0115bdaf116 Mon Sep 17 00:00:00 2001 From: eitsupi Date: Mon, 12 Oct 2026 03:53:00 +0000 Subject: [PATCH 4/6] fix: keep external terminal attachment busy until completion --- src/api.d.ts | 18 +++--- src/session.ts | 10 ++-- .../suite/sessionTerminalIntegration.test.ts | 56 ++++++++++++++++++- 3 files changed, 71 insertions(+), 13 deletions(-) diff --git a/src/api.d.ts b/src/api.d.ts index 3dae0890e..cd0fea7d4 100644 --- a/src/api.d.ts +++ b/src/api.d.ts @@ -140,15 +140,19 @@ export interface RSessionApi { * invoked before native PID discovery; false, rejection, busy, or cancellation never falls * back to a native attach command. On true, vscode-R requires a current live session binding * for this terminal, then focuses that session. Consumers should bind with - * `activate(..., { terminal, connectionId, select: false })` before resolving true. If the - * terminal already has a current live owner, that owner is focused and the hook is skipped. - * While a hook is in flight, repeated activations of that terminal do not invoke it again. + * `activate(..., { terminal, connectionId, select: false })` before resolving true. When no + * hook is in flight, a current live owner is focused and the hook is skipped. While a hook is + * in flight, repeated activations report busy without selecting or focusing its bound owner, + * and do not invoke another handler. The terminal remains busy until the handler promise + * settles, even after cancellation is requested. * Registered terminals are excluded from native PID association, so consumers must bind * their session explicitly. Registration during host shutdown throws. - * If the user selects another session while the hook is pending, vscode-R leaves that selection in - * place and retains the terminal binding. Cancellation is requested when focus leaves the - * terminal, it closes, registration is disposed, or the host shuts down; submitted R code - * cannot be recalled. + * If the user selects another session while the hook is pending, vscode-R leaves that selection + * in place and retains the terminal binding. Cancellation is requested when focus leaves the + * terminal, it closes, registration is disposed, or the host shuts down. Consumers should + * stop cancellable waits and settle the handler on cancellation. Submitted R code cannot be + * recalled; the consumer must retain its preparation while the runtime keeps its submission + * queue occupied until execution finishes. */ registerTerminalIntegration(terminal: vscode.Terminal, integration: RExternalTerminalIntegration): vscode.Disposable; } diff --git a/src/session.ts b/src/session.ts index 01409db5b..60b4b8ccf 100644 --- a/src/session.ts +++ b/src/session.ts @@ -1543,16 +1543,16 @@ export async function activateRSession(): Promise { } }); }; + const registration = terminal ? terminalIntegrations.get(terminal) : undefined; + if (registration?.active) { + void window.showInformationMessage('An R session attach is already in progress for this terminal.'); + return; + } const focused = focusOwner(); if (focused) { await focused; return; } if (terminal) { - const registration = terminalIntegrations.get(terminal); if (registration) { - if (registration.active) { - void window.showInformationMessage('An R session attach is already in progress for this terminal.'); - return; - } if (!isLiveTerminal(terminal) || window.activeTerminal !== terminal) { return; } // Supersede native attach operations that may still be waiting for PID or command I/O. sessionSelectionRevision++; diff --git a/src/test/suite/sessionTerminalIntegration.test.ts b/src/test/suite/sessionTerminalIntegration.test.ts index b6057e1a5..788aedbf1 100644 --- a/src/test/suite/sessionTerminalIntegration.test.ts +++ b/src/test/suite/sessionTerminalIntegration.test.ts @@ -239,6 +239,7 @@ suite('External Terminal Integration', () => { const terminal = makeTerminal(); setTerminals([terminal]); const owner = session.registerSessionTransport('integration-busy-owner', 'host', '/tmp', () => Promise.resolve({})); + const createRTerm = sandbox.stub(rTerminal, 'createRTerm').resolves(true); const gate = deferred(); const started = deferred(); const attach = sandbox.stub().callsFake(() => { started.resolve(); return gate.promise; }); @@ -246,10 +247,14 @@ suite('External Terminal Integration', () => { try { const first = session.activateRSession(); await started.promise; + assert.strictEqual(await session.activateSessionById(owner.sessionId, { terminal, select: false }), true); await session.activateRSession(); sinon.assert.calledOnce(attach); sinon.assert.calledOnceWithExactly(showInformation, 'An R session attach is already in progress for this terminal.'); - assert.strictEqual(await session.activateSessionById(owner.sessionId, { terminal, select: false }), true); + assert.strictEqual(session.activeSession, undefined, 'busy click must not focus a bound owner'); + sinon.assert.notCalled(terminalSpies.get(terminal)!.show); + sinon.assert.notCalled(terminalSpies.get(terminal)!.sendText); + sinon.assert.notCalled(createRTerm); gate.resolve(true); await first; assert.strictEqual(session.activeSession, owner, 'the successful hook should still select its live bound owner'); @@ -261,6 +266,55 @@ suite('External Terminal Integration', () => { } }); + test('busy clicks stay inert while a bound hook awaits decline or cancellation completion', async () => { + const createRTerm = sandbox.stub(rTerminal, 'createRTerm').resolves(true); + for (const outcome of ['false', 'cancelled'] as const) { + const terminal = makeTerminal(); + const other = makeTerminal(); + setTerminals(outcome === 'cancelled' ? [terminal, other] : [terminal]); + const owner = session.registerSessionTransport(`integration-busy-${outcome}`, 'host', '/tmp', () => Promise.resolve({})); + const gate = deferred(); + const started = deferred(); + const attach = sandbox.stub().callsFake((token: vscode.CancellationToken) => { + started.resolve(token); + return gate.promise; + }); + const registration = session.registerTerminalIntegration(terminal, { attach }); + try { + const first = session.activateRSession(); + const token = await started.promise; + assert.strictEqual(await session.activateSessionById(owner.sessionId, { terminal, select: false }), true); + if (outcome === 'cancelled') { + activeTerminal.value(other); + activeEvents.fire(other); + activeTerminal.value(terminal); + activeEvents.fire(terminal); + assert.strictEqual(token.isCancellationRequested, true); + } + + await session.activateRSession(); + sinon.assert.calledOnce(attach); + sinon.assert.calledOnceWithExactly(showInformation, + 'An R session attach is already in progress for this terminal.'); + assert.strictEqual(session.activeSession, undefined, `${outcome} busy click must not focus the bound owner`); + sinon.assert.notCalled(terminalSpies.get(terminal)!.show); + sinon.assert.notCalled(terminalSpies.get(terminal)!.sendText); + sinon.assert.notCalled(createRTerm); + + gate.resolve(outcome === 'cancelled'); + await first; + assert.strictEqual(session.activeSession, undefined, `${outcome} hook must not activate its owner`); + sinon.assert.notCalled(terminalSpies.get(terminal)!.show); + } finally { + gate.resolve(false); + registration.dispose(); + session.unregisterSessionTransport(owner); + } + showInformation.resetHistory(); + createRTerm.resetHistory(); + } + }); + test('focus away and back cancels the hook without reviving it or attaching natively', async () => { const terminal = makeTerminal(); const other = makeTerminal(); From 3ba94a3a3bc71c539dc8010c0d0c401e30e10b7b Mon Sep 17 00:00:00 2001 From: eitsupi Date: Mon, 12 Oct 2026 11:39:15 +0000 Subject: [PATCH 5/6] fix: invalidate native PID waits with terminal ownership --- src/nativeTerminalPidWait.ts | 123 ++++++++++++ src/session.ts | 172 +++++++++++++---- src/test/node/nativeTerminalPidWait.test.ts | 170 +++++++++++++++++ src/test/suite/session.test.ts | 4 + .../suite/sessionTerminalIntegration.test.ts | 178 +++++++++++++++++- 5 files changed, 613 insertions(+), 34 deletions(-) create mode 100644 src/nativeTerminalPidWait.ts create mode 100644 src/test/node/nativeTerminalPidWait.test.ts diff --git a/src/nativeTerminalPidWait.ts b/src/nativeTerminalPidWait.ts new file mode 100644 index 000000000..e545377e0 --- /dev/null +++ b/src/nativeTerminalPidWait.ts @@ -0,0 +1,123 @@ +export type NativeTerminalPidWaitResult = + | { readonly kind: 'resolved'; readonly value: Value; readonly isCurrent: () => boolean } + | { readonly kind: 'invalidated'; readonly reason: 'terminal' | 'lifecycle' | 'operation' }; + +export interface NativeTerminalPidWaitOptions { + readonly signal?: AbortSignal; + readonly isEligible?: () => boolean; +} + +interface NativeTerminalPidWaiter { + readonly terminal: T; + invalidate(reason: 'terminal' | 'lifecycle'): void; +} + +/** Interruptible reads of VS Code's otherwise uncancellable terminal processId promise. */ +export class NativeTerminalPidWaitRegistry { + private lifecycleRevision = 0; + private readonly terminalRevisions = new WeakMap(); + private readonly waiters = new Set>(); + + public invalidate(terminal: T): void { + this.terminalRevisions.set(terminal, this.revisionOf(terminal) + 1); + for (const waiter of this.waiters) { + if (waiter.terminal === terminal) { waiter.invalidate('terminal'); } + } + } + + public invalidateAll(): void { + this.lifecycleRevision++; + for (const waiter of this.waiters) { waiter.invalidate('lifecycle'); } + } + + public waitFor( + terminal: T, + read: () => PromiseLike | Value, + options: NativeTerminalPidWaitOptions = {}, + ): Promise> { + const terminalRevision = this.revisionOf(terminal); + const lifecycleRevision = this.lifecycleRevision; + + return new Promise((resolve, reject) => { + let settled = false; + const isCurrent = (): boolean => !options.signal?.aborted + && lifecycleRevision === this.lifecycleRevision + && terminalRevision === this.revisionOf(terminal) + && (options.isEligible?.() ?? true); + const checkCurrent = (): boolean => { + try { return isCurrent(); } + catch (error) { finishError(error); return false; } + }; + const finish = (result: NativeTerminalPidWaitResult): void => { + if (settled) { return; } + settled = true; + cleanup(); + resolve(result); + }; + const finishError = (error: unknown): void => { + if (settled) { return; } + settled = true; + cleanup(); + reject(error); + }; + const waiter: NativeTerminalPidWaiter = { + terminal, + invalidate: reason => finish({ kind: 'invalidated', reason }), + }; + const onAbort = () => finish({ kind: 'invalidated', reason: 'operation' }); + const cleanup = (): void => { + this.waiters.delete(waiter); + options.signal?.removeEventListener('abort', onAbort); + }; + + // Install invalidation listeners before reading processId: its getter may + // synchronously trigger registration, close, or shutdown callbacks. + this.waiters.add(waiter); + options.signal?.addEventListener('abort', onAbort, { once: true }); + if (options.signal?.aborted) { + onAbort(); + return; + } + if (lifecycleRevision !== this.lifecycleRevision) { + finish({ kind: 'invalidated', reason: 'lifecycle' }); + return; + } + let eligible: boolean; + try { eligible = options.isEligible?.() ?? true; } + catch (error) { finishError(error); return; } + if (terminalRevision !== this.revisionOf(terminal) || !eligible) { + finish({ kind: 'invalidated', reason: 'terminal' }); + return; + } + + let pending: Promise; + try { pending = Promise.resolve(read()); } + catch (error) { finishError(error); return; } + // Keep a rejection observer after invalidation so the underlying VS Code + // promise cannot become an unhandled rejection if it settles late. + void pending.then(value => { + if (settled) { return; } + if (!checkCurrent()) { + if (settled) { return; } + const reason = options.signal?.aborted ? 'operation' + : lifecycleRevision !== this.lifecycleRevision ? 'lifecycle' : 'terminal'; + finish({ kind: 'invalidated', reason }); + return; + } + finish({ kind: 'resolved', value, isCurrent }); + }, error => { + if (settled) { return; } + if (!checkCurrent()) { + if (settled) { return; } + const reason = options.signal?.aborted ? 'operation' + : lifecycleRevision !== this.lifecycleRevision ? 'lifecycle' : 'terminal'; + finish({ kind: 'invalidated', reason }); + return; + } + finishError(error); + }); + }); + } + + private revisionOf(terminal: T): number { return this.terminalRevisions.get(terminal) ?? 0; } +} diff --git a/src/session.ts b/src/session.ts index 60b4b8ccf..04edee001 100644 --- a/src/session.ts +++ b/src/session.ts @@ -17,6 +17,7 @@ import { getProcessAncestors } from './processTree'; import { TerminalSessionRegistry } from './terminalSessionRegistry'; import { SessionProcessMonitor } from './sessionProcessMonitor'; import { sessionProcessIdentity } from './sessionIdentity'; +import { NativeTerminalPidWaitRegistry } from './nativeTerminalPidWait'; import { purgeAddinPickerItems, RSEditOperation, RSRange } from './rstudioapi'; import { extensionContext, rWorkspace, globalRHelp, globalPlotManager, sessionStatusBarItem, enableSessionWatcher } from './extension'; @@ -184,11 +185,17 @@ export function getSessionReference(sessionId: string): RSessionReference | unde const sessionProcessMonitor = new SessionProcessMonitor(isLocalHost); // Only the newest handshake for a stable ID may commit after terminal discovery. -const pendingSessionAttachments = new Map(); +interface PendingSessionAttachment { + readonly socket: IpcSocket; + readonly controller: AbortController; +} +const pendingSessionAttachments = new Map(); const terminalRegistry = new TerminalSessionRegistry( target => sessions.get(target.sessionId) === target && isConnectedSession(target), terminal => !terminal.exitStatus && window.terminals.includes(terminal), ); +const nativeTerminalPidWaits = new NativeTerminalPidWaitRegistry(); +const pendingManualNativePidWaits = new WeakMap(); interface TerminalIntegrationRun { readonly cancellation: vscode.CancellationTokenSource; cancel(): void; @@ -725,6 +732,8 @@ function isLiveTerminal(terminal: vscode.Terminal): boolean { /** Called for every terminal close, including extension-owned pseudoterminals. */ export function cleanupTerminalBinding(terminal: vscode.Terminal): void { + nativeTerminalPidWaits.invalidate(terminal); + pendingManualNativePidWaits.get(terminal)?.abort(); terminalRegistry.closeTerminal(terminal); removeTerminalIntegration(terminal); } @@ -785,6 +794,7 @@ export function registerTerminalIntegration( }), }; terminalIntegrations.set(terminal, registration); + nativeTerminalPidWaits.invalidate(terminal); // Invalidate an earlier native PID/command path that may still be awaiting I/O. if (window.activeTerminal === terminal) { sessionSelectionRevision++; } return { dispose: () => removeTerminalIntegrationRegistration(registration) }; @@ -831,10 +841,16 @@ export async function createSessionDiscoveryFile(endpoint: string): Promise { +async function writeSessionDiscoveryFile( + filePath: string, + endpoint: string, + terminalPid?: number, + shouldPublish: () => boolean = () => true, +): Promise { if (!isExtensionDiscoveryPath(filePath)) { throw new Error('Refusing to write session discovery data outside extension global storage'); } + if (!shouldPublish()) { return; } const data: SessionDiscoveryFile = { version: 1, endpoint, jgdSocket: getSessionJgdSocket() }; if (terminalPid !== undefined) { data.terminalPid = terminalPid; @@ -843,17 +859,19 @@ async function writeSessionDiscoveryFile(filePath: string, endpoint: string, ter if (typeof existing.terminalPid === 'number') { data.terminalPid = existing.terminalPid; } + if (!shouldPublish()) { return; } } await fsp.mkdir(path.dirname(filePath), { recursive: true }); const temporaryPath = `${filePath}.${crypto.randomBytes(8).toString('hex')}.tmp`; + let published = false; try { await fsp.writeFile(temporaryPath, `${JSON.stringify(data)}\n`, { mode: 0o600 }); await setOwnerOnlyPermissions(temporaryPath); + if (!shouldPublish()) { return; } await fsp.rename(temporaryPath, filePath); - await setOwnerOnlyPermissions(filePath); - } catch (error) { - await fsp.rm(temporaryPath, { recursive: true, force: true }).catch(() => undefined); - throw error; + published = true; + } finally { + if (!published) { await fsp.rm(temporaryPath, { recursive: true, force: true }).catch(() => undefined); } } } @@ -866,10 +884,11 @@ export async function updateTerminalSessionDiscoveryFile( filePath: string, endpoint: string, terminalPid?: number, + isCurrent: () => boolean = () => true, ): Promise { await queueTerminalDiscoveryOperation(terminal, async () => { - if (!isTerminalClosed(terminal)) { - await writeSessionDiscoveryFile(filePath, endpoint, terminalPid); + if (!isTerminalClosed(terminal) && isCurrent()) { + await writeSessionDiscoveryFile(filePath, endpoint, terminalPid, isCurrent); } }); } @@ -906,35 +925,55 @@ export async function refreshTerminalDiscoveryFiles( pipePath: string, terminals: readonly vscode.Terminal[] = vscode.window.terminals, ): Promise { + const generation = attachmentLifecycleGeneration; + if (attachmentLifecycleStopping) { return; } for (const term of terminals) { - if (isTerminalClosed(term)) { + if (generation !== attachmentLifecycleGeneration || attachmentLifecycleStopping) { return; } + if (!isNativeTerminalEligible(term)) { continue; } const envPath = terminalDiscoveryPath(term); - const terminalPid = await term.processId; - if (!terminalPid || isTerminalClosed(term)) { + const pidWait = await nativeTerminalPidWaits.waitFor(term, () => term.processId, { + isEligible: () => isNativeTerminalEligible(term), + }); + if (pidWait.kind === 'invalidated') { + if (pidWait.reason === 'lifecycle') { return; } + continue; + } + const terminalPid = pidWait.value; + const isCurrent = () => generation === attachmentLifecycleGeneration + && !attachmentLifecycleStopping && pidWait.isCurrent() && isNativeTerminalEligible(term); + if (!terminalPid || !isCurrent()) { continue; } let discoveryPath = envPath; discoveryPath ??= await findDiscoveryFileForTerminal(terminalPid); - if (discoveryPath && !isTerminalClosed(term)) { - await updateTerminalSessionDiscoveryFile(term, discoveryPath, pipePath, terminalPid); + if (discoveryPath && isCurrent()) { + await updateTerminalSessionDiscoveryFile(term, discoveryPath, pipePath, terminalPid, isCurrent); } } } export async function updateTerminalDiscovery(terminal: vscode.Terminal): Promise { - if (!globalPipePath || isTerminalClosed(terminal)) { + const generation = attachmentLifecycleGeneration; + const endpoint = globalPipePath; + if (!endpoint || attachmentLifecycleStopping || !isNativeTerminalEligible(terminal)) { return; } - const terminalPid = await terminal.processId; - if (terminalPid === undefined || isTerminalClosed(terminal)) { + const pidWait = await nativeTerminalPidWaits.waitFor(terminal, () => terminal.processId, { + isEligible: () => isNativeTerminalEligible(terminal), + }); + if (pidWait.kind === 'invalidated') { return; } + const terminalPid = pidWait.value; + const isCurrent = () => generation === attachmentLifecycleGeneration && !attachmentLifecycleStopping + && endpoint === globalPipePath && pidWait.isCurrent() && isNativeTerminalEligible(terminal); + if (terminalPid === undefined || !isCurrent()) { return; } let discoveryPath = terminalDiscoveryPath(terminal); discoveryPath ??= await findDiscoveryFileForTerminal(terminalPid); - if (discoveryPath && !isTerminalClosed(terminal)) { - await updateTerminalSessionDiscoveryFile(terminal, discoveryPath, globalPipePath, terminalPid); + if (discoveryPath && isCurrent()) { + await updateTerminalSessionDiscoveryFile(terminal, discoveryPath, endpoint, terminalPid, isCurrent); } } @@ -1074,7 +1113,10 @@ function startGlobalSessionServer(generation: number): Promise { readBufferLength = 0; activeConnections.delete(socket); for (const [id, pending] of pendingSessionAttachments) { - if (pending === socket) { pendingSessionAttachments.delete(id); } + if (pending.socket === socket) { + pendingSessionAttachments.delete(id); + pending.controller.abort(); + } } for (const [id, pending] of pendingRequests.entries()) { if (pending.socket === socket) { @@ -1434,6 +1476,7 @@ async function removePathIfExists(pathLike: string): Promise { } export async function shutdownSessionWatcher(): Promise { + nativeTerminalPidWaits.invalidateAll(); attachmentLifecycleGeneration++; attachmentLifecycleStopping = true; const terminalIntegrationSnapshot = [...terminalIntegrations.values()]; @@ -1461,8 +1504,11 @@ export async function shutdownSessionWatcher(): Promise { if (activeSession === target) { clearActive = true; } } if (clearActive) { void clearActiveSession(); } - for (const [sessionId, socket] of pendingSessionAttachments) { - if (socket._server === serverAtShutdown) { pendingSessionAttachments.delete(sessionId); } + for (const [sessionId, pending] of pendingSessionAttachments) { + if (pending.socket._server === serverAtShutdown) { + pendingSessionAttachments.delete(sessionId); + pending.controller.abort(); + } } // Queue removal of the fixed manual-attach path before a redeployed lifecycle // can publish a replacement at the same path. @@ -1493,7 +1539,10 @@ export async function shutdownSessionWatcher(): Promise { for (const socket of oldSockets) { socket.destroy(); for (const [sessionId, pending] of pendingSessionAttachments) { - if (pending === socket) { pendingSessionAttachments.delete(sessionId); } + if (pending.socket === socket) { + pendingSessionAttachments.delete(sessionId); + pending.controller.abort(); + } } } if (pipeClient && pipeClient._server === server) { pipeClient = undefined; } @@ -1527,6 +1576,7 @@ export async function shutdownSessionWatcher(): Promise { } export async function activateRSession(): Promise { + const lifecycleGeneration = attachmentLifecycleGeneration; if (!config().get('sessionWatcher')) { void window.showInformationMessage('This command requires that r.sessionWatcher be enabled.'); return; @@ -1548,6 +1598,7 @@ export async function activateRSession(): Promise { void window.showInformationMessage('An R session attach is already in progress for this terminal.'); return; } + if (terminal) { pendingManualNativePidWaits.get(terminal)?.abort(); } const focused = focusOwner(); if (focused) { await focused; return; } @@ -1607,12 +1658,36 @@ export async function activateRSession(): Promise { } const selectionRevision = ++sessionSelectionRevision; - const isCurrent = () => selectionRevision === sessionSelectionRevision && terminal === window.activeTerminal + let nativePidWaitIsCurrent = () => true; + const isCurrent = () => lifecycleGeneration === attachmentLifecycleGeneration + && !attachmentLifecycleStopping + && selectionRevision === sessionSelectionRevision && terminal === window.activeTerminal + && nativePidWaitIsCurrent() && (!terminal || (!isTerminalClosed(terminal) && !terminal.exitStatus)); let discoveryPath = terminal && terminalDiscoveryPath(terminal); if (terminal && !discoveryPath) { - const terminalPid = await terminal.processId; + const controller = new AbortController(); + pendingManualNativePidWaits.set(terminal, controller); + const focusSubscription = window.onDidChangeActiveTerminal(changed => { + if (changed !== terminal) { controller.abort(); } + }); + let terminalPid: number | undefined; + try { + if (!isCurrent()) { return; } + const pidWait = await nativeTerminalPidWaits.waitFor(terminal, () => terminal.processId, { + signal: controller.signal, + isEligible: () => isNativeTerminalEligible(terminal), + }); + if (pidWait.kind === 'invalidated' || !pidWait.isCurrent() || !isCurrent()) { return; } + nativePidWaitIsCurrent = pidWait.isCurrent; + terminalPid = pidWait.value; + } finally { + focusSubscription.dispose(); + if (pendingManualNativePidWaits.get(terminal) === controller) { + pendingManualNativePidWaits.delete(terminal); + } + } if (!isCurrent()) { return; } const focused = focusOwner(); if (focused) { await focused; return; } @@ -2770,17 +2845,29 @@ interface NativeTerminalCandidate { isCurrent: () => boolean; } -async function findLocalTerminal(rPid: string): Promise { +function isNativeTerminalEligible(terminal: vscode.Terminal): boolean { + return isLiveTerminal(terminal) && !terminalIntegrations.has(terminal); +} + +async function findLocalTerminal(rPid: string, signal?: AbortSignal): Promise { const candidates = new Map(); for (const terminal of window.terminals) { - if (terminalIntegrations.has(terminal)) { continue; } + if (signal?.aborted) { return undefined; } + if (!isNativeTerminalEligible(terminal)) { continue; } const isCurrent = terminalRegistry.snapshot(terminal); - const terminalPid = await terminal.processId; - if (terminalPid === undefined || !isCurrent() || terminalIntegrations.has(terminal)) { continue; } + const pidWait = await nativeTerminalPidWaits.waitFor(terminal, () => terminal.processId, { + signal, isEligible: () => isNativeTerminalEligible(terminal), + }); + if (pidWait.kind === 'invalidated') { + if (pidWait.reason !== 'terminal') { return undefined; } + continue; + } + const terminalPid = pidWait.value; + if (terminalPid === undefined || !pidWait.isCurrent() || !isCurrent()) { continue; } const candidate = { terminal, pid: String(terminalPid), - isCurrent: () => isCurrent() && !terminalIntegrations.has(terminal), + isCurrent: () => pidWait.isCurrent() && isCurrent() && isNativeTerminalEligible(terminal), }; if (candidate.pid === rPid) { return candidate; } candidates.set(terminalPid, candidate); @@ -2875,11 +2962,27 @@ async function handleNotification(message: Record, socket: IpcS } const rPid = typeof params.pid === 'string' || typeof params.pid === 'number' ? String(params.pid) : ''; - pendingSessionAttachments.set(sessionId, socket); + pendingSessionAttachments.get(sessionId)?.controller.abort(); + const pendingAttachment: PendingSessionAttachment = { socket, controller: new AbortController() }; + pendingSessionAttachments.set(sessionId, pendingAttachment); const selectionRevision = sessionSelectionRevision; const selectedTerminal = window.activeTerminal; - const candidate = autoActivate && rPid && isLocalHost(host) ? await findLocalTerminal(rPid) : undefined; + const discardPendingAttachment = (): void => { + if (pendingSessionAttachments.get(sessionId) !== pendingAttachment) { return; } + pendingSessionAttachments.delete(sessionId); + pendingAttachment.controller.abort(); + if (!socket._sessionId) { socket.destroy(); } + }; + let candidate: NativeTerminalCandidate | undefined; + try { + candidate = autoActivate && rPid && isLocalHost(host) + ? await findLocalTerminal(rPid, pendingAttachment.controller.signal) : undefined; + } catch (error) { + discardPendingAttachment(); + throw error; + } if (!isLiveSessionSocket(socket)) { + discardPendingAttachment(); return; } // Another attach notification on this socket may have completed while @@ -2892,8 +2995,11 @@ async function handleNotification(message: Record, socket: IpcS } return; } - if (pendingSessionAttachments.get(sessionId) !== socket) { - socket.destroy(); + const currentPendingAttachment = pendingSessionAttachments.get(sessionId); + if (currentPendingAttachment !== pendingAttachment) { + // A newer handshake can reuse the same socket. Only destroy a + // different obsolete socket; never close the newer handshake's socket. + if (currentPendingAttachment?.socket !== socket) { socket.destroy(); } return; } pendingSessionAttachments.delete(sessionId); diff --git a/src/test/node/nativeTerminalPidWait.test.ts b/src/test/node/nativeTerminalPidWait.test.ts new file mode 100644 index 000000000..b6de39fa5 --- /dev/null +++ b/src/test/node/nativeTerminalPidWait.test.ts @@ -0,0 +1,170 @@ +import * as assert from 'assert'; +import { NativeTerminalPidWaitRegistry } from '../../nativeTerminalPidWait'; + +interface Terminal { id: string } + +function deferred() { + let resolve!: (value: Value) => void; + let reject!: (error: Error) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; +} + +function trackedSignal() { + let listener: (() => void) | undefined; + let removals = 0; + const signal = { + aborted: false, + addEventListener: (_type: string, callback: EventListenerOrEventListenerObject) => { + listener = typeof callback === 'function' ? () => callback(new Event('abort')) : () => callback.handleEvent(new Event('abort')); + }, + removeEventListener: () => { removals++; listener = undefined; }, + } as unknown as AbortSignal; + return { + signal, + get removals() { return removals; }, + abort() { + (signal as { aborted: boolean }).aborted = true; + listener?.(); + }, + }; +} + +suite('Native terminal PID wait registry', () => { + let registry: NativeTerminalPidWaitRegistry; + let first: Terminal; + let second: Terminal; + + setup(() => { + registry = new NativeTerminalPidWaitRegistry(); + first = { id: 'first' }; + second = { id: 'second' }; + }); + + test('undefined is a successful PID read, distinct from invalidation', async () => { + const result = await registry.waitFor(first, () => undefined); + assert.deepStrictEqual(result.kind, 'resolved'); + if (result.kind === 'resolved') { + assert.strictEqual(result.value, undefined); + assert.strictEqual(result.isCurrent(), true); + } + }); + + test('resolved snapshots become stale after terminal or lifecycle invalidation', async () => { + const terminalRead = await registry.waitFor(first, () => 48); + assert.strictEqual(terminalRead.kind, 'resolved'); + if (terminalRead.kind === 'resolved') { + assert.strictEqual(terminalRead.isCurrent(), true); + registry.invalidate(first); + assert.strictEqual(terminalRead.isCurrent(), false); + } + const freshTerminalRead = await registry.waitFor(first, () => 49); + assert.strictEqual(freshTerminalRead.kind, 'resolved'); + if (freshTerminalRead.kind === 'resolved') { + assert.strictEqual(freshTerminalRead.isCurrent(), true); + if (terminalRead.kind === 'resolved') { assert.strictEqual(terminalRead.isCurrent(), false); } + } + + const lifecycleRead = await registry.waitFor(second, () => 50); + assert.strictEqual(lifecycleRead.kind, 'resolved'); + registry.invalidateAll(); + if (lifecycleRead.kind === 'resolved') { assert.strictEqual(lifecycleRead.isCurrent(), false); } + const freshLifecycleRead = await registry.waitFor(second, () => 51); + assert.strictEqual(freshLifecycleRead.kind, 'resolved'); + if (freshLifecycleRead.kind === 'resolved') { assert.strictEqual(freshLifecycleRead.isCurrent(), true); } + }); + + test('registration during the processId getter permanently invalidates that read', async () => { + const pending = deferred(); + const resultPromise = registry.waitFor(first, () => { + registry.invalidate(first); + return pending.promise; + }); + const result = await resultPromise; + assert.deepStrictEqual(result, { kind: 'invalidated', reason: 'terminal' }); + pending.resolve(42); + await new Promise(resolve => setImmediate(resolve)); + + const fresh = await registry.waitFor(first, () => undefined); + assert.strictEqual(fresh.kind, 'resolved', 'disposing registration cannot revive an old operation, but a fresh read can run'); + }); + + test('a late rejection after invalidation is observed without unhandled rejection', async () => { + const pending = deferred(); + const resultPromise = registry.waitFor(first, () => pending.promise); + registry.invalidate(first); + assert.deepStrictEqual(await resultPromise, { kind: 'invalidated', reason: 'terminal' }); + pending.reject(new Error('late processId failure')); + await new Promise(resolve => setImmediate(resolve)); + }); + + test('lifecycle invalidation affects old waits but not a fresh lifecycle read', async () => { + const oldPending = deferred(); + const oldWait = registry.waitFor(first, () => oldPending.promise); + registry.invalidateAll(); + assert.deepStrictEqual(await oldWait, { kind: 'invalidated', reason: 'lifecycle' }); + const freshWait = await registry.waitFor(first, () => 43); + assert.deepStrictEqual(freshWait.kind, 'resolved'); + if (freshWait.kind === 'resolved') { assert.strictEqual(freshWait.value, 43); } + }); + + test('terminal invalidation is scoped to its terminal', async () => { + const otherPending = deferred(); + const firstPending = deferred(); + const otherWait = registry.waitFor(second, () => otherPending.promise); + const firstWait = registry.waitFor(first, () => firstPending.promise); + registry.invalidate(first); + assert.deepStrictEqual(await firstWait, { kind: 'invalidated', reason: 'terminal' }); + otherPending.resolve(44); + const otherResult = await otherWait; + assert.strictEqual(otherResult.kind, 'resolved'); + if (otherResult.kind === 'resolved') { assert.strictEqual(otherResult.value, 44); } + }); + + test('AbortSignal invalidation cleans up its listener', async () => { + const pending = deferred(); + const tracked = trackedSignal(); + const wait = registry.waitFor(first, () => pending.promise, { signal: tracked.signal }); + tracked.abort(); + assert.deepStrictEqual(await wait, { kind: 'invalidated', reason: 'operation' }); + assert.strictEqual(tracked.removals, 1); + pending.resolve(45); + }); + + test('normal completion and read or eligibility errors clean up AbortSignal listeners', async () => { + const completedSignal = trackedSignal(); + const complete = await registry.waitFor(first, () => 46, { signal: completedSignal.signal }); + assert.strictEqual(complete.kind, 'resolved'); + assert.strictEqual(completedSignal.removals, 1); + + const readSignal = trackedSignal(); + await assert.rejects(registry.waitFor(first, () => { throw new Error('read failed'); }, { signal: readSignal.signal }), /read failed/); + assert.strictEqual(readSignal.removals, 1); + + const eligibilitySignal = trackedSignal(); + await assert.rejects(registry.waitFor(first, () => 47, { + signal: eligibilitySignal.signal, isEligible: () => { throw new Error('eligibility failed'); }, + }), /eligibility failed/); + assert.strictEqual(eligibilitySignal.removals, 1); + }); + + test('eligibility failure after the PID read starts rejects and removes its listener', async () => { + const pending = deferred(); + const tracked = trackedSignal(); + let eligibilityFailure = false; + const wait = registry.waitFor(first, () => pending.promise, { + signal: tracked.signal, + isEligible: () => { + if (eligibilityFailure) { throw new Error('terminal eligibility changed'); } + return true; + }, + }); + eligibilityFailure = true; + pending.resolve(52); + await assert.rejects(wait, /terminal eligibility changed/); + assert.strictEqual(tracked.removals, 1); + }); +}); diff --git a/src/test/suite/session.test.ts b/src/test/suite/session.test.ts index 5648dc743..0172ecc6e 100644 --- a/src/test/suite/session.test.ts +++ b/src/test/suite/session.test.ts @@ -1148,6 +1148,7 @@ suite('Session Communication', () => { processId: Promise.resolve(terminalPid), creationOptions: { name: 'R Interactive', env: { SESS_DISCOVERY_FILE: discoveryFile } }, } as unknown as vscode.Terminal; + const liveTerminals = sandbox.stub(vscode.window, 'terminals').value([terminal]); let wrapperDiscoveryFile: string | undefined; try { await session.refreshTerminalDiscoveryFiles(newEndpoint, [terminal]); @@ -1169,6 +1170,7 @@ suite('Session Communication', () => { processId: Promise.resolve(wrapperTerminalPid), creationOptions: { name: 'R Interactive' }, } as unknown as vscode.Terminal; + liveTerminals.value([wrapperTerminal]); await session.refreshTerminalDiscoveryFiles(newEndpoint, [wrapperTerminal]); assert.deepStrictEqual(await readJson(wrapperDiscoveryFile), { version: 1, @@ -1197,6 +1199,7 @@ suite('Session Communication', () => { name: 'Renamed R', processId: Promise.resolve(45244), creationOptions: { name: 'R Interactive' }, } as unknown as vscode.Terminal; + sandbox.stub(vscode.window, 'terminals').value([terminal]); try { await session.updateSessionDiscoveryFile(file, 'old-sess', 45244); jgdSocket = 'new-jgd-socket'; @@ -1233,6 +1236,7 @@ suite('Session Communication', () => { processId: Promise.resolve(terminalPid), creationOptions: { name: 'R Interactive' }, } as unknown as vscode.Terminal; + sandbox.stub(vscode.window, 'terminals').value([terminal]); await session.refreshTerminalDiscoveryFiles(newEndpoint, [terminal]); const olderDiscovery: unknown = await readJson(olderFile); diff --git a/src/test/suite/sessionTerminalIntegration.test.ts b/src/test/suite/sessionTerminalIntegration.test.ts index 788aedbf1..20df29bb7 100644 --- a/src/test/suite/sessionTerminalIntegration.test.ts +++ b/src/test/suite/sessionTerminalIntegration.test.ts @@ -1,9 +1,11 @@ import * as assert from 'assert'; import * as os from 'os'; import * as path from 'path'; +import * as fsp from 'node:fs/promises'; import * as sinon from 'sinon'; import * as vscode from 'vscode'; import * as extension from '../../extension'; +import { readJson } from '../../fileSystem'; import * as rTerminal from '../../rTerminal'; import * as session from '../../session'; import * as util from '../../util'; @@ -142,6 +144,119 @@ suite('External Terminal Integration', () => { } }); + test('registering a scanned terminal releases discovery and lets the scan continue', async () => { + let pidRead!: () => void; + const readStarted = new Promise(resolve => { pidRead = resolve; }); + const externalSendText = sandbox.stub(); + const externalShow = sandbox.stub(); + const external = { + name: 'External R Console', + get processId() { pidRead(); return new Promise(() => undefined); }, + sendText: externalSendText, show: externalShow, dispose: sandbox.stub(), + } as unknown as vscode.Terminal; + terminalSpies.set(external, { sendText: externalSendText, show: externalShow }); + const native = makeTerminal(Promise.resolve(55109)); + setTerminals([external, native], native); + const socketPromise = connections.attach('native-scan-register-race', 55109, os.hostname()); + await readStarted; + const registration = session.registerTerminalIntegration(external, { attach: () => Promise.resolve(false) }); + try { + const socket = await socketPromise; + assert.strictEqual(socket._terminalPid, 55109); + assert.strictEqual(session.activeSession?.sessionId, 'native-scan-register-race'); + } finally { + registration.dispose(); + await session.cleanupSession('native-scan-register-race'); + } + }); + + test('discovery refresh skips a newly registered terminal and continues with live native terminals', async () => { + let pidRead!: () => void; + const readStarted = new Promise(resolve => { pidRead = resolve; }); + const external = { + name: 'External R Console', + get processId() { pidRead(); return new Promise(() => undefined); }, + sendText: sandbox.stub(), show: sandbox.stub(), dispose: sandbox.stub(), + } as unknown as vscode.Terminal; + const discoveryFile = await session.createSessionDiscoveryFile('old-sess'); + await session.updateSessionDiscoveryFile(discoveryFile, 'old-sess', 55112); + const native = { + name: 'Native R', processId: Promise.resolve(55112), + creationOptions: { name: 'R Interactive', env: { SESS_DISCOVERY_FILE: discoveryFile } }, + sendText: sandbox.stub(), show: sandbox.stub(), + } as unknown as vscode.Terminal; + setTerminals([external, native], native); + const refresh = session.refreshTerminalDiscoveryFiles('new-sess', [external, native]); + await readStarted; + const registration = session.registerTerminalIntegration(external, { attach: () => Promise.resolve(false) }); + try { + await refresh; + assert.strictEqual((await readJson(discoveryFile) as { endpoint?: string }).endpoint, 'new-sess'); + } finally { + registration.dispose(); + await fsp.rm(discoveryFile, { force: true }); + } + }); + + test('terminal discovery update cannot publish after its PID lookup is invalidated', async () => { + let pidRead!: () => void; + const readStarted = new Promise(resolve => { pidRead = resolve; }); + const discoveryFile = await session.createSessionDiscoveryFile('old-sess'); + await session.updateSessionDiscoveryFile(discoveryFile, 'old-sess', 55113); + const external = { + name: 'External R Console', + get processId() { pidRead(); return new Promise(() => undefined); }, + creationOptions: { name: 'External R Console', env: { SESS_DISCOVERY_FILE: discoveryFile } }, + sendText: sandbox.stub(), show: sandbox.stub(), dispose: sandbox.stub(), + } as unknown as vscode.Terminal; + setTerminals([external]); + const update = session.updateTerminalDiscovery(external); + await readStarted; + const registration = session.registerTerminalIntegration(external, { attach: () => Promise.resolve(false) }); + try { + await update; + assert.strictEqual((await readJson(discoveryFile) as { endpoint?: string }).endpoint, 'old-sess'); + } finally { + registration.dispose(); + await fsp.rm(discoveryFile, { force: true }); + } + }); + + test('a superseded socket is closed after discovery while the new same-ID connection commits', async () => { + let resolveFirstPid!: (pid: number | undefined) => void; + let pidRead!: () => void; + const firstRead = new Promise(resolve => { pidRead = resolve; }); + let reads = 0; + const sendText = sandbox.stub(); + const show = sandbox.stub(); + const terminal = { + name: 'Native R', + get processId() { + if (reads++ === 0) { pidRead(); return new Promise(resolve => { resolveFirstPid = resolve; }); } + return Promise.resolve(55111); + }, + sendText, show, + } as unknown as vscode.Terminal; + terminalSpies.set(terminal, { sendText, show }); + setTerminals([terminal]); + const oldSocket = await connections.startAttach('same-id-discovery-replacement', 55111, os.hostname()); + await firstRead; + const currentSocket = await connections.startAttach('same-id-discovery-replacement', 55111, os.hostname()); + try { + const reference = await waitForValue(() => session.getSessionReference('same-id-discovery-replacement')); + await waitForValue(() => oldSocket.destroyed ? true : undefined); + assert.strictEqual(currentSocket.destroyed, false); + assert.strictEqual(session.getSessionReference('same-id-discovery-replacement')?.connectionId, reference.connectionId); + resolveFirstPid(55111); + await new Promise(resolve => setImmediate(resolve)); + assert.strictEqual(currentSocket.destroyed, false, 'late old discovery must not close the committed socket'); + assert.strictEqual(session.getSessionReference('same-id-discovery-replacement')?.connectionId, reference.connectionId); + } finally { + resolveFirstPid(55111); + await session.cleanupSession('same-id-discovery-replacement'); + } + }); + test('registering the active terminal during native PID discovery cancels the native path', async () => { let resolvePid!: (pid: number | undefined) => void; let pidRead!: () => void; @@ -160,9 +275,12 @@ suite('External Terminal Integration', () => { await readStarted; const registration = session.registerTerminalIntegration(terminal, { attach: () => Promise.resolve(false) }); try { + // Registration invalidates the pending PID lookup synchronously; the + // command must finish even though VS Code never resolves processId. + await nativeOperation; await session.activateRSession(); resolvePid(undefined); - await nativeOperation; + await new Promise(resolve => setImmediate(resolve)); sinon.assert.notCalled(terminalSpies.get(terminal)!.sendText); sinon.assert.notCalled(createRTerm); } finally { @@ -171,6 +289,64 @@ suite('External Terminal Integration', () => { } }); + test('disposing an integration cannot revive an older native PID wait', async () => { + let resolvePid!: (pid: number | undefined) => void; + let pidRead!: () => void; + const readStarted = new Promise(resolve => { pidRead = resolve; }); + const sendText = sandbox.stub(); + const show = sandbox.stub(); + const terminal = { + name: 'Native R', + get processId() { pidRead(); return new Promise(resolve => { resolvePid = resolve; }); }, + sendText, show, + } as unknown as vscode.Terminal; + terminalSpies.set(terminal, { sendText, show }); + setTerminals([terminal]); + const createRTerm = sandbox.stub(rTerminal, 'createRTerm').resolves(true); + const operation = session.activateRSession(); + await readStarted; + const registration = session.registerTerminalIntegration(terminal, { attach: () => Promise.resolve(false) }); + registration.dispose(); + try { + await operation; + resolvePid(55110); + await new Promise(resolve => setImmediate(resolve)); + sinon.assert.notCalled(sendText); + sinon.assert.notCalled(createRTerm); + } finally { + resolvePid(undefined); + } + }); + + test('terminal cleanup and watcher shutdown release manual PID waits and observe late rejection', async () => { + for (const invalidation of ['terminal', 'lifecycle'] as const) { + let rejectPid!: (error: Error) => void; + let pidRead!: () => void; + const readStarted = new Promise(resolve => { pidRead = resolve; }); + const sendText = sandbox.stub(); + const show = sandbox.stub(); + const terminal = { + name: 'Native R', + get processId() { pidRead(); return new Promise((_resolve, reject) => { rejectPid = reject; }); }, + sendText, show, + } as unknown as vscode.Terminal; + terminalSpies.set(terminal, { sendText, show }); + setTerminals([terminal]); + const operation = session.activateRSession(); + await readStarted; + if (invalidation === 'terminal') { + session.cleanupTerminalBinding(terminal); + } else { + await session.shutdownSessionWatcher(); + } + await operation; + rejectPid(new Error('late processId rejection')); + await new Promise(resolve => setImmediate(resolve)); + sinon.assert.notCalled(sendText); + if (invalidation === 'lifecycle') { break; } + } + }); + test('registering a background terminal does not invalidate an unrelated native operation', async () => { let resolvePid!: (pid: number | undefined) => void; let pidRead!: () => void; From b846038ba26690e67729320e57875660ddbe4c6d Mon Sep 17 00:00:00 2001 From: eitsupi Date: Mon, 12 Oct 2026 11:59:41 +0000 Subject: [PATCH 6/6] test: keep manual recovery terminals in the live fixture --- src/test/suite/session.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/test/suite/session.test.ts b/src/test/suite/session.test.ts index 0172ecc6e..916f83a0e 100644 --- a/src/test/suite/session.test.ts +++ b/src/test/suite/session.test.ts @@ -1089,7 +1089,7 @@ suite('Session Communication', () => { name: 'Renamed R', processId: Promise.resolve(45241), creationOptions: { name: 'R Interactive' }, show: sandbox.spy(), sendText: sandbox.spy(), }; - sandbox.stub(vscode.window, 'terminals').value([first, second]); + const liveTerminals = sandbox.stub(vscode.window, 'terminals').value([first, second]); const activeTerminal = sandbox.stub(vscode.window, 'activeTerminal').value(second); sandbox.stub(util, 'config').returns({ get: (key: string) => key === 'sessionWatcher' ? true : undefined, @@ -1125,7 +1125,9 @@ suite('Session Communication', () => { // The R Interactive name alone must not grant ownership of another terminal. first.show.resetHistory(); - activeTerminal.value({ ...second, name: 'R Interactive', processId: Promise.resolve(45242) }); + const sameLabelTerminal = { ...second, name: 'R Interactive', processId: Promise.resolve(45242) }; + liveTerminals.value([first, second, sameLabelTerminal]); + activeTerminal.value(sameLabelTerminal); await session.activateRSession(); sinon.assert.calledOnce(first.show); sinon.assert.calledOnce(second.sendText);