fix(sse): copy-on-write in the JSON-schema fallbacks and skip non-JSON response_format

changelog.d/fixes/18241-json-schema-fallback-retry.md

@@ -0,0 +1 @@
+- **fix(sse):** the JSON-schema fallback for kilocode / `openai-compatible-*` and opencode-zen's DeepSeek models no longer writes its instruction into the caller's system content, so a 429 retry inside one request stops sending it twice; `response_format: { type: "text" }` and other non-JSON formats no longer get the JSON instruction (kilocode still drops the field) (#18241).

open-sse/executors/default.ts

@@ -740,10 +740,13 @@ export class DefaultExecutor extends BaseExecutor {
       { type?: string; json_schema?: { schema?: unknown } } | undefined;
     if (!rf) return body;
 
-    // openai-compatible-* providers accept json_object natively — only the
-    // json_schema form needs downgrading there. kilocode rejects BOTH forms,
-    // so it enters the strip path below regardless.
+    // openai-compatible-* accept json_object natively; only json_schema needs downgrading there.
+    // kilocode rejects BOTH forms, so it strips below. Non-JSON formats (text/unknown/malformed)
+    // get no JSON prompt: kilocode still drops the field it rejects, others pass through.
+    const { response_format: _dropped, ...rest } = record;
     if (isOpenAiCompatible && rf.type === "json_object") return body;
+    if (rf.type !== "json_schema" && rf.type !== "json_object")
+      return isKiloCode ? (rest as T) : body;
     // Native DeepSeek accepts `text` and `json_object` — only json_schema is
     // rewritten, and always (a schema-less json_schema 400s there too).
     if (isDeepSeek && rf.type !== "json_schema") return body;
@@ -765,7 +768,7 @@ export class DefaultExecutor extends BaseExecutor {
       if (typeof sys.content === "string") {
         sys.content = `${sys.content}\n\n${prompt}`;
       } else if (Array.isArray(sys.content)) {
-        sys.content.push({ type: "text", text: `\n\n${prompt}` });
+        sys.content = [...sys.content, { type: "text", text: `\n\n${prompt}` }];
       }
     } else {
       messages.unshift({ role: "system", content: prompt });
@@ -775,10 +778,7 @@ export class DefaultExecutor extends BaseExecutor {
     // both json_schema AND json_object 400 with `param: response_format`) — strip
     // it entirely and rely on the schema prompt. openai-compatible-* providers
     // accept json_object, so keep the downgrade there.
-    if (isKiloCode) {
-      const { response_format: _dropped, ...rest } = record;
-      return { ...rest, messages } as T;
-    }
+    if (isKiloCode) return { ...rest, messages } as T;
     return { ...record, messages, response_format: { type: "json_object" } } as T;
   }
 

open-sse/executors/opencode.ts

@@ -1497,10 +1497,10 @@ export class OpencodeExecutor extends BaseExecutor {
       if (typeof systemMessage.content === "string") {
         systemMessage.content = `${systemMessage.content}\\n\\n${prompt}`;
       } else if (Array.isArray(systemMessage.content)) {
-        systemMessage.content.push({
-          type: "text",
-          text: `\\n\\n${prompt}`,
-        });
+        systemMessage.content = [
+          ...systemMessage.content,
+          { type: "text", text: `\\n\\n${prompt}` },
+        ];
       }
     } else {
       messages.unshift({

tests/unit/json-schema-fallback-retry.test.ts

@@ -0,0 +1,190 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+
+import { DefaultExecutor } from "../../open-sse/executors/default.ts";
+
+// applyJsonSchemaFallback (DefaultExecutor) for kilocode and openai-compatible-*:
+//  1. the JSON instruction must not be pushed into the caller's array-valued system
+//     content, otherwise every retry inside one execute() (the 429 intra-URL retry
+//     re-runs transformRequest on the same body) stacks another copy;
+//  2. only json_schema / json_object are JSON requests; for `text` (or any other type)
+//     no JSON instruction is added and the messages stay untouched. kilocode rejects ANY
+//     response_format, so it still drops the field; openai-compatible-* forwards it as-is.
+//  Every transformRequest case snapshots the input first and compares against the snapshot,
+//  so an in-place write into the caller's body cannot pass.
+
+const INSTRUCTION = "You must respond with valid JSON";
+const OPENAI_COMPAT_BASE = "https://compat.example/v1";
+const PROVIDERS = [
+  {
+    name: "openai-compatible-*",
+    id: "openai-compatible-deepseek",
+    credentials: { apiKey: "k", providerSpecificData: { baseUrl: OPENAI_COMPAT_BASE } },
+    urlPrefix: OPENAI_COMPAT_BASE,
+    stripsNonJson: false,
+  },
+  {
+    name: "kilocode",
+    id: "kilocode",
+    credentials: { apiKey: "k" },
+    urlPrefix: "https://api.kilo.ai/",
+    stripsNonJson: true,
+  },
+] as const;
+
+const SCHEMA = {
+  type: "object",
+  properties: { answer: { type: "string" } },
+  required: ["answer"],
+};
+
+type Message = { role?: string; content?: unknown };
+type Sent = { messages: Message[]; response_format?: unknown } & Record<string, unknown>;
+type Out = Sent;
+
+function countInstructions(messages: Message[]): number {
+  let n = 0;
+  for (const m of messages) {
+    const parts = Array.isArray(m.content)
+      ? (m.content as Array<{ text?: unknown }>).map((p) => p?.text ?? "")
+      : [m.content];
+    for (const text of parts) {
+      if (typeof text === "string") n += text.split(INSTRUCTION).length - 1;
+    }
+  }
+  return n;
+}
+
+// Drives the real BaseExecutor.execute(): the first upstream call is a 429 (retried
+// intra-URL after the 1 ms Retry-After hint), the second is a 200. The fetch stub
+// fails closed on any URL outside the provider's own endpoint.
+async function runWithRetry(provider: (typeof PROVIDERS)[number], body: Record<string, unknown>) {
+  const originalFetch = globalThis.fetch;
+  const sent: Sent[] = [];
+  globalThis.fetch = (async (url: unknown, init: RequestInit = {}) => {
+    const target = String(url instanceof Request ? url.url : url);
+    if (!target.startsWith(provider.urlPrefix)) {
+      throw new Error(`unexpected outbound URL in test: ${target}`);
+    }
+    sent.push(JSON.parse(String(init.body)));
+    if (sent.length === 1) {
+      return new Response(JSON.stringify({ error: { message: "slow down" } }), {
+        status: 429,
+        headers: { "Content-Type": "application/json", "Retry-After": "0.001" },
+      });
+    }
+    return new Response(JSON.stringify({ ok: true }), {
+      status: 200,
+      headers: { "Content-Type": "application/json" },
+    });
+  }) as typeof fetch;
+  try {
+    const executor = new DefaultExecutor(provider.id);
+    const result = await executor.execute({
+      model: "deepseek-chat",
+      body,
+      stream: false,
+      credentials: provider.credentials,
+    } as unknown as Parameters<DefaultExecutor["execute"]>[0]);
+    assert.equal(result.response.status, 200);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+  return sent;
+}
+
+for (const provider of PROVIDERS) {
+  test(`${provider.name}: array system content gets ONE instruction across a 429 retry and the caller body is untouched`, async () => {
+    const body = {
+      model: "deepseek-chat",
+      messages: [
+        { role: "system", content: [{ type: "text", text: "Be terse." }] },
+        { role: "user", content: "give me JSON" },
+      ],
+      response_format: { type: "json_schema", json_schema: { name: "s", schema: SCHEMA } },
+    };
+    const snapshot = structuredClone(body);
+
+    const sent = await runWithRetry(provider, body);
+
+    assert.equal(sent.length, 2, "expected exactly one 429 attempt and one retry");
+    assert.deepStrictEqual(body, snapshot, "caller body was mutated");
+    assert.equal(countInstructions(sent[0].messages), 1, "first attempt");
+    assert.equal(countInstructions(sent[1].messages), 1, "retry must not stack a second copy");
+  });
+
+  for (const [label, system] of [
+    ["array", [{ type: "text", text: "Be terse." }]],
+    ["string", "Be terse."],
+  ] as const) {
+    test(`${provider.name}: response_format text with ${label} system content gets no JSON instruction through execute()`, async () => {
+      const body = {
+        model: "deepseek-chat",
+        messages: [
+          { role: "system", content: system },
+          { role: "user", content: "hello" },
+        ],
+        response_format: { type: "text" },
+      };
+      const snapshot = structuredClone(body);
+
+      const sent = await runWithRetry(provider, body);
+
+      assert.equal(sent.length, 2);
+      for (const upstream of sent) {
+        if (provider.stripsNonJson) assert.equal("response_format" in upstream, false);
+        else assert.deepStrictEqual(upstream.response_format, { type: "text" });
+        assert.equal(countInstructions(upstream.messages), 0, "no JSON instruction for text");
+        assert.deepStrictEqual(upstream.messages, snapshot.messages);
+      }
+      assert.deepStrictEqual(body, snapshot, "caller body was mutated");
+    });
+  }
+
+  for (const responseFormat of [
+    { type: "text" },
+    { type: "something_new" },
+    {},
+    "json_object",
+  ] as const) {
+    test(`${provider.name}: transformRequest adds no instruction for response_format ${JSON.stringify(responseFormat)}`, () => {
+      const executor = new DefaultExecutor(provider.id);
+      const body = {
+        model: "deepseek-chat",
+        messages: [{ role: "user", content: "hello" }],
+        response_format: responseFormat,
+      };
+      const snapshot = structuredClone(body);
+
+      const out = executor.transformRequest(
+        "deepseek-chat",
+        body,
+        false,
+        provider.credentials
+      ) as Out;
+
+      assert.deepStrictEqual(body, snapshot, "caller body was mutated");
+      assert.deepStrictEqual(out.messages, snapshot.messages);
+      assert.equal(countInstructions(out.messages), 0);
+      if (provider.stripsNonJson) assert.equal("response_format" in out, false);
+      else assert.deepStrictEqual(out.response_format, snapshot.response_format);
+    });
+  }
+
+  test(`${provider.name}: json_schema without a schema is forwarded unchanged`, () => {
+    const executor = new DefaultExecutor(provider.id);
+    const body = {
+      model: "m",
+      messages: [{ role: "user", content: "hello" }],
+      response_format: { type: "json_schema", json_schema: { name: "s" } },
+    };
+    const snapshot = structuredClone(body);
+
+    const out = executor.transformRequest("m", body, false, provider.credentials) as Out;
+
+    assert.deepStrictEqual(body, snapshot, "caller body was mutated");
+    assert.deepStrictEqual(out.response_format, snapshot.response_format);
+    assert.deepStrictEqual(out.messages, snapshot.messages);
+    assert.equal(countInstructions(out.messages), 0);
+  });
+}

tests/unit/opencode-deepseek-json-schema-fallback-retry.test.ts

@@ -0,0 +1,117 @@
+import test, { before, after } from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+import { OpencodeExecutor } from "../../open-sse/executors/opencode.ts";
+
+// OpencodeExecutor.applyDeepSeekJsonSchemaFallback must not push its instruction into the
+// caller's array-valued system content: BaseExecutor.execute re-runs transformRequest on the
+// same body for the 429 intra-URL retry, so every retry would stack another copy.
+// The assertions count the instruction marker, never newlines.
+
+// Never read the shared /tmp marker from the real machine (only the multi-account loop does).
+// Set and restored in hooks so it cannot leak into other files under --test-isolation=none.
+const MARKER_ENV = "OPENCODE_POOL_STRAIN_MARKER_PATH";
+let priorMarker: string | undefined;
+let markerDir = "";
+
+before(() => {
+  priorMarker = process.env[MARKER_ENV];
+  markerDir = fs.mkdtempSync(path.join(os.tmpdir(), "oc-schema-retry-"));
+  process.env[MARKER_ENV] = path.join(markerDir, "strain.json");
+});
+
+after(() => {
+  if (priorMarker === undefined) delete process.env[MARKER_ENV];
+  else process.env[MARKER_ENV] = priorMarker;
+  fs.rmSync(markerDir, { recursive: true, force: true });
+});
+
+const MARKER = "You must respond with valid JSON that strictly follows";
+const MODEL = "deepseek-v4-flash-free";
+const SCHEMA = {
+  type: "object",
+  properties: { ok: { type: "boolean" } },
+  required: ["ok"],
+};
+
+type Message = { role?: string; content?: unknown };
+type Sent = { messages: Message[]; response_format?: unknown } & Record<string, unknown>;
+
+function countMarker(messages: Message[]): number {
+  let n = 0;
+  for (const m of messages) {
+    const parts = Array.isArray(m.content)
+      ? (m.content as Array<{ text?: unknown }>).map((p) => p?.text ?? "")
+      : [m.content];
+    for (const text of parts) {
+      if (typeof text === "string") n += text.split(MARKER).length - 1;
+    }
+  }
+  return n;
+}
+
+// Real OpencodeExecutor.execute() -> BaseExecutor.execute(): the first upstream call is a 429
+// (retried intra-URL after the 1 ms Retry-After hint), the second a 200. The fetch stub fails
+// closed on any URL outside opencode.ai.
+async function runWithRetry(body: Record<string, unknown>) {
+  const originalFetch = globalThis.fetch;
+  const sent: Sent[] = [];
+  const urls: string[] = [];
+  globalThis.fetch = (async (url: unknown, init: RequestInit = {}) => {
+    const target = String(url instanceof Request ? url.url : url);
+    urls.push(target);
+    if (!target.startsWith("https://opencode.ai/")) {
+      throw new Error(`unexpected outbound URL in test: ${target}`);
+    }
+    sent.push(JSON.parse(String(init.body)));
+    if (sent.length === 1) {
+      return new Response(JSON.stringify({ error: { message: "slow down" } }), {
+        status: 429,
+        headers: { "Content-Type": "application/json", "Retry-After": "0.001" },
+      });
+    }
+    return new Response(JSON.stringify({ ok: true }), {
+      status: 200,
+      headers: { "Content-Type": "application/json" },
+    });
+  }) as typeof fetch;
+  try {
+    const executor = new OpencodeExecutor("opencode-zen");
+    const result = await executor.execute({
+      model: MODEL,
+      body,
+      stream: false,
+      credentials: { apiKey: "test-key" },
+    } as unknown as Parameters<OpencodeExecutor["execute"]>[0]);
+    assert.equal((result as { response: Response }).response.status, 200);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+  return { sent, urls };
+}
+
+// opencode-zen is the only registered provider id the fallback gate applies to (the gate also
+// names a bare "opencode" id, which no registry entry defines; opencode-go is not gated).
+test(`opencode-zen: array system content gets ONE schema instruction across a 429 retry and the caller body is untouched`, async () => {
+  const body = {
+    model: MODEL,
+    messages: [
+      { role: "system", content: [{ type: "text", text: "Be terse." }] },
+      { role: "user", content: "give me JSON" },
+    ],
+    response_format: { type: "json_schema", json_schema: { name: "s", schema: SCHEMA } },
+  };
+  const snapshot = structuredClone(body);
+
+  const { sent, urls } = await runWithRetry(body);
+
+  assert.equal(sent.length, 2, `expected one 429 attempt and one retry, saw ${urls.join(",")}`);
+  assert.deepStrictEqual(body.messages, snapshot.messages, "caller messages were mutated");
+  assert.deepStrictEqual(body, snapshot, "caller body was mutated");
+  assert.equal(countMarker(sent[0].messages), 1, "first attempt");
+  assert.equal(countMarker(sent[1].messages), 1, "retry must not stack a second copy");
+  assert.deepStrictEqual(sent[1].response_format, { type: "json_object" });
+});