diff --git a/src/client/common/pipes/namedPipes.ts b/src/client/common/pipes/namedPipes.ts
index 9bffe78f2b9f..ccad22da1249 100644
--- a/src/client/common/pipes/namedPipes.ts
+++ b/src/client/common/pipes/namedPipes.ts
@@ -7,6 +7,7 @@ import * as fs from 'fs-extra';
 import * as net from 'net';
 import * as os from 'os';
 import * as path from 'path';
+import { PassThrough } from 'stream';
 import * as rpc from 'vscode-jsonrpc/node';
 import { CancellationError, CancellationToken, Disposable } from 'vscode';
 import { traceVerbose } from '../../logging';
@@ -15,6 +16,9 @@ import { createDeferred } from '../utils/async';
 import { noop } from '../utils/misc';
 
 const { XDG_RUNTIME_DIR } = process.env;
+const FIFO_READ_RETRY_DELAY_MS = 10;
+const FIFO_READ_BUFFER_SIZE = 64 * 1024;
+
 export function generateRandomPipeName(prefix: string): string {
     // length of 10 picked because of the name length restriction for sockets
     const randomSuffix = crypto.randomBytes(10).toString('hex');
@@ -149,6 +153,111 @@ class CombinedReader implements rpc.MessageReader {
     }
 }
 
+// A net.Socket does not surface EOF for a nonblocking FIFO descriptor, so read
+// the descriptor directly and close after observing its writer disconnect.
+class FifoMessageReader extends rpc.AbstractMessageReader {
+    private readonly stream = new PassThrough();
+
+    private readonly messageReader = new rpc.StreamMessageReader(this.stream, 'utf-8');
+
+    private readonly buffer = Buffer.allocUnsafe(FIFO_READ_BUFFER_SIZE);
+
+    private readonly disposables: rpc.Disposable[] = [];
+
+    private hasReadData = false;
+
+    private hasObservedWriter = false;
+
+    private disposed = false;
+
+    private closePending = false;
+
+    constructor(
+        private readonly pipeName: string,
+        private readonly fd: number,
+        private readonly token?: CancellationToken,
+    ) {
+        super();
+        this.disposables.push(
+            this.messageReader.onError((error) => this.fireError(error)),
+            this.messageReader.onPartialMessage((info) => this.firePartialMessage(info)),
+        );
+    }
+
+    listen(callback: rpc.DataCallback): rpc.Disposable {
+        const listener = this.messageReader.listen(callback);
+        void this.read();
+        return listener;
+    }
+
+    private async read(): Promise<void> {
+        while (!this.disposed && !this.closePending) {
+            try {
+                const { bytesRead } = await fs.read(this.fd, this.buffer, 0, this.buffer.length, null);
+                if (bytesRead > 0) {
+                    this.hasReadData = true;
+                    this.stream.write(Buffer.from(this.buffer.subarray(0, bytesRead)));
+                    continue;
+                }
+
+                if (this.hasReadData || this.hasObservedWriter || this.token?.isCancellationRequested) {
+                    this.complete();
+                    return;
+                }
+            } catch (error) {
+                if (this.isRetryableReadError(error)) {
+                    this.hasObservedWriter = true;
+                } else {
+                    if (!this.disposed) {
+                        this.fireError(error);
+                        this.complete();
+                    }
+                    return;
+                }
+            }
+
+            await new Promise<void>((resolve) => {
+                setTimeout(resolve, FIFO_READ_RETRY_DELAY_MS);
+            });
+        }
+    }
+
+    private isRetryableReadError(error: unknown): boolean {
+        if (!(error instanceof Error) || !('code' in error)) {
+            return false;
+        }
+        const { code } = error as NodeJS.ErrnoException;
+        return code === 'EAGAIN' || code === 'EWOULDBLOCK';
+    }
+
+    private complete(): void {
+        if (this.disposed || this.closePending) {
+            return;
+        }
+        this.closePending = true;
+        this.stream.end();
+        setImmediate(() => {
+            if (!this.disposed) {
+                this.fireClose();
+            }
+        });
+    }
+
+    dispose(): void {
+        if (this.disposed) {
+            return;
+        }
+        this.disposed = true;
+        this.stream.destroy();
+        this.messageReader.dispose();
+        this.disposables.forEach((disposable) => disposable.dispose());
+        this.disposables.length = 0;
+        fs.close(this.fd).catch(noop);
+        fs.unlink(this.pipeName).catch(noop);
+        super.dispose();
+    }
+}
+
 export async function createReaderPipe(pipeName: string, token?: CancellationToken): Promise<rpc.MessageReader> {
     if (isWindows()) {
         // windows implementation of FIFO using named pipes
@@ -189,12 +298,5 @@ export async function createReaderPipe(pipeName: string, token?: CancellationTok
         // Intentionally ignored
     }
     const fd = await fs.open(pipeName, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK);
-    const socket = new net.Socket({ fd });
-    const reader = new rpc.SocketMessageReader(socket, 'utf-8');
-    socket.on('close', () => {
-        fs.close(fd).catch(noop);
-        reader.dispose();
-    });
-
-    return reader;
+    return new FifoMessageReader(pipeName, fd, token);
 }
diff --git a/src/client/testing/testController/common/utils.ts b/src/client/testing/testController/common/utils.ts
index 583d57648548..ecba099968ab 100644
--- a/src/client/testing/testController/common/utils.ts
+++ b/src/client/testing/testController/common/utils.ts
@@ -115,7 +115,7 @@ export async function startRunResultNamedPipe(
     dataReceivedCallback: (payload: ExecutionTestPayload) => void,
     deferredTillServerClose: Deferred<void>,
     cancellationToken?: CancellationToken,
-): Promise<string> {
+): Promise<{ name: string; dispose: () => void }> {
     traceVerbose('Starting Test Result named pipe');
     const pipeName: string = generateRandomPipeName('python-test-results');
 
@@ -147,7 +147,10 @@ export async function startRunResultNamedPipe(
         }),
     );
 
