Cleanup Response (microsoft#2876)

Author: chrmarti

chat-lib/package-lock.json

@@ -16,9 +16,9 @@
 	"dependencies": {
 		"@microsoft/tiktokenizer": "^1.0.10",
 		"@sinclair/typebox": "^0.34.41",
-		"@vscode/copilot-api": "^0.2.5",
+		"@vscode/copilot-api": "^0.2.8",
 		"@vscode/l10n": "^0.0.18",
-		"@vscode/prompt-tsx": "^0.4.0-alpha.5",
+		"@vscode/prompt-tsx": "^0.4.0-alpha.6",
 		"@vscode/tree-sitter-wasm": "0.0.5-php.2",
 		"applicationinsights": "^2.9.7",
 		"jsonc-parser": "^3.3.1",
@@ -30,7 +30,7 @@
 		"yaml": "^2.8.0"
 	},
 	"devDependencies": {
-		"@anthropic-ai/sdk": "^0.68.0",
+		"@anthropic-ai/sdk": "^0.71.2",
 		"@octokit/types": "^14.1.0",
 		"@types/node": "^22.16.3",
 		"@types/vscode": "^1.102.0",

chat-lib/package.json

@@ -9,7 +9,6 @@ dotenv.config({ path: '../.env' });
 
 import { readFile } from 'fs/promises';
 import { join } from 'path';
-import * as stream from 'stream';
 import { assert, describe, expect, it } from 'vitest';
 import type { AuthenticationGetSessionOptions, AuthenticationSession, LanguageModelChat } from 'vscode';
 import { ResultType } from '../src/_internal/extension/completions-core/vscode-node/lib/src/ghostText/ghostText';
@@ -51,13 +50,12 @@ class TestFetcher implements IFetcher {
 		};
 
 		const found = typeof responseText === 'string';
-		return new Response(
+		const text = responseText || '';
+		return Response.fromText(
 			found ? 200 : 404,
 			found ? 'OK' : 'Not Found',
 			headers,
-			async () => responseText || '',
-			async () => JSON.parse(responseText || ''),
-			async () => stream.Readable.from([responseText || '']),
+			text,
 			'node-http'
 		);
 	}

chat-lib/test/getInlineCompletions.spec.ts

@@ -10,7 +10,6 @@ dotenv.config({ path: '../.env' });
 import { promises as fs } from 'fs';
 import { outdent } from 'outdent';
 import * as path from 'path';
-import * as stream from 'stream';
 import { assert, describe, expect, it } from 'vitest';
 import { CopilotToken, createTestExtendedTokenInfo } from '../src/_internal/platform/authentication/common/copilotToken';
 import { ICopilotTokenManager } from '../src/_internal/platform/authentication/common/copilotTokenManager';
@@ -51,13 +50,12 @@ class TestFetcher implements IFetcher {
 		};
 
 		const found = typeof responseText === 'string';
-		return new Response(
+		const text = responseText || '';
+		return Response.fromText(
 			found ? 200 : 404,
 			found ? 'OK' : 'Not Found',
 			headers,
-			async () => responseText || '',
-			async () => JSON.parse(responseText || ''),
-			async () => stream.Readable.from([responseText || '']),
+			text,
 			'node-http'
 		);
 	}

chat-lib/test/nesProvider.spec.ts

@@ -7,7 +7,6 @@ import { MessageParam } from '@anthropic-ai/sdk/resources';
 import { RequestMetadata, RequestType } from '@vscode/copilot-api';
 import { Raw } from '@vscode/prompt-tsx';
 import * as http from 'http';
-import { ClientHttp2Stream } from 'http2';
 import { IChatMLFetcher, Source } from '../../../../platform/chat/common/chatMLFetcher';
 import { ChatLocation, ChatResponse } from '../../../../platform/chat/common/commonTypes';
 import { CustomModel, EndpointEditToolName, IEndpointProvider } from '../../../../platform/endpoint/common/endpointProvider';
@@ -528,7 +527,7 @@ class ClaudeStreamingPassThroughEndpoint implements IChatEndpoint {
 		telemetryData: TelemetryData,
 		cancellationToken?: CancellationToken
 	): Promise<AsyncIterableObject<ChatCompletion>> {
-		const body = (await response.body()) as ClientHttp2Stream;
+		const body = response.body;
 		return new AsyncIterableObject<ChatCompletion>(async feed => {
 			// We parse the stream just to return a correct ChatCompletion for logging the response and token usage details.
 			const requestId = response.headers.get('X-Request-ID') ?? generateUuid();
@@ -566,9 +565,7 @@ class ClaudeStreamingPassThroughEndpoint implements IChatEndpoint {
 					parser.feed(chunk);
 				}
 			} finally {
-				if (!body.destroyed) {
-					body.destroy();
-				}
+				await body.destroy();
 			}
 		});
 	}

src/extension/agents/claude/node/claudeLanguageModelServer.ts

@@ -3,7 +3,6 @@
  *  Licensed under the MIT License. See License.txt in the project root for license information.
  *--------------------------------------------------------------------------------------------*/
 
-import { ClientHttp2Stream } from 'http2';
 import { IAuthenticationService } from '../../../../../../platform/authentication/common/authentication';
 import { CopilotAnnotations, StreamCopilotAnnotations } from '../../../../../../platform/completions-core/common/openai/copilotAnnotations';
 import { IEnvService } from '../../../../../../platform/env/common/envService';
@@ -475,15 +474,10 @@ export class LiveOpenAIFetcher extends OpenAIFetcher {
 			return { type: 'canceled', reason: 'before fetch request' };
 		}
 		if (cancel?.isCancellationRequested) {
-			const body = await response.body();
 			try {
 				// Destroy the stream so that the server is hopefully notified we don't want any more data
 				// and can cancel/forget about the request itself.
-				if (body && typeof (body as ClientHttp2Stream).destroy === 'function') {
-					(body as ClientHttp2Stream).destroy();
-				} else if (body instanceof ReadableStream) {
-					void body.cancel();
-				}
+				await response.body.destroy();
 			} catch (e) {
 				this.instantiationService.invokeFunction(acc => logger.exception(acc, e, `Error destroying stream`));
 			}

src/extension/completions-core/vscode-node/lib/src/openai/fetch.ts

@@ -3,9 +3,9 @@
  *  Licensed under the MIT License. See License.txt in the project root for license information.
  *--------------------------------------------------------------------------------------------*/
 
-import { ClientHttp2Stream } from 'http2';
 import { CopilotAnnotation, CopilotAnnotations, CopilotNamedAnnotationList, StreamCopilotAnnotations } from '../../../../../../platform/completions-core/common/openai/copilotAnnotations';
 import { getRequestId, RequestId } from '../../../../../../platform/networking/common/fetch';
+import { DestroyableStream } from '../../../../../../platform/networking/common/fetcherService';
 import { IInstantiationService, ServicesAccessor } from '../../../../../../util/vs/platform/instantiation/common/instantiation';
 import { CancellationToken as ICancellationToken } from '../../../types/src';
 import { ICompletionsLogTargetService, Logger } from '../logger';
@@ -211,7 +211,7 @@ export class SSEProcessor {
 	private constructor(
 		private readonly expectedNumChoices: number,
 		private readonly response: Response,
-		private readonly body: NodeJS.ReadableStream,
+		private readonly body: DestroyableStream<string>,
 		private readonly telemetryData: TelemetryWithExp,
 		private readonly dropCompletionReasons: string[],
 		private readonly cancellationToken: ICancellationToken | undefined = undefined,
@@ -236,23 +236,7 @@ export class SSEProcessor {
 		const instantiationService = accessor.get(IInstantiationService);
 		const logTargetService = accessor.get(ICompletionsLogTargetService);
 
-		// Handle both NodeJS.ReadableStream and ReadableStream for web.  Once
-		// helix fetcher is removed, the NodeJS.ReadableStream support can be
-		let body = response.body() as unknown as NodeJS.ReadableStream;
-		if (body === null) {
-			throw new Error('No response body available');
-		}
-		// OLD
-		/* if (typeof body.setEncoding === 'function') {
-			body.setEncoding('utf8');
-		} else {
-			// Convert fetch response to utf-8 decoded stream
-			body = (body as unknown as ReadableStream).pipeThrough(new TextDecoderStream()) as unknown as NodeJS.ReadableStream;
-		} */
-
-		// NEW
-		body = await body as NodeJS.ReadableStream;
-		body.setEncoding('utf8');
+		const body = response.body.pipeThrough(new TextDecoderStream());
 
 		// TODO@benibenj can we switch to our SSEProcessor implementation?
 		// It seems like they build more on top of the shared impl
@@ -288,7 +272,7 @@ export class SSEProcessor {
 		try {
 			yield* this.processSSEInner(finishedCb);
 		} finally {
-			this.cancel();
+			await this.cancel();
 			streamChoicesLogger.debug(this.logTarget,
 				`request done: headerRequestId: [${this.requestId.headerRequestId}] model deployment ID: [${this.requestId.deploymentId}]`
 			);
@@ -307,7 +291,7 @@ export class SSEProcessor {
 
 		// Iterate over arbitrarily sized chunks coming in from the network.
 		networkRead: for await (const chunk of this.body) {
-			if (this.maybeCancel('after awaiting body chunk')) {
+			if (await this.maybeCancel('after awaiting body chunk')) {
 				return;
 			}
 
@@ -431,7 +415,7 @@ export class SSEProcessor {
 							})
 						);
 
-						if (this.maybeCancel('after awaiting finishedCb')) {
+						if (await this.maybeCancel('after awaiting finishedCb')) {
 							return;
 						}
 					}
@@ -486,7 +470,7 @@ export class SSEProcessor {
 						solution.yielded = true;
 					}
 
-					if (this.maybeCancel('after yielding finished choice')) {
+					if (await this.maybeCancel('after yielding finished choice')) {
 						return;
 					}
 
@@ -522,7 +506,7 @@ export class SSEProcessor {
 				usage: usage,
 			};
 
-			if (this.maybeCancel('after yielding after iteration done')) {
+			if (await this.maybeCancel('after yielding after iteration done')) {
 				return;
 			}
 		}
@@ -605,7 +589,7 @@ export class SSEProcessor {
 				usage: usage,
 			};
 
-			if (this.maybeCancel('after yielding on DONE')) {
+			if (await this.maybeCancel('after yielding on DONE')) {
 				return;
 			}
 		}
@@ -615,22 +599,18 @@ export class SSEProcessor {
 	 * Returns whether the cancellation token was cancelled and closes the
 	 * stream if it was.
 	 */
-	private maybeCancel(description: string) {
+	private async maybeCancel(description: string) {
 		if (this.cancellationToken?.isCancellationRequested) {
 			streamChoicesLogger.debug(this.logTarget, 'Cancelled: ' + description);
-			this.cancel();
+			await this.cancel();
 			return true;
 		}
 		return false;
 	}
 
 	/** Cancels the network request to the proxy. */
-	private cancel() {
-		if (this.body && 'destroy' in this.body && typeof this.body.destroy === 'function') {
-			(this.body as ClientHttp2Stream).destroy();
-		} else if (this.body instanceof ReadableStream) {
-			void this.body.cancel();
-		}
+	private async cancel() {
+		await this.body.destroy();
 	}
 
 	/** Returns whether we've finished receiving all expected solutions. */

src/extension/completions-core/vscode-node/lib/src/openai/stream.ts

@@ -3,7 +3,6 @@
  *  Licensed under the MIT License. See License.txt in the project root for license information.
  *--------------------------------------------------------------------------------------------*/
 
-import { Readable } from 'stream';
 import { CopilotNamedAnnotationList } from '../../../../../../platform/completions-core/common/openai/copilotAnnotations';
 import { FetchOptions, IAbortController, ICompletionsFetcherService, IHeaders, Response } from '../networking';
 
@@ -15,13 +14,11 @@ export function createFakeResponse(statusCode: number, response?: string, header
 	for (const [key, value] of Object.entries(headers || {})) {
 		fakeHeaders.set(key, value);
 	}
-	return new Response(
+	return Response.fromText(
 		statusCode,
 		'status text',
 		fakeHeaders,
-		() => Promise.resolve(response ?? ''),
-		() => Promise.resolve(response ? JSON.parse(response) : {}),
-		() => Promise.resolve(null),
+		response ?? '',
 		'test-stub'
 	);
 }
@@ -37,13 +34,11 @@ export function createFakeJsonResponse(statusCode: number, response: string | ob
 }
 
 export function createFakeStreamResponse(body: string): Response {
-	return new Response(
+	return Response.fromText(
 		200,
 		'Success',
 		new FakeHeaders(),
-		() => Promise.resolve(body),
-		() => Promise.resolve(JSON.parse(body.replace(/^data: /gm, '').replace(/\n\[DONE\]\n$/, ''))),
-		() => Promise.resolve(toStream(body)),
+		body,
 		'test-stub'
 	);
 }
@@ -133,16 +128,6 @@ export class NoFetchFetcher extends FakeFetcher {
 	}
 }
 
-function toStream(...strings: string[]): NodeJS.ReadableStream {
-	const stream = new Readable();
-	stream._read = () => { };
-	for (const s of strings) {
-		stream.push(s);
-	}
-	stream.push(null);
-	return stream;
-}
-
 class FakeHeaders implements IHeaders {
 	private readonly headers: Map<string, string> = new Map();