fix(streaming): preserve valid native empty terminal turns

changelog.d/fixes/16072-native-empty-terminals.md

@@ -0,0 +1 @@
+- fix(streaming): preserve valid empty native completions through JSON, SSE, combos and deduplicated requests for official OpenAI, Anthropic, Codex, OpenRouter and documented Azure endpoints, while retaining unknown-origin, incomplete-stream and upstream-error guards (#16072).

docs/routing/EMPTY_TURNS.md

@@ -0,0 +1,73 @@
+---
+title: "Valid Empty Assistant Turns"
+---
+
+# Valid empty assistant turns
+
+An assistant turn can finish normally without visible text. OmniRoute preserves
+such a turn when the selected execution used a supported native endpoint and the
+original upstream response supplied the protocol's normal terminal. An empty
+response without that evidence retains the existing empty-content guard.
+
+The execution boundary records this policy internally on the response identity.
+Client headers, a provider name, a planned connection, and a terminal synthesized
+by translation cannot grant the exception. The context survives response wrappers,
+JSON-to-SSE conversion, and non-streaming deduplication. It contains protocol and
+terminal flags, not credentials or a URL exposed to clients.
+
+## Supported execution endpoints
+
+| Execution                                        | Endpoint contract                                                                                                   |
+| ------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------- |
+| OpenAI Chat                                      | The registered `https://api.openai.com/v1/chat/completions` endpoint                                                |
+| Anthropic Messages, including Claude connections | The registered `https://api.anthropic.com/v1/messages` endpoint                                                     |
+| Codex Responses                                  | The registered `https://chatgpt.com/backend-api/codex/responses` endpoint                                           |
+| OpenRouter Chat                                  | The registered `https://openrouter.ai/api/v1/chat/completions` endpoint                                             |
+| Azure OpenAI / Foundry Responses                 | HTTPS resource hosts ending exactly in `.openai.azure.com` or `.services.ai.azure.com`, with `/openai/v1/responses` |
+| Azure Foundry Anthropic                          | HTTPS resource hosts ending exactly in `.services.ai.azure.com`, with `/anthropic/v1/messages`                      |
+
+The Azure paths follow Microsoft's [endpoint documentation](https://learn.microsoft.com/en-us/azure/ai-foundry/model-inference/concepts/endpoints?tabs=python),
+[Responses REST API](https://learn.microsoft.com/en-us/rest/api/aifoundry/azureopenai/responses),
+and [Anthropic SDK guidance](https://learn.microsoft.com/en-us/azure/foundry/how-to/develop/sdk-overview).
+Nonstandard Azure ports, URL credentials, lookalike hostname suffixes, and other
+custom endpoints do not inherit this policy.
+
+## Terminal and error handling
+
+- OpenAI Chat requires a native `finish_reason: "stop"`.
+- Claude requires `end_turn` or `stop_sequence` and a completed message lifecycle.
+  A lifecycle that opened a block but produced no useful output remains guarded.
+- Responses requires a native completed response. An initial `response.created`
+  followed by EOF cannot use the translator's synthetic stop as proof.
+- Native errors take priority over normal terminals. Existing quota errors,
+  missing-terminal handling, token-limit exemptions, reasoning output, and tool
+  output retain their separate classification.
+
+For a valid empty turn, direct requests return the completion, and combos stop
+without generating a response from a sibling target. Deduplicated callers share
+the policy of the execution they joined. The empty-turn exception does not write
+connection cooldown or model lockout state.
+
+A native terminal describes protocol completion; it cannot establish whether the
+model intended silence or a backend malfunctioned. Origin checks use the actual
+response URL when available and otherwise the effective request URL returned by
+the executor. An executor that rebuilds its response before this boundary can
+hide the final destination of an HTTP redirect; this is not a redirect attestation
+mechanism.
+
+## Regression evidence
+
+[The HTTP regression suite](../../tests/unit/valid-empty-stop-dispatch-16072.test.ts)
+uses the real API routes, stored synthetic connections, provider executors,
+translation, and combos. It intercepts outbound transport and counts generation
+requests. It covers official endpoint shapes, native JSON and SSE, unknown origins,
+errors, deduplication, and the optional empty-turn retry. These tests do not contact
+real provider accounts or reproduce the reporter's deployment or token totals.
+
+```sh
+node --import tsx/esm --import ./open-sse/utils/setupPolyfill.ts --test --test-force-exit tests/unit/valid-empty-stop-dispatch-16072.test.ts
+```
+
+The policy implementation is in
+[`emptyTurnPolicy.ts`](../../open-sse/utils/emptyTurnPolicy.ts); combo validation
+reads the execution's native terminal after consuming the response.

open-sse/handlers/chatCore/emptyTurnRetryLoop.ts

@@ -19,6 +19,7 @@ import {
   type RetryRouting,
 } from "../../utils/emptyTurnRetry.ts";
 import { noteBufferedVerdictOutcome } from "./emptyTurnResilienceNotes.ts";
+import { hasTrustedEmptyTurn } from "../../utils/emptyTurnPolicy.ts";
 import { STREAM_RECOVERY } from "../../config/constants.ts";
 
 type LoggerLike = {
@@ -82,16 +83,15 @@ export async function runEmptyTurnRetryLoop(
   let adopted = false;
 
   for (let attempt = 0; attempt <= maxRetries; attempt++) {
-    const verdict = judgeBufferedTurn(
-      await readBoundedResponseOutcome(
-        providerResponse,
-        FLUSH_EMPTY_RETRY_MAX_BYTES,
-        deps.timeoutMs
-      ),
-      deps.targetFormat,
-      deps.clientResponseFormat,
-      deps.isAborted()
+    const read = await readBoundedResponseOutcome(
+      providerResponse,
+      FLUSH_EMPTY_RETRY_MAX_BYTES,
+      deps.timeoutMs
     );
+    const verdict: Verdict =
+      read.kind === "text" && !deps.isAborted() && hasTrustedEmptyTurn(providerResponse)
+        ? { kind: "pass", why: "native execution completed normally" }
+        : judgeBufferedTurn(read, deps.targetFormat, deps.clientResponseFormat, deps.isAborted());
     if (verdict.kind === "pass") {
       notePass(deps, note, verdict);
       break;

open-sse/handlers/chatCore/executeProviderRequest.ts

@@ -3,6 +3,7 @@
  * implementation; the only change is that closed-over request state arrives
  * through deps.
  */
+import { markEmptyTurnExecution, inheritEmptyTurnPolicy } from "../../utils/emptyTurnPolicy.ts";
 
 import type { BaseExecutor } from "../../executors/base.ts";
 import type { AgentGoalPolicy } from "../../utils/agentGoalPolicy.ts";
@@ -365,6 +366,7 @@ export async function executeProviderRequest(
               }
             );
             const res = normalizeExecutorResult(rawExecutorResult);
+            res.response = markEmptyTurnExecution(res.response, res.url, execCreds);
             trace("post_executor", { status: res?.response?.status });
 
             // When a payload override rewrote body.model (custom-model alias →
@@ -620,11 +622,14 @@ export async function executeProviderRequest(
               return {
                 ...res,
                 _executionCredentials: execCreds,
-                response: new Response(clientBody, {
-                  status: res.response.status,
-                  statusText: res.response.statusText,
-                  headers: new Headers(normalizeHeaders(res.response.headers)),
-                }),
+                response: inheritEmptyTurnPolicy(
+                  res.response,
+                  new Response(clientBody, {
+                    status: res.response.status,
+                    statusText: res.response.statusText,
+                    headers: new Headers(normalizeHeaders(res.response.headers)),
+                  })
+                ),
               };
             }
 
@@ -693,7 +698,10 @@ export async function executeProviderRequest(
 
       return {
         ...rawResult,
-        response: new Response(payload, { status, statusText, headers: responseHeaders }),
+        response: inheritEmptyTurnPolicy(
+          rawResult.response,
+          new Response(payload, { status, statusText, headers: responseHeaders })
+        ),
         _dedupSnapshot: {
           status,
           statusText,

open-sse/handlers/chatCore/jsonBodyToSse.ts

@@ -13,6 +13,7 @@
  * existing readiness/error path still runs unchanged). Behaviour is byte-identical to the previous
  * inline block.
  */
+import { inheritEmptyTurnPolicy } from "../../utils/emptyTurnPolicy.ts";
 import { withBodyTimeout as defaultWithBodyTimeout } from "../../utils/stream.ts";
 import { synthesizeOpenAiSseFromJson as defaultSynthesize } from "../../utils/jsonToSse.ts";
 import { prependBufferedChunks } from "../../utils/streamReadiness.ts";
@@ -117,7 +118,7 @@ export async function maybeConvertJsonBodyToSse(
   }
 
   const { sseResponse, jsonBody } = await sniffJsonBodyForSse(providerResponse, ctx, deps);
-  if (sseResponse) return sseResponse;
+  if (sseResponse) return inheritEmptyTurnPolicy(providerResponse, sseResponse);
 
   const jsonText = await deps.withBodyTimeout<string>(jsonBody.text());
   const synthesizedSse = deps.synthesizeOpenAiSseFromJson(jsonText);
@@ -129,17 +130,23 @@ export async function maybeConvertJsonBodyToSse(
       `Upstream returned application/json on a streaming request — converting to SSE (${ctx.provider}/${ctx.model})`
     );
     rebuiltHeaders.set("content-type", "text/event-stream");
-    return new Response(synthesizedSse, {
-      status: providerResponse.status,
-      statusText: providerResponse.statusText,
-      headers: rebuiltHeaders,
-    });
+    return inheritEmptyTurnPolicy(
+      providerResponse,
+      new Response(synthesizedSse, {
+        status: providerResponse.status,
+        statusText: providerResponse.statusText,
+        headers: rebuiltHeaders,
+      })
+    );
   }
   // Not a convertible chat-completion JSON — rebuild the consumed body so the existing
   // readiness/error path still runs unchanged.
-  return new Response(jsonText, {
-    status: providerResponse.status,
-    statusText: providerResponse.statusText,
-    headers: rebuiltHeaders,
-  });
+  return inheritEmptyTurnPolicy(
+    providerResponse,
+    new Response(jsonText, {
+      status: providerResponse.status,
+      statusText: providerResponse.statusText,
+      headers: rebuiltHeaders,
+    })
+  );
 }

open-sse/handlers/chatCore/nonStreamingProviderLeg.ts

@@ -35,6 +35,7 @@ import {
   getModelFamily,
 } from "../../services/modelFamilyFallback.ts";
 import { isEmptyContentResponse } from "../../services/errorClassifier.ts";
+import { hasTrustedEmptyTurn } from "../../utils/emptyTurnPolicy.ts";
 import { FORMATS } from "../../translator/formats.ts";
 import { hasActiveClaudeThinking } from "../../utils/thinkingBudget.ts";
 
@@ -981,7 +982,12 @@ export async function runNonStreamingProviderLeg(
   // #14160: pass the provider so first-party APIs (antigravity) keep empty
   // completions with a normal stop reason as valid 200s instead of synthetic
   // 502s feeding model lockout.
-  if (isEmptyContentResponse(responseBody, { provider })) {
+  if (
+    isEmptyContentResponse(responseBody, {
+      provider,
+      trustedEmptyTurn: hasTrustedEmptyTurn(executorResult.response),
+    })
+  ) {
     const errMsg = "Provider returned empty content";
     if (allowModelFallback) {
       const triedModels = new Set<string>([currentModel]);

open-sse/handlers/chatCore/nonStreamingResponse.ts

@@ -64,6 +64,7 @@ import {
   toAntigravityDiagnosticPayload,
 } from "../../executors/antigravityUpstreamError.ts";
 import { routingFinishReason } from "../chatCore/routingFinishReason.ts";
+import { hasTrustedEmptyTurn, inheritEmptyTurnPolicy } from "../../utils/emptyTurnPolicy.ts";
 import { getProviderCredentials } from "@/sse/services/auth";
 import { extractFacts } from "@/lib/memory/extraction";
 // The leaf body is unchanged from the barrel, so its closed-over values keep
@@ -828,7 +829,11 @@ export async function runNonStreamingResponse(deps: NonStreamingDeps) {
     // this check runs after translation + sanitization + tool-call execution to catch
     // cases where a provider returns a structurally valid raw body that translates into
     // choices:[] or output:[] with no usable content (Responses API shape included).
-    const malformedTranslatedReason = detectMalformedNonStream(translatedResponse, provider);
+    const malformedTranslatedReason = detectMalformedNonStream(
+      translatedResponse,
+      provider,
+      hasTrustedEmptyTurn(providerResponse)
+    );
     if (malformedTranslatedReason) {
       const malformedDiagnostic = buildMalformedResponseDiagnostic(
         malformedTranslatedReason,
@@ -1065,11 +1070,14 @@ export async function runNonStreamingResponse(deps: NonStreamingDeps) {
     return {
       response: {
         success: true,
-        response: maybeWrapForcedNonStreamingResponsesJson({
-          clientRequestedResponsesStream,
-          body: translatedResponse,
-          headers: responseHeaders,
-        }),
+        response: inheritEmptyTurnPolicy(
+          providerResponse,
+          maybeWrapForcedNonStreamingResponsesJson({
+            clientRequestedResponsesStream,
+            body: translatedResponse,
+            headers: responseHeaders,
+          })
+        ),
       },
       carry: {
         translatedBody,

open-sse/handlers/chatCore/responseHeaders.ts

@@ -1,3 +1,4 @@
+import { inheritEmptyTurnPolicy } from "../../utils/emptyTurnPolicy.ts";
 import {
   attachOmniRouteMetaHeaders,
   buildOmniRouteResponseMetaHeaders,
@@ -370,11 +371,14 @@ export function materializeDeduplicatedExecutionResult<T extends Record<string,
 
   return {
     ...result,
-    response: new Response(snapshot.payload, {
-      status: snapshot.status,
-      statusText: snapshot.statusText,
-      headers: snapshot.headers,
-    }),
+    response: inheritEmptyTurnPolicy(
+      result.response as Response,
+      new Response(snapshot.payload, {
+        status: snapshot.status,
+        statusText: snapshot.statusText,
+        headers: snapshot.headers,
+      })
+    ),
   } as T;
 }
 

open-sse/handlers/chatCore/streamingTail.ts

@@ -1,3 +1,4 @@
+import { inheritEmptyTurnPolicy } from "../../utils/emptyTurnPolicy.ts";
 import { getCodexClientSessionId } from "../../config/codexIdentity.ts";
 import {
   noteCodexTurnStateProvenance,
@@ -745,9 +746,12 @@ export async function runStreamingTail(deps: StreamingTailDeps) {
     response: { status: 200, streamed: true },
   });
 
-  const response = new Response(clientFacingStream, {
-    headers: responseHeaders,
-  });
+  const response = inheritEmptyTurnPolicy(
+    providerResponse,
+    new Response(clientFacingStream, {
+      headers: responseHeaders,
+    })
+  );
   turnExecutionHandedOffToStream = true;
   return {
     result: {

open-sse/services/combo/emptyTurnTrust.ts

@@ -1,29 +1,10 @@
-import { getCachedProviderConnectionById } from "@/lib/db/readCache";
-import { isOfficialAnthropicBaseUrl } from "../../utils/anthropicHost.ts";
+import { hasTrustedEmptyTurn } from "../../utils/emptyTurnPolicy.ts";
 
-/** Only the selected first-party Anthropic connection can complete without content blocks. */
+/** Read after consumption: a translated stop alone is not native termination. */
 export async function isTrustedEmptyTurn(
-  provider: string | null,
+  _provider: string | null,
   response: Response,
-  fallbackConnectionId?: string | null
-): Promise<boolean> {
-  if (provider !== null && provider !== "claude" && provider !== "anthropic") return false;
-
-  const selectedConnectionId = response.headers.get("X-OmniRoute-Selected-Connection-Id");
-  // An unresolved alias cannot use a planned connection as proof of actual dispatch.
-  const connectionId = selectedConnectionId || (provider === null ? null : fallbackConnectionId);
-  if (!connectionId) return false;
-
-  try {
-    const connection = await getCachedProviderConnectionById(connectionId);
-    if (connection?.provider !== "claude" && connection?.provider !== "anthropic") return false;
-    if (provider !== null && connection.provider !== provider) return false;
-    const data = connection.providerSpecificData;
-    if (data != null && (typeof data !== "object" || Array.isArray(data))) return false;
-    const baseUrl = (data as Record<string, unknown> | null)?.baseUrl;
-    if (baseUrl != null && typeof baseUrl !== "string") return false;
-    return isOfficialAnthropicBaseUrl(typeof baseUrl === "string" ? baseUrl : "");
-  } catch {
-    return false;
-  }
+  _fallbackConnectionId?: string | null
+): Promise<() => boolean> {
+  return () => hasTrustedEmptyTurn(response);
 }