-    return pipeName;
+    return {
+        name: pipeName,
+        dispose: () => disposable.dispose(),
+    };
 }
 
 interface DiscoveryResultMessage extends Message {
diff --git a/src/client/testing/testController/pytest/pytestExecutionAdapter.ts b/src/client/testing/testController/pytest/pytestExecutionAdapter.ts
index a982037d90e5..ea60297b1ffe 100644
--- a/src/client/testing/testController/pytest/pytestExecutionAdapter.ts
+++ b/src/client/testing/testController/pytest/pytestExecutionAdapter.ts
@@ -53,7 +53,7 @@ export class PytestTestExecutionAdapter implements ITestExecutionAdapter {
         const cSource = new CancellationTokenSource();
         runInstance.token.onCancellationRequested(() => cSource.cancel());
 
-        const name = await utils.startRunResultNamedPipe(
+        const resultPipe = await utils.startRunResultNamedPipe(
             dataReceivedCallback, // callback to handle data received
             deferredTillServerClose, // deferred to resolve when server closes
             cSource.token, // token to cancel
@@ -66,7 +66,7 @@ export class PytestTestExecutionAdapter implements ITestExecutionAdapter {
             await this.runTestsNew(
                 uri,
                 testIds,
-                name,
+                resultPipe.name,
                 cSource,
                 runInstance,
                 profileKind,
@@ -77,6 +77,7 @@ export class PytestTestExecutionAdapter implements ITestExecutionAdapter {
             );
         } finally {
             await utils.awaitDeferredWithTimeout(deferredTillServerClose, utils.RESULT_PIPE_DRAIN_TIMEOUT_MS);
+            resultPipe.dispose();
         }
     }
 
diff --git a/src/client/testing/testController/unittest/testExecutionAdapter.ts b/src/client/testing/testController/unittest/testExecutionAdapter.ts
index 755da2e3c65a..2061a6e948a6 100644
--- a/src/client/testing/testController/unittest/testExecutionAdapter.ts
+++ b/src/client/testing/testController/unittest/testExecutionAdapter.ts
@@ -64,7 +64,7 @@ export class UnittestTestExecutionAdapter implements ITestExecutionAdapter {
         };
         const cSource = new CancellationTokenSource();
         runInstance.token.onCancellationRequested(() => cSource.cancel());
-        const name = await utils.startRunResultNamedPipe(
+        const resultPipe = await utils.startRunResultNamedPipe(
             dataReceivedCallback, // callback to handle data received
             deferredTillServerClose, // deferred to resolve when server closes
             cSource.token, // token to cancel
@@ -79,7 +79,7 @@ export class UnittestTestExecutionAdapter implements ITestExecutionAdapter {
             await this.runTestsNew(
                 uri,
                 testIds,
-                name,
+                resultPipe.name,
                 cSource,
                 runInstance,
                 profileKind,
@@ -91,6 +91,7 @@ export class UnittestTestExecutionAdapter implements ITestExecutionAdapter {
             traceError(`Error in running unittest tests: ${error}`);
         } finally {
             await utils.awaitDeferredWithTimeout(deferredTillServerClose, utils.RESULT_PIPE_DRAIN_TIMEOUT_MS);
+            resultPipe.dispose();
         }
     }
 
diff --git a/src/test/common/pipes/namedPipes.unit.test.ts b/src/test/common/pipes/namedPipes.unit.test.ts
new file mode 100644
index 000000000000..981107bdfce7
--- /dev/null
+++ b/src/test/common/pipes/namedPipes.unit.test.ts
@@ -0,0 +1,119 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+import * as assert from 'assert';
+import * as fs from 'fs';
+import { CancellationTokenSource } from 'vscode';
+import * as rpc from 'vscode-jsonrpc/node';
+import { createDeferred } from '../../../client/common/utils/async';
+import { createReaderPipe, generateRandomPipeName } from '../../../client/common/pipes/namedPipes';
+
+const TEST_TIMEOUT_MS = 2_000;
+
+async function waitFor<T>(promise: Promise<T>, description: string): Promise<T> {
+    let timeout: NodeJS.Timeout | undefined;
+    try {
+        return await Promise.race([
+            promise,
+            new Promise<T>((_, reject) => {
+                timeout = setTimeout(() => reject(new Error(`Timed out waiting for ${description}`)), TEST_TIMEOUT_MS);
+            }),
+        ]);
+    } finally {
+        if (timeout) {
+            clearTimeout(timeout);
+        }
+    }
+}
+
+suite('POSIX named pipe reader', () => {
+    test('delivers buffered messages and closes when the FIFO writer closes', async function () {
+        if (process.platform === 'win32') {
+            this.skip();
+        }
+
+        const pipeName = generateRandomPipeName('python-test-fifo-reader');
+        const reader = await createReaderPipe(pipeName);
+        const received: rpc.Message[] = [];
+        const closed = createDeferred<void>();
+        const listener = reader.listen((message) => received.push(message));
+        const closeListener = reader.onClose(() => closed.resolve());
+        const stream = fs.createWriteStream(pipeName);
+        const writer = new rpc.StreamMessageWriter(stream, 'utf-8');
+        const expected: rpc.NotificationMessage[] = Array.from({ length: 20 }, (_, value) => ({
+            jsonrpc: '2.0',
+            method: 'test/message',
+            params: { value, data: 'x'.repeat(4_096) },
+        }));
+
+        try {
+            for (const message of expected) {
+                await writer.write(message);
+            }
+            writer.end();
+
+            await waitFor(closed.promise, 'the FIFO reader to close');
+
+            assert.deepStrictEqual(received, expected);
+        } finally {
+            listener.dispose();
+            closeListener.dispose();
+            reader.dispose();
+            writer.dispose();
+            await fs.promises.rm(pipeName, { force: true });
+        }
+    });
+
+    test('closes when a connected FIFO writer sends no messages', async function () {
+        if (process.platform === 'win32') {
+            this.skip();
+        }
+
+        const pipeName = generateRandomPipeName('python-test-fifo-empty');
+        const reader = await createReaderPipe(pipeName);
+        const closed = createDeferred<void>();
+        const listener = reader.listen(() => undefined);
+        const closeListener = reader.onClose(() => closed.resolve());
+        const stream = fs.createWriteStream(pipeName);
+
+        try {
+            await new Promise<void>((resolve, reject) => {
+                stream.once('open', () => setTimeout(resolve, 50));
+                stream.once('error', reject);
+            });
+            stream.end();
+
+            await waitFor(closed.promise, 'the empty FIFO reader to close');
+        } finally {
+            listener.dispose();
+            closeListener.dispose();
+            reader.dispose();
+            stream.destroy();
+            await fs.promises.rm(pipeName, { force: true });
+        }
+    });
+
+    test('closes on cancellation when no FIFO writer connected', async function () {
+        if (process.platform === 'win32') {
+            this.skip();
+        }
+
+        const pipeName = generateRandomPipeName('python-test-fifo-cancel');
+        const cancellation = new CancellationTokenSource();
+        const reader = await createReaderPipe(pipeName, cancellation.token);
+        const closed = createDeferred<void>();
+        const listener = reader.listen(() => undefined);
+        const closeListener = reader.onClose(() => closed.resolve());
+
+        try {
+            cancellation.cancel();
+            await waitFor(closed.promise, 'the cancelled FIFO reader to close');
+        } finally {
+            listener.dispose();
+            closeListener.dispose();
+            reader.dispose();
+            cancellation.dispose();
+            await fs.promises.rm(pipeName, { force: true });
+        }
+    });
+});
diff --git a/src/test/testing/common/testingAdapter.test.ts b/src/test/testing/common/testingAdapter.test.ts
index d73efba2a8ec..dcc57c3d20ee 100644
--- a/src/test/testing/common/testingAdapter.test.ts
+++ b/src/test/testing/common/testingAdapter.test.ts
@@ -1,7 +1,7 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
-import { TestController, TestRun, TestRunProfileKind, Uri } from 'vscode';
+import { CancellationTokenSource, TestController, TestRun, TestRunProfileKind, Uri } from 'vscode';
 import * as typeMoq from 'typemoq';
 import * as path from 'path';
 import * as assert from 'assert';
@@ -694,6 +694,75 @@ suite('End to End Tests: test adapters', () => {
             });
     });
 
+    test('pytest cancellation closes the POSIX result pipe promptly', async () => {
+        if (process.platform === 'win32') {
+            return;
+        }
+
+        workspaceUri = Uri.parse(rootPathSmallWorkspace);
+        configService.getSettings(workspaceUri).testing.pytestArgs = [];
+
+        const cancellation = new CancellationTokenSource();
+        const completedTestId = `${rootPathSmallWorkspace}/test_cancellation.py::test_result_before_cancellation`;
+        const slowTestId = `${rootPathSmallWorkspace}/test_cancellation.py::test_waits_for_cancellation`;
+        const previousRuntimeDir = process.env.XDG_RUNTIME_DIR;
+        const runtimeDir = fs.mkdtempSync(path.join(os.tmpdir(), 'vscode-python-cancellation-'));
+        process.env.XDG_RUNTIME_DIR = runtimeDir;
+        let completedResultReceived = false;
+        let cancellationStartedAt: number | undefined;
+        let watchdogTriggered = false;
+        const watchdog = setTimeout(() => {
+            watchdogTriggered = true;
+            cancellation.cancel();
+        }, 60_000);
+
+        resultResolver = new PythonResultResolver(testController, pytestProvider, workspaceUri);
+        resultResolver.resolveExecution = (payload, _runInstance) => {
+            const completedResult = Object.values(payload.result ?? {}).find(
+                (result) => result.test === completedTestId && result.outcome === 'success',
+            );
+            if (completedResult && !cancellation.token.isCancellationRequested) {
+                completedResultReceived = true;
+                cancellationStartedAt = performance.now();
+                cancellation.cancel();
+            }
+        };
+
+        const testRun = typeMoq.Mock.ofType<TestRun>();
+        testRun.setup((t) => t.token).returns(() => cancellation.token);
+        testRun.setup((t) => t.appendOutput(typeMoq.It.isAny())).returns(() => false);
+
+        const executionAdapter = new PytestTestExecutionAdapter(configService, resultResolver, envVarsService);
+        try {
+            await executionAdapter.runTests(
+                workspaceUri,
+                [completedTestId, slowTestId],
+                TestRunProfileKind.Run,
+                testRun.object,
+                pythonExecFactory,
+            );
+        } finally {
+            clearTimeout(watchdog);
+            cancellation.cancel();
+            cancellation.dispose();
+            if (previousRuntimeDir === undefined) {
+                delete process.env.XDG_RUNTIME_DIR;
+            } else {
+                process.env.XDG_RUNTIME_DIR = previousRuntimeDir;
+            }
+            fs.rmSync(runtimeDir, { force: true, recursive: true });
+        }
+
+        assert.ok(!watchdogTriggered, 'Expected pytest to report a result before the cancellation watchdog fired');
+        assert.ok(completedResultReceived, 'Expected the completed test result before cancellation');
+        assert.ok(cancellationStartedAt, 'Expected cancellation to start after receiving a completed test result');
+        const cancellationDuration = performance.now() - cancellationStartedAt;
+        assert.ok(
+            cancellationDuration < 4_000,
+            `Expected cancellation to close the result pipe promptly, but it took ${cancellationDuration}ms`,
+        );
+    }).timeout(75_000);
+
     test('Unittest execution with coverage, small workspace', async () => {
         // result resolver and saved data for assertions
         resultResolver = new PythonResultResolver(testController, unittestProvider, workspaceUri);
diff --git a/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts b/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts
index 9b83f1bbd088..f7755be1b5f5 100644
--- a/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts
+++ b/src/test/testing/testController/pytest/pytestExecutionAdapter.unit.test.ts
@@ -89,7 +89,9 @@ suite('pytest test execution adapter', () => {
         myTestPath = path.join('/', 'my', 'test', 'path', '/');
 
         utilsStartRunResultNamedPipeStub = sinon.stub(util, 'startRunResultNamedPipe');
-        utilsStartRunResultNamedPipeStub.callsFake(() => Promise.resolve('runResultPipe-mockName'));
+        utilsStartRunResultNamedPipeStub.callsFake(() =>
+            Promise.resolve({ name: 'runResultPipe-mockName', dispose: sinon.stub() }),
+        );
 
         execService.setup((x) => x.getExecutablePath()).returns(() => Promise.resolve('/mock/path/to/python'));
     });
@@ -186,7 +188,7 @@ suite('pytest test execution adapter', () => {
         utilsWriteTestIdsFileStub.resolves('testIdPipe-mockName');
         utilsStartRunResultNamedPipeStub.callsFake((_callback, deferredTillServerClose, token) => {
             token?.onCancellationRequested(() => deferredTillServerClose.resolve());
-            return Promise.resolve('runResultPipe-mockName');
+            return Promise.resolve({ name: 'runResultPipe-mockName', dispose: sinon.stub() });
         });
         const cancellationToken = new CancellationTokenSource();
         const testRun = typeMoq.Mock.ofType<TestRun>();
diff --git a/src/test/testing/testController/testCancellationRunAdapters.unit.test.ts b/src/test/testing/testController/testCancellationRunAdapters.unit.test.ts
index cdf0d00c5dc4..899308f81b13 100644
--- a/src/test/testing/testController/testCancellationRunAdapters.unit.test.ts
+++ b/src/test/testing/testController/testCancellationRunAdapters.unit.test.ts
@@ -112,7 +112,7 @@ suite('Execution Flow Run Adapters', () => {
                     deferredTillServerCloseTester?.resolve();
                 });
 
-                return Promise.resolve('named-pipes-socket-name');
+                return Promise.resolve({ name: 'named-pipes-socket-name', dispose: serverDisposeStub });
             });
             serverDisposeStub.callsFake(() => {
                 console.log('server disposed');
@@ -138,6 +138,7 @@ suite('Execution Flow Run Adapters', () => {
             );
             // wait for server to start to keep test from failing
             await deferredStartTestIdsNamedPipe.promise;
+            sinon.assert.calledOnce(serverDisposeStub);
         });
         test(`Adapter ${adapter}: token called mid-debug resolves correctly`, async () => {
             // mock test run and cancelation token
@@ -180,7 +181,7 @@ suite('Execution Flow Run Adapters', () => {
                 token?.onCancellationRequested(() => {
                     deferredTillServerCloseTester?.resolve();
                 });
-                return Promise.resolve('named-pipes-socket-name');
+                return Promise.resolve({ name: 'named-pipes-socket-name', dispose: serverDisposeStub });
             });
             serverDisposeStub.callsFake(() => {
                 console.log('server disposed');
@@ -219,6 +220,7 @@ suite('Execution Flow Run Adapters', () => {
             );
             // wait for server to start to keep test from failing
             await deferredStartTestIdsNamedPipe.promise;
+            sinon.assert.calledOnce(serverDisposeStub);
         });
     });
 });
diff --git a/src/test/testing/testController/unittest/testExecutionAdapter.unit.test.ts b/src/test/testing/testController/unittest/testExecutionAdapter.unit.test.ts
index 8a86e9228567..a8d40bad477f 100644
--- a/src/test/testing/testController/unittest/testExecutionAdapter.unit.test.ts
+++ b/src/test/testing/testController/unittest/testExecutionAdapter.unit.test.ts
@@ -89,7 +89,9 @@ suite('Unittest test execution adapter', () => {
         myTestPath = path.join('/', 'my', 'test', 'path', '/');
 
         utilsStartRunResultNamedPipeStub = sinon.stub(util, 'startRunResultNamedPipe');
-        utilsStartRunResultNamedPipeStub.callsFake(() => Promise.resolve('runResultPipe-mockName'));
+        utilsStartRunResultNamedPipeStub.callsFake(() =>
+            Promise.resolve({ name: 'runResultPipe-mockName', dispose: sinon.stub() }),
+        );
 
         execService.setup((x) => x.getExecutablePath()).returns(() => Promise.resolve('/mock/path/to/python'));
     });
@@ -510,7 +512,7 @@ suite('Unittest test execution adapter', () => {
         let serverCloseDeferred: Deferred<void> | undefined;
         utilsStartRunResultNamedPipeStub.callsFake((_callback: unknown, deferred: Deferred<void>, _token: unknown) => {
             serverCloseDeferred = deferred;
-            return Promise.resolve('runResultPipe-mockName');
+            return Promise.resolve({ name: 'runResultPipe-mockName', dispose: sinon.stub() });
         });
 
         const projectPath = path.join('/', 'workspace', 'myproject');
diff --git a/src/testTestingRootWkspc/smallWorkspace/test_cancellation.py b/src/testTestingRootWkspc/smallWorkspace/test_cancellation.py
new file mode 100644
index 000000000000..ffa4ada51cfe
--- /dev/null
+++ b/src/testTestingRootWkspc/smallWorkspace/test_cancellation.py
@@ -0,0 +1,11 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import time
+
+
+def test_result_before_cancellation():
+    assert True
+
+
+def test_waits_for_cancellation():
+    time.sleep(120)