[http-client-csharp] Add JSONL and SSE streaming support (#11462)

## Summary

- preserve JSONL and SSE stream metadata in the C# emitter code model
- generate incremental `IAsyncEnumerable<T>` JSONL requests and
`AsyncStreamingClientResult<T>` JSONL responses
- generate unbuffered SSE responses as
`AsyncStreamingClientResult<SseItem<BinaryData>>`, including
terminal-event suppression
- add live Spector coverage for JSONL send/receive and all unnamed,
named, and retrieval SSE scenarios
- consume the streaming result APIs introduced by
Azure/azure-sdk-for-net#60994

Closes #10738

## Validation

- `npm run build`
- `npm test`
- `npm run cop`
- `Test-Spector.ps1 -filter "http/streaming/jsonl"`
- `Test-Spector.ps1 -filter "http/streaming/sse"`

`npm run lint -- --emitter` is currently blocked by the package
resolving ESLint 10 while retaining a legacy ESLint configuration
(`eslint.config.*` is not present).

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: e02ee7d0-8d27-46f8-a856-222d6a938ecd

Author: JoshLove-msft

packages/http-client-csharp/emitter/src/lib/operation-converter.ts

@@ -19,6 +19,7 @@ import type {
   SdkQueryParameter,
   SdkServiceMethod,
   SdkServiceResponseHeader,
+  SdkStreamMetadata,
   SdkType,
 } from "@azure-tools/typespec-client-generator-core";
 import {
@@ -30,15 +31,18 @@ import {
   shouldGenerateConvenient,
   shouldGenerateProtocol,
 } from "@azure-tools/typespec-client-generator-core";
-import type { Diagnostic } from "@typespec/compiler";
+import type { Diagnostic, Union } from "@typespec/compiler";
 import {
+  compilerAssert,
   createDiagnosticCollector,
   getDeprecated,
   isErrorModel,
   NoTarget,
 } from "@typespec/compiler";
+import { unsafe_getEventDefinitions } from "@typespec/events/experimental";
 import type { HttpStatusCodeRange } from "@typespec/http";
 import { getResourceOperation } from "@typespec/rest";
+import { isTerminalEvent } from "@typespec/sse";
 import type { CSharpEmitterContext } from "../sdk-context.js";
 import { collectionFormatToDelimMap } from "../type/collection-format.js";
 import type { HttpResponseHeader } from "../type/http-response-header.js";
@@ -64,6 +68,7 @@ import type {
   InputMethodParameter,
   InputPathParameter,
   InputQueryParameter,
+  InputStreamingType,
   InputType,
 } from "../type/input-type.js";
 import { convertLroFinalStateVia } from "../type/operation-final-state-via.js";
@@ -234,7 +239,9 @@ export function fromSdkServiceMethodOperation(
     path: method.operation.path,
     externalDocsUrl: getExternalDocs(sdkContext, method.operation.__raw.operation)?.url,
     requestMediaTypes: requestMediaTypes,
-    bufferResponse: true,
+    bufferResponse: !method.operation.responses.some((response) =>
+      isSupportedStream(response.streamMetadata),
+    ),
     generateProtocolMethod: shouldGenerateProtocol(sdkContext, method.operation.__raw.operation),
     generateConvenienceMethod: generateConvenience,
     crossLanguageDefinitionId: method.crossLanguageDefinitionId,
@@ -351,8 +358,23 @@ function fromSdkServiceMethodParameters(
   for (const p of method.parameters) {
     const methodInputParameter = diagnostics.pipe(fromMethodParameter(sdkContext, p, namespace));
     const operationHttpParameter = getHttpOperationParameter(method, p);
+    const streamMetadata = method.operation.bodyParam?.streamMetadata;
+    const isStreamingBodyParameter =
+      isJsonLinesStream(streamMetadata) &&
+      method.operation.bodyParam!.methodParameterSegments.some((segments) =>
+        segments.some(
+          (segment) =>
+            segment === p || segment.crossLanguageDefinitionId === p.crossLanguageDefinitionId,
+        ),
+      );
 
     if (!operationHttpParameter) {
+      if (isStreamingBodyParameter) {
+        methodInputParameter.type = diagnostics.pipe(
+          fromSdkStreamMetadata(sdkContext, streamMetadata),
+        );
+        methodInputParameter.location = RequestLocation.Body;
+      }
       parameters.push(methodInputParameter);
       continue;
     }
@@ -365,6 +387,12 @@ function fromSdkServiceMethodParameters(
       rootApiVersions,
       diagnostics,
     );
+    if (isStreamingBodyParameter) {
+      methodInputParameter.type = diagnostics.pipe(
+        fromSdkStreamMetadata(sdkContext, streamMetadata),
+      );
+      methodInputParameter.location = RequestLocation.Body;
+    }
     parameters.push(methodInputParameter);
   }
 
@@ -378,6 +406,15 @@ function updateMethodParameter(
   rootApiVersions: string[],
   diagnostics: ReturnType<typeof createDiagnosticCollector>,
 ): void {
+  if (
+    operationHttpParameter.kind === "body" &&
+    isJsonLinesStream(operationHttpParameter.streamMetadata)
+  ) {
+    methodParameter.type = diagnostics.pipe(
+      fromSdkStreamMetadata(sdkContext, operationHttpParameter.streamMetadata),
+    );
+  }
+
   // for content type parameter
   if (isContentType(operationHttpParameter)) {
     methodParameter.type = diagnostics.pipe(
@@ -408,7 +445,9 @@ function fromSdkServiceMethodResponse(
   const diagnostics = createDiagnosticCollector();
 
   return diagnostics.wrap({
-    type: diagnostics.pipe(getResponseType(sdkContext, methodResponse.type)),
+    type: isSupportedStream(methodResponse.streamMetadata)
+      ? diagnostics.pipe(fromSdkStreamMetadata(sdkContext, methodResponse.streamMetadata))
+      : diagnostics.pipe(getResponseType(sdkContext, methodResponse.type)),
     resultSegments: methodResponse.resultSegments?.map((segment) =>
       getResponseSegmentName(segment),
     ),
@@ -607,7 +646,9 @@ function fromBodyParameter(
   rootApiVersions: string[],
 ): [InputBodyParameter, readonly Diagnostic[]] {
   const diagnostics = createDiagnosticCollector();
-  const parameterType = diagnostics.pipe(fromSdkType(sdkContext, p.type, p));
+  const parameterType = isJsonLinesStream(p.streamMetadata)
+    ? diagnostics.pipe(fromSdkStreamMetadata(sdkContext, p.streamMetadata))
+    : diagnostics.pipe(fromSdkType(sdkContext, p.type, p));
 
   const retVar: InputBodyParameter = {
     kind: "body",
@@ -725,7 +766,9 @@ export function fromSdkHttpOperationResponse(
   const range = sdkResponse.statusCodes;
   retVar = {
     statusCodes: toStatusCodesArray(range),
-    bodyType: diagnostics.pipe(getResponseType(sdkContext, sdkResponse.type)),
+    bodyType: isSupportedStream(sdkResponse.streamMetadata)
+      ? diagnostics.pipe(fromSdkStreamMetadata(sdkContext, sdkResponse.streamMetadata))
+      : diagnostics.pipe(getResponseType(sdkContext, sdkResponse.type)),
     headers: diagnostics.pipe(fromSdkServiceResponseHeaders(sdkContext, sdkResponse.headers)),
     isErrorResponse:
       sdkResponse.type !== undefined && isErrorModel(sdkContext.program, sdkResponse.type.__raw!),
@@ -737,6 +780,70 @@ export function fromSdkHttpOperationResponse(
   return diagnostics.wrap(retVar);
 }
 
+function fromSdkStreamMetadata(
+  sdkContext: CSharpEmitterContext,
+  streamMetadata: SdkStreamMetadata,
+): [InputStreamingType, readonly Diagnostic[]] {
+  const diagnostics = createDiagnosticCollector();
+  const originalType = streamMetadata.originalType;
+  const streamKind = getStreamKind(streamMetadata);
+  compilerAssert(
+    streamKind !== undefined,
+    "Stream metadata must have a supported stream kind before it is converted.",
+  );
+  let terminalEventType: string | undefined;
+  let terminalEventValue: string | undefined;
+
+  if (streamKind === "sse" && streamMetadata.streamType.__raw?.kind === "Union") {
+    const eventDefinitions = diagnostics.pipe(
+      unsafe_getEventDefinitions(sdkContext.program, streamMetadata.streamType.__raw as Union),
+    );
+    const terminalDefinition = eventDefinitions.find((definition) =>
+      isTerminalEvent(sdkContext.program, definition.root),
+    );
+    if (terminalDefinition?.payloadType.kind === "String") {
+      terminalEventType = terminalDefinition.eventType;
+      terminalEventValue = terminalDefinition.payloadType.value;
+    }
+  }
+
+  return diagnostics.wrap({
+    kind: "streaming",
+    name: originalType.kind === "model" ? originalType.name : "Stream",
+    valueType: diagnostics.pipe(fromSdkType(sdkContext, streamMetadata.streamType)),
+    streamKind,
+    contentTypes: streamMetadata.contentTypes,
+    terminalEventType,
+    terminalEventValue,
+    crossLanguageDefinitionId:
+      originalType.kind === "model" ? originalType.crossLanguageDefinitionId : "",
+  });
+}
+
+function getStreamKind(
+  streamMetadata: SdkStreamMetadata | undefined,
+): InputStreamingType["streamKind"] | undefined {
+  if (streamMetadata?.contentTypes.includes("application/jsonl")) {
+    return "jsonl";
+  }
+  if (streamMetadata?.contentTypes.includes("text/event-stream")) {
+    return "sse";
+  }
+  return undefined;
+}
+
+function isSupportedStream(
+  streamMetadata: SdkStreamMetadata | undefined,
+): streamMetadata is SdkStreamMetadata {
+  return getStreamKind(streamMetadata) !== undefined;
+}
+
+function isJsonLinesStream(
+  streamMetadata: SdkStreamMetadata | undefined,
+): streamMetadata is SdkStreamMetadata {
+  return getStreamKind(streamMetadata) === "jsonl";
+}
+
 function fromSdkServiceResponseHeaders(
   sdkContext: CSharpEmitterContext,
   headers: SdkServiceResponseHeader[],

packages/http-client-csharp/emitter/src/type/input-type.ts

@@ -79,6 +79,7 @@ export type InputType =
   | InputEnumType
   | InputEnumValueType
   | InputArrayType
+  | InputStreamingType
   | InputDictionaryType
   | InputNullableType;
 
@@ -305,6 +306,17 @@ export interface InputArrayType extends InputTypeBase {
   crossLanguageDefinitionId: string;
 }
 
+export interface InputStreamingType extends InputTypeBase {
+  kind: "streaming";
+  name: string;
+  valueType: InputType;
+  streamKind: "jsonl" | "sse";
+  contentTypes: string[];
+  terminalEventType?: string;
+  terminalEventValue?: string;
+  crossLanguageDefinitionId: string;
+}
+
 export interface InputDictionaryType extends InputTypeBase {
   kind: "dict";
   keyType: InputType;

packages/http-client-csharp/emitter/test/Unit/operation-converter.test.ts

@@ -215,6 +215,95 @@ describe("Operation Converter", () => {
   });
 
   describe("Operation response type conversion", () => {
+    it("preserves JSONL stream item types", async () => {
+      const program = await typeSpecCompile(
+        `
+          model Info {
+            desc: string;
+          }
+
+          @post op send(stream: JsonlStream<Info>): void;
+          op receive(): JsonlStream<Info>;
+        `,
+        runner,
+      );
+      const context = createEmitterContext(program);
+      const sdkContext = await createCSharpSdkContext(context);
+      const [root] = createModel(sdkContext);
+
+      const send = root.clients[0].methods.find((method) => method.name === "send");
+      ok(send);
+      const sendParameter = send.parameters.find((parameter) => parameter.name === "stream");
+      ok(sendParameter);
+      ok(sendParameter.type.kind === "streaming");
+      ok(sendParameter.type.valueType.kind === "model");
+      strictEqual(sendParameter.type.valueType.name, "Info");
+
+      const bodyParameter = send.operation.parameters.find(
+        (parameter) => parameter.kind === "body",
+      );
+      ok(bodyParameter);
+      ok(bodyParameter.type.kind === "streaming");
+      ok(bodyParameter.type.valueType.kind === "model");
+      strictEqual(bodyParameter.type.valueType.name, "Info");
+
+      const receive = root.clients[0].methods.find((method) => method.name === "receive");
+      ok(receive);
+      ok(receive.response.type?.kind === "streaming");
+      ok(receive.response.type.valueType.kind === "model");
+      strictEqual(receive.response.type.valueType.name, "Info");
+      ok(receive.operation.responses[0].bodyType?.kind === "streaming");
+      ok(receive.operation.responses[0].bodyType.valueType.kind === "model");
+      strictEqual(receive.operation.responses[0].bodyType.valueType.name, "Info");
+      strictEqual(receive.operation.bufferResponse, false);
+    });
+
+    it("preserves SSE event unions and terminal events", async () => {
+      const program = await typeSpecCompile(
+        `
+          model ResponseCreated {
+            id: string;
+          }
+
+          model ResponseDelta {
+            delta: string;
+          }
+
+          @events
+          union ResponseEvents {
+            @Events.contentType("application/json")
+            responseCreated: ResponseCreated,
+
+            @Events.contentType("application/json")
+            responseDelta: ResponseDelta,
+
+            @Events.contentType("text/plain")
+            @terminalEvent
+            "[DONE]",
+          }
+
+          op receive(): SSEStream<ResponseEvents>;
+        `,
+        runner,
+        { IsSseNeeded: true },
+      );
+      const context = createEmitterContext(program);
+      const sdkContext = await createCSharpSdkContext(context);
+      const [root] = createModel(sdkContext);
+
+      const receive = root.clients[0].methods.find((method) => method.name === "receive");
+      ok(receive);
+      ok(receive.response.type?.kind === "streaming");
+      strictEqual(receive.response.type.streamKind, "sse");
+      ok(receive.response.type.valueType.kind === "union");
+      strictEqual(receive.response.type.valueType.name, "ResponseEvents");
+      strictEqual(receive.response.type.terminalEventType, undefined);
+      strictEqual(receive.response.type.terminalEventValue, "[DONE]");
+      ok(receive.operation.responses[0].bodyType?.kind === "streaming");
+      strictEqual(receive.operation.responses[0].bodyType.streamKind, "sse");
+      strictEqual(receive.operation.bufferResponse, false);
+    });
+
     describe("With anonymous union enum response type", () => {
       it("should convert anonymous union enum response type to value type", async () => {
         const program = await typeSpecCompile(

packages/http-client-csharp/emitter/test/Unit/utils/test-util.ts

@@ -4,8 +4,11 @@ import { SdkTestLibrary } from "@azure-tools/typespec-client-generator-core/test
 import type { CompilerOptions, EmitContext, Program } from "@typespec/compiler";
 import type { TestHost } from "@typespec/compiler/testing";
 import { createTestHost } from "@typespec/compiler/testing";
+import { EventsTestLibrary } from "@typespec/events/testing";
 import { HttpTestLibrary } from "@typespec/http/testing";
 import { RestTestLibrary } from "@typespec/rest/testing";
+import { SSETestLibrary } from "@typespec/sse/testing";
+import { StreamsTestLibrary } from "@typespec/streams/testing";
 import { VersioningTestLibrary } from "@typespec/versioning/testing";
 import { XmlTestLibrary } from "@typespec/xml/testing";
 import { LoggerLevel } from "../../../src/lib/logger-level.js";
@@ -18,6 +21,9 @@ export async function createEmitterTestHost(): Promise<TestHost> {
     libraries: [
       RestTestLibrary,
       HttpTestLibrary,
+      EventsTestLibrary,
+      SSETestLibrary,
+      StreamsTestLibrary,
       VersioningTestLibrary,
       AzureCoreTestLibrary,
       SdkTestLibrary,
@@ -45,6 +51,7 @@ export interface TypeSpecCompileOptions {
   AuthDecorator?: string;
   NoEmit?: boolean;
   IsVersionNeeded?: boolean;
+  IsSseNeeded?: boolean;
 }
 
 export async function typeSpecCompile(
@@ -57,6 +64,7 @@ export async function typeSpecCompile(
   const needTCGC = options?.IsTCGCNeeded ?? false;
   const needXml = options?.IsXmlNeeded ?? false;
   const needVersion = options?.IsVersionNeeded ?? true;
+  const needSse = options?.IsSseNeeded ?? false;
   const authDecorator =
     options?.AuthDecorator ?? `@useAuth(ApiKeyAuth<ApiKeyLocation.header, "api-key">)`;
   const versions = `enum Versions {
@@ -77,12 +85,16 @@ export async function typeSpecCompile(
   const fileContent = `
     import "@typespec/rest";
     import "@typespec/http";
+    import "@typespec/http/streams";
+    ${needSse ? 'import "@typespec/events";\nimport "@typespec/sse";' : ""}
     import "@typespec/versioning";
     ${needXml ? 'import  "@typespec/xml";' : ""}
     ${needAzureCore ? 'import "@azure-tools/typespec-azure-core";' : ""}
     ${needTCGC ? 'import "@azure-tools/typespec-client-generator-core";' : ""}
     using TypeSpec.Rest; 
     using TypeSpec.Http;
+    using TypeSpec.Http.Streams;
+    ${needSse ? "using TypeSpec.Events;\nusing TypeSpec.SSE;" : ""}
     using TypeSpec.Versioning;
     ${needXml ? "using TypeSpec.Xml;" : ""}
     ${needAzureCore ? "using Azure.Core;\nusing Azure.Core.Traits;" : ""}

packages/http-client-csharp/eng/scripts/Spector-Helper.psm1

@@ -5,7 +5,6 @@ $failingSpecs = @(
   Join-Path 'http' 'type' 'model' 'templated'
   Join-Path 'http' 'type' 'file'
   Join-Path 'http' 'client' 'naming' # pending until https://github.com/microsoft/typespec/issues/5653 is resolved
-  Join-Path 'http' 'streaming' 'jsonl'
   Join-Path 'http' 'type' 'union' 'discriminated' # pending design
   Join-Path 'http' 'authentication' 'noauth' 'union' # pending design
 )