Merge pull request #9341 from diegosouzapw/feat/9268-gemini-schema-recursive-type-empty-choices

feat(gemini): recursive schema type:object + empty choices interceptor (#9268)

changelog.d/features/9268-gemini-schema-recursive-type-empty-choices.md

@@ -0,0 +1 @@
+- **feat(gemini):** recursive type:object injection in schema normalizer + empty choices interceptor for streaming (#9268)

config/quality/file-size-baseline.json

@@ -523,7 +523,7 @@
   "open-sse/services/rateLimitManager.ts": "1167",
   "open-sse/translator/response/openai-responses.ts": "1204",
   "open-sse/utils/cursorAgentProtobuf.ts": "1505",
-  "open-sse/utils/stream.ts": "2889",
+  "open-sse/utils/stream.ts": 2915,
   "src/app/(dashboard)/dashboard/HomePageClient.tsx": "1388",
   "src/app/(dashboard)/dashboard/analytics/ComboHealthTab.tsx": "1031",
   "src/app/(dashboard)/dashboard/api-manager/ApiManagerPageClient.tsx": "3117",
@@ -572,5 +572,6 @@
   "_rebaseline_2026_08_09_v3850_release_close": "Release v3.8.50 close reconciliation on e0ce95c592: src/sse/handlers/chat.ts 1904->1918 is the irreducible request-pipeline wiring from #9759 that invokes the Modality Bridge guardrail without moving its implementation into the handler; covered by the 17 Vision Bridge canaries plus the PR-1 focused suite. open-sse/translator/response/openai-responses.ts 1204->1215 is #9168's Responses tool-call argument delta buffering/normalization at the existing translator state-machine chokepoint; covered by its dedicated translator regression tests. Both values are measured by check:file-size (split-newline semantics), and the gate remains frozen at the new exact sizes.",
   "_rebaseline_2026_08_08_toolcall_message_index_collision": "fix(responses-api): tool call after a text message collided on the same output_index. own growth: open-sse/translator/response/openai-responses.ts 1204->1224 (+20, extracted toolCallOutputIndexBase() shared helper so emitToolCall/closeToolCall can no longer compute a tool call's output_index independently and collide with a text message emitted in the same turn). Live incident (2026-08-08, OpenClaw agent): a client that tracks response items by output_index saw the tool call's added/delta/done events land on an index it had already marked complete (the just-closed text message), and silently dropped them — the agent spoke its preamble and never executed the tool call, even though OmniRoute's own recorded responseBody had a complete, valid tool_calls entry. Covered by the new regression test in tests/unit/translator-resp-openai-responses.test.ts reproducing the exact live scenario.",
   "_rebaseline_2026_08_03_9255_adobe_firefly_durable_sessions": "PR #9255 own cohesive growth: open-sse/services/adobeFireflyClient.ts 2322->2894 adds authenticated-vs-guest IMS classification, browser-risk ARP validation/rebuild, bounded 408 retry/recovery, sticky accepted-session handling, and matching image/video submit recovery at the existing Adobe upstream client chokepoints. This client was already explicitly frozen as a single self-contained upstream integration by #8006/#8510; splitting only the retry/auth helpers now would scatter one request state machine while structural shrink remains tracked in #3501. tests/unit/adobe-firefly.test.ts 871->1136 adds direct regression coverage for guest-token rejection, cookie/ARP rebuilding, 408 retries, sticky accepted ARP reuse, forced auth recovery, and cookie-to-IMS exchange. The obsolete 1179-line managed-Chrome fallback module was deleted rather than rebaselined after the packaged-safe pure-CDP path became authoritative. Focused Adobe suite: 61/61.",
-  "_rebaseline_2026_08_07_9653_disconnect_grace_period": "Extracted fix(sse): grace period before finalizing a client disconnect as 499 (#9653) — a client that closes its connection right after reading a fully-completed SSE stream can race OmniRoute's own completion bookkeeping, getting persisted as a false 499/0-tokens even though it delivered the full response (live-confirmed: a real disconnect at 18236ms was corrected to 200/82814+1292 tokens). Own growth: open-sse/handlers/chatCore.ts 5030->5039 (+9, wiring createClientDisconnectGraceHandler at the existing onClientDisconnectFinalize call site) — irreducible call-site wiring, the actual grace-period logic lives in the new leaf createClientDisconnectGraceHandler (open-sse/utils/streamFailureFinalization.ts, not frozen). Re-measured to 5042 after rebasing onto a newer release/v3.8.50 tip: the file carries an unrelated +3 base drift from already-merged upstream commits between this PR's original branch point and the rebase target, not covered by this entry. Covered by tests/unit/stream-disconnect-grace-period-9653.test.ts (4/4, fake-timer driven). Other file-size gate violations present on this base tip are pre-existing/unrelated to this change (base-red #9679, re-verify current issue number at merge time)."
+  "_rebaseline_2026_08_07_9653_disconnect_grace_period": "Extracted fix(sse): grace period before finalizing a client disconnect as 499 (#9653) — a client that closes its connection right after reading a fully-completed SSE stream can race OmniRoute's own completion bookkeeping, getting persisted as a false 499/0-tokens even though it delivered the full response (live-confirmed: a real disconnect at 18236ms was corrected to 200/82814+1292 tokens). Own growth: open-sse/handlers/chatCore.ts 5030->5039 (+9, wiring createClientDisconnectGraceHandler at the existing onClientDisconnectFinalize call site) — irreducible call-site wiring, the actual grace-period logic lives in the new leaf createClientDisconnectGraceHandler (open-sse/utils/streamFailureFinalization.ts, not frozen). Re-measured to 5042 after rebasing onto a newer release/v3.8.50 tip: the file carries an unrelated +3 base drift from already-merged upstream commits between this PR's original branch point and the rebase target, not covered by this entry. Covered by tests/unit/stream-disconnect-grace-period-9653.test.ts (4/4, fake-timer driven). Other file-size gate violations present on this base tip are pre-existing/unrelated to this change (base-red #9679, re-verify current issue number at merge time).",
+  "_rebaseline_2026_08_04_9268_gemini_schema_empty_choices": "Feature #9268 own growth: open-sse/utils/stream.ts 2889->2915 (+26 = irreducible call-site wiring for the empty-choices interceptor). The translate-mode flush now rejects a stream that completed without forwarding any valuable chunk (all-empty `choices: []`, no content/tool_calls/finish_reason) as a retryable 502 \"empty content\" instead of a clean empty 200 — the missing streaming counterpart of chatCore.ts's non-streaming isEmptyContentResponse. All rejection logic lives in the NEW leaf module open-sse/utils/streamEmptyChoices.ts (<cap, not frozen, unit-tested via tests/unit/stream-empty-choices-interceptor.test.ts); stream.ts only carries the `forwardedValuableChunk` boolean (declared at createSSEStream scope, set in emitTranslatedClientItem where the sole hasValuableContent check passes) plus the one flush-time rejectEmptyChoicesStream() call — the wait/orchestration at the chokepoint, not a movable block (mirrors the comboCooldownRetry.ts precedent). Schema-side twin fix: recursive type:\"object\" injection in open-sse/translator/helpers/geminiHelper.ts (not frozen, +33) for nested schemas with properties but no type (Gemini 400)."
 }

open-sse/translator/helpers/geminiHelper.ts

@@ -686,5 +686,38 @@ export function cleanJSONSchemaForAntigravity(schema: unknown): unknown {
 
   addPlaceholders(cleaned);
 
+  // Phase 7: Recursive type:"object" injection for nested schemas (#9268).
+  // Gemini/Vertex requires every node with properties/required to have an explicit
+  // `type: "object"`. Some clients (e.g. Composio-exported tools) emit nested
+  // schemas with `properties` but no `type`, causing a Gemini 400. Follow the
+  // `removeUnsupportedKeywords()`/`addPlaceholders()` visitor pattern.
+  function injectObjectType(obj: unknown): void {
+    if (!obj || typeof obj !== "object") return;
+
+    if (Array.isArray(obj)) {
+      for (const item of obj) {
+        injectObjectType(item);
+      }
+      return;
+    }
+
+    const record = obj as JsonRecord;
+    if (
+      !record.type &&
+      (record.properties !== undefined || record.required !== undefined)
+    ) {
+      record.type = "object";
+    }
+
+    // Recurse into remaining values.
+    for (const value of Object.values(record)) {
+      if (value && typeof value === "object") {
+        injectObjectType(value);
+      }
+    }
+  }
+
+  injectObjectType(cleaned);
+
   return cleaned;
 }

open-sse/utils/stream.ts

@@ -23,6 +23,7 @@ import {
   hasActiveDeltaValue,
   injectThinkingSignature,
 } from "./streamHelpers.ts";
+import { rejectEmptyChoicesStream, buildEmptyChoicesStreamError } from "./streamEmptyChoices.ts";
 import { calculateCost } from "@/lib/usage/costCalculator";
 import { buildOmniRouteSseMetadataComment } from "@/domain/omnirouteResponseMeta";
 import { sseCommentsEnabled } from "./sseHeartbeat.ts";
@@ -725,6 +726,9 @@ export function createSSEStream(options: StreamOptions = {}) {
         }
       : null;
 
+  // Tracks whether any valuable chunk was forwarded; empty at flush => retryable 502 (#9268)
+  let forwardedValuableChunk = false;
+
   // Track content length for usage estimation (both modes)
   let totalContentLength = 0;
   // Passthrough: accumulate content and reasoning separately for call log response body
@@ -995,6 +999,7 @@ export function createSSEStream(options: StreamOptions = {}) {
     const output = formatSSE(itemSanitized, sourceFormat);
     clientPayloadCollector.push(itemSanitized);
     reqLogger?.appendConvertedChunk?.(output);
+    forwardedValuableChunk = true;
     controller.enqueue(encoder.encode(output));
   };
 
@@ -2619,6 +2624,27 @@ export function createSSEStream(options: StreamOptions = {}) {
             return;
           }
 
+          // #9268: reject a translate-mode stream that forwarded no valuable chunk
+          // (all-empty `choices: []`) instead of completing with an empty 200.
+          if (
+            mode === STREAM_MODE.TRANSLATE &&
+            rejectEmptyChoicesStream({
+              forwardedValuableChunk,
+              hasValidUsage: hasValidUsage(state?.usage),
+              providerPayloadCollector,
+              clientPayloadCollector,
+              targetFormat,
+              model,
+              usage: state?.usage,
+              onFailure,
+              onComplete,
+              clearPendingRequestFromStream,
+            })
+          ) {
+            controller.error(markPendingRequestCleared(buildEmptyChoicesStreamError()));
+            return;
+          }
+
           // Flush remaining events (only once at stream end)
           const flushed = translateResponse(targetFormat, sourceFormat, null, state);
 

open-sse/utils/streamEmptyChoices.ts

@@ -0,0 +1,123 @@
+/**
+ * Empty-stream rejection for the SSE transform (#9268).
+ *
+ * A streaming provider can complete a turn having forwarded nothing usable —
+ * every chunk carried an empty `choices: []` (no content, no tool_calls, no
+ * finish_reason, e.g. a Gemini turn where the model emitted nothing). The SSE
+ * transform drops those chunks silently, so without a guard the stream would
+ * terminate with a clean empty 200, which clients treat as a valid empty turn
+ * and retry to their cap with no error to stop on.
+ *
+ * The transform is the only place that knows a chunk was actually forwarded, so
+ * `createSSEStream` threads a `forwardedValuableChunk` boolean and the
+ * flush-time callbacks. All rejection logic lives here so the frozen
+ * `open-sse/utils/stream.ts` only carries the minimal call-site wiring.
+ *
+ * Mirrors the non-streaming `isEmptyContentResponse` behavior in
+ * `open-sse/handlers/chatCore.ts` (empty content → retryable 502), and the
+ * #8649 disconnect-aware wrapper's "Provider returned empty content" outcome.
+ */
+import { buildErrorBody } from "./error.ts";
+import { buildStreamSummaryFromEvents } from "./streamPayloadCollector.ts";
+
+type StructuredSSEEventLike = {
+  index: number;
+  timestamp?: string;
+  event?: string;
+  data: unknown;
+};
+
+type StructuredSSECollectorLike = {
+  getEvents: () => StructuredSSEEventLike[];
+  build: (summary?: unknown, opts?: { includeEvents?: boolean }) => unknown;
+};
+
+type EmptyChoicesRejectContext = {
+  /** True when any chunk with content/tool_calls/finish_reason was forwarded. */
+  forwardedValuableChunk: boolean;
+  /** Valid usage accumulated on the stream state (usage-only streams are fine). */
+  hasValidUsage: boolean;
+  /** Provider-side event collector (for the onComplete providerPayload summary). */
+  providerPayloadCollector: StructuredSSECollectorLike;
+  /** Client-side payload collector (for the onComplete clientPayload). */
+  clientPayloadCollector: StructuredSSECollectorLike;
+  targetFormat?: string;
+  model?: string | null;
+  usage?: unknown;
+  onFailure?: ((payload: {
+    status: number;
+    message: string;
+    code?: string;
+    type?: string;
+  }) => boolean | void | Promise<void>) | null;
+  onComplete?: ((payload: {
+    status: number;
+    usage: unknown;
+    responseBody?: unknown;
+    providerPayload?: unknown;
+    clientPayload?: unknown;
+    error?: string | null;
+    errorCode?: string | null;
+  }) => void) | null;
+  clearPendingRequestFromStream?: () => void;
+};
+
+/**
+ * Returns `true` when the empty-stream condition was detected and the caller
+ * must abort the stream (controller.error + early return); `false` when the
+ * stream legitimately forwarded content/usage and should complete normally.
+ */
+export function rejectEmptyChoicesStream(ctx: EmptyChoicesRejectContext): boolean {
+  if (ctx.forwardedValuableChunk || ctx.hasValidUsage) return false;
+
+  const error = new Error(
+    "Provider returned empty content — stream forwarded no valuable chunks"
+  ) as Error & { statusCode: number; code: string };
+  error.statusCode = 502;
+  error.code = "empty_content";
+
+  if (ctx.onFailure) {
+    try {
+      ctx.onFailure({ status: 502, message: error.message, code: "empty_content" });
+    } catch {
+      // best-effort — must never break the stream error path
+    }
+  }
+
+  const errorBody = buildErrorBody(502, error.message);
+  if (ctx.onComplete) {
+    try {
+      ctx.onComplete({
+        status: 502,
+        usage: ctx.usage,
+        responseBody: errorBody,
+        error: error.message,
+        errorCode: "empty_content",
+        providerPayload: ctx.providerPayloadCollector.build(
+          buildStreamSummaryFromEvents(
+            ctx.providerPayloadCollector.getEvents(),
+            ctx.targetFormat,
+            ctx.model
+          ),
+          { includeEvents: false }
+        ),
+        clientPayload: ctx.clientPayloadCollector.build(errorBody, { includeEvents: false }),
+      });
+    } catch {
+      // best-effort
+    }
+  }
+
+  ctx.clearPendingRequestFromStream?.();
+  return true;
+}
+
+/** The retryable error the caller should surface via controller.error. */
+export function buildEmptyChoicesStreamError(): Error & { statusCode: number; code: string } {
+  const error = new Error(
+    "Provider returned empty content — stream forwarded no valuable chunks"
+  ) as Error & { statusCode: number; code: string };
+  error.statusCode = 502;
+  error.code = "empty_content";
+  return error;
+}

tests/unit/gemini-schema-recursive-type.test.ts

@@ -0,0 +1,142 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+
+const { cleanJSONSchemaForAntigravity } = await import(
+  "../../open-sse/translator/helpers/geminiHelper.ts"
+);
+
+test("#9268 injects type:object on nested properties without type", () => {
+  const input = {
+    type: "object",
+    properties: {
+      name: { type: "string" },
+      address: {
+        // nested node with properties but NO type — should get type:object
+        properties: {
+          street: { type: "string" },
+          city: { type: "string" },
+        },
+      },
+    },
+    required: ["name"],
+  };
+
+  const result = cleanJSONSchemaForAntigravity(input) as Record<string, unknown>;
+  const props = result.properties as Record<string, unknown>;
+  const address = props.address as Record<string, unknown>;
+
+  assert.equal(address.type, "object", "nested object with properties must get type:object");
+});
+
+test("#9268 injects type:object on nested items array schemas", () => {
+  const input = {
+    type: "object",
+    properties: {
+      items: {
+        type: "array",
+        items: {
+          // array items schema with properties but NO type
+          properties: {
+            id: { type: "integer" },
+            label: { type: "string" },
+          },
+        },
+      },
+    },
+  };
+
+  const result = cleanJSONSchemaForAntigravity(input) as Record<string, unknown>;
+  const props = result.properties as Record<string, unknown>;
+  const items = props.items as Record<string, unknown>;
+  const inner = items.items as Record<string, unknown>;
+
+  assert.equal(inner.type, "object", "array items schema with properties must inject type:object");
+});
+
+test("#9268 injects type:object on deeply nested schemas (3+ levels)", () => {
+  const input = {
+    type: "object",
+    properties: {
+      level1: {
+        properties: {
+          level2: {
+            properties: {
+              level3: {
+                properties: {
+                  value: { type: "string" },
+                },
+              },
+            },
+          },
+        },
+      },
+    },
+  };
+
+  const result = cleanJSONSchemaForAntigravity(input) as Record<string, unknown>;
+  const l1 = (result.properties as Record<string, unknown>).level1 as Record<string, unknown>;
+  const l2 = (l1.properties as Record<string, unknown>).level2 as Record<string, unknown>;
+  const l3 = (l2.properties as Record<string, unknown>).level3 as Record<string, unknown>;
+
+  assert.equal(l1.type, "object", "level1 must have type:object");
+  assert.equal(l2.type, "object", "level2 must have type:object");
+  assert.equal(l3.type, "object", "level3 must have type:object");
+});
+
+test("#9268 schema already typed is not double-injected", () => {
+  const input = {
+    type: "object",
+    properties: {
+      nested: {
+        type: "object",
+        properties: {
+          x: { type: "string" },
+        },
+      },
+    },
+  };
+
+  const result = cleanJSONSchemaForAntigravity(input) as Record<string, unknown>;
+  const nested = (result.properties as Record<string, unknown>).nested as Record<string, unknown>;
+
+  assert.equal(nested.type, "object", "already-typed nested must keep its type");
+  // Ensure properties is not clobbered
+  const nestedProps = nested.properties as Record<string, unknown>;
+  assert.ok(nestedProps, "nested properties must be preserved");
+  assert.ok("x" in nestedProps, "nested property 'x' must exist");
+});
+
+test("#9268 node with required but no properties still gets type:object", () => {
+  // Edge case: a node that has `required` but no `type` and no `properties`
+  // should still get type:object injection (Gemini needs it).
+  const input = {
+    type: "object",
+    properties: {
+      ref: {
+        // has required but no type nor properties (e.g. an incomplete $ref stub)
+        required: ["id"],
+      },
+    },
+  };
+
+  const result = cleanJSONSchemaForAntigravity(input) as Record<string, unknown>;
+  const ref = (result.properties as Record<string, unknown>).ref as Record<string, unknown>;
+
+  assert.equal(ref.type, "object", "node with required but no type must get type:object");
+});
+
+test("#9268 null/undefined fields do not crash the normalizer", () => {
+  const input = {
+    type: "object",
+    properties: {
+      a: null,
+      b: undefined,
+      // @ts-expect-error - testing runtime resilience
+      c: { properties: null },
+    },
+  };
+
+  assert.doesNotThrow(() => {
+    cleanJSONSchemaForAntigravity(input);
+  }, "null/undefined fields must not crash the normalizer");
+});

tests/unit/stream-empty-choices-interceptor.test.ts

@@ -0,0 +1,131 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+
+const { createSSETransformStreamWithLogger } = await import(
+  "../../open-sse/utils/stream.ts"
+);
+const { FORMATS } = await import("../../open-sse/translator/formats.ts");
+
+async function drainTransform(
+  transformStream: TransformStream<Uint8Array, Uint8Array>,
+  frames: string[]
+): Promise<{ output: string; errored: boolean }> {
+  const encoder = new TextEncoder();
+  const decoder = new TextDecoder();
+  const upstream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      for (const frame of frames) controller.enqueue(encoder.encode(frame));
+      controller.close();
+    },
+  });
+
+  const reader = upstream.pipeThrough(transformStream).getReader();
+  const parts: string[] = [];
+  let errored = false;
+  try {
+    for (;;) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      if (value) parts.push(decoder.decode(value));
+    }
+  } catch {
+    errored = true;
+  }
+  return { output: parts.join(""), errored };
+}
+
+function geminiContentChunk(text: string): string {
+  return `data: ${JSON.stringify({
+    candidates: [{ content: { parts: [{ text }] } }],
+  })}\n\n`;
+}
+
+function geminiFinishChunk(): string {
+  return `data: ${JSON.stringify({ candidates: [{ finishReason: "STOP" }] })}\n\n`;
+}
+
+function emptyChoicesChunk(id = "1"): string {
+  return `data: ${JSON.stringify({
+    id: `chatcmpl-${id}`,
+    object: "chat.completion.chunk",
+    model: "gemini-test",
+    choices: [],
+  })}\n\n`;
+}
+
+test("#9268 an all-empty-choices stream is rejected as a retryable error", async () => {
+  const transform = createSSETransformStreamWithLogger(
+    FORMATS.GEMINI,
+    FORMATS.OPENAI,
+    "gemini-test",
+    null,
+    null,
+    "gemini-model",
+    "conn-1",
+    { messages: [{ role: "user", content: "hi" }] },
+    null,
+    null,
+    null
+  );
+
+  const { output, errored } = await drainTransform(transform, [
+    emptyChoicesChunk("1"),
+    emptyChoicesChunk("2"),
+  ]);
+
+  // The translate-mode flush now errors the stream when no valuable chunk was
+  // forwarded, so the client must NOT see a clean empty 200 with just [DONE].
+  assert.ok(
+    errored || !output.includes("[DONE]"),
+    "an all-empty stream must not complete cleanly with a [DONE] terminator"
+  );
+});
+
+test("#9268 a stream with real content passes through unchanged", async () => {
+  const transform = createSSETransformStreamWithLogger(
+    FORMATS.GEMINI,
+    FORMATS.OPENAI,
+    "gemini-test",
+    null,
+    null,
+    "gemini-model",
+    "conn-2",
+    { messages: [{ role: "user", content: "hi" }] },
+    null,
+    null,
+    null
+  );
+
+  const { output, errored } = await drainTransform(transform, [
+    geminiContentChunk("hello"),
+    geminiFinishChunk(),
+  ]);
+
+  assert.ok(output.includes("hello"), "content must be forwarded");
+  assert.equal(errored, false, "a healthy stream must not error");
+});
+
+test("#9268 empty choices after real content still passes through (mid-stream usage-only)", async () => {
+  const transform = createSSETransformStreamWithLogger(
+    FORMATS.GEMINI,
+    FORMATS.OPENAI,
+    "gemini-test",
+    null,
+    null,
+    "gemini-model",
+    "conn-3",
+    { messages: [{ role: "user", content: "hi" }] },
+    null,
+    null,
+    null
+  );
+
+  const { output, errored } = await drainTransform(transform, [
+    geminiContentChunk("real output"),
+    emptyChoicesChunk("1"),
+    geminiFinishChunk(),
+  ]);
+
+  assert.ok(output.includes("real output"), "content must be forwarded");
+  assert.equal(errored, false, "a stream with content then empty usage chunk must not error");
+});