fix(perplexity-web): parse schematized diff_block stream so answers aren't empty (#4001)

Integrated into release/v3.8.27 — schematized diff_block parsing follow-up to #3938.

config/quality/file-size-baseline.json

@@ -34,7 +34,7 @@
     "open-sse/executors/duckduckgo-web.ts": 917,
     "open-sse/executors/grok-web.ts": 1871,
     "open-sse/executors/muse-spark-web.ts": 1284,
-    "open-sse/executors/perplexity-web.ts": 939,
+    "open-sse/executors/perplexity-web.ts": 1013,
     "open-sse/handlers/audioSpeech.ts": 965,
     "open-sse/handlers/chatCore.ts": 5830,
     "open-sse/handlers/imageGeneration.ts": 3777,
@@ -154,5 +154,6 @@
   "_rebaseline_2026_06_14_r3_3848_compression": "PR #3848 own growth: chatCore.ts 5808→5811 (+3 = compression engine pipeline hooks). Also carries inherited release/v3.8.25 drift not touched by this PR: models/route.ts 2487→2489 (+2, post-#3836/prettier). Updating the frozen values restores Fast Quality Gates on the current base.",
   "_rebaseline_2026_06_14_3861_gitlab_duo": "PR #3861 own growth: oauth/[provider]/[action]/route.ts 903→916 (+13 = gitlab-duo authorize guard mirroring the existing qoder guard — returns a clear 'register an OAuth app + set GITLAB_DUO_OAUTH_CLIENT_ID' message instead of letting buildAuthUrl's throw become an opaque 500). Cohesive with the qoder branch right above it; not separately extractable.",
   "_rebaseline_2026_06_15_3941_provider_request_capture": "PR #3941 own growth: chatCore.ts 5823->5830 (+7 by check-file-size counting = run executor attempts inside the unified provider request capture scope), antigravity.ts 1649->1664 (+15) and codex.ts 1439->1447 (+8) = bridge hand-written upstream transports that bypass normal fetch/BaseExecutor capture. Cohesive logging-fidelity refactor; not extractable without hiding the actual dispatch boundary.",
-  "_rebaseline_2026_06_16_3958_qwen_body_check": "PR #3958 own growth: validation.ts 4407->4428 (+21 = validateQwenWebProvider now parses the /api/v2/user 200 body and requires a real user object, since Qwen returns HTTP 200 even for invalid tokens — fixes the validation false-positive, #3931). Cohesive with the existing qwen-web validation branch; not separately extractable."
+  "_rebaseline_2026_06_16_3958_qwen_body_check": "PR #3958 own growth: validation.ts 4407->4428 (+21 = validateQwenWebProvider now parses the /api/v2/user 200 body and requires a real user object, since Qwen returns HTTP 200 even for invalid tokens — fixes the validation false-positive, #3931). Cohesive with the existing qwen-web validation branch; not separately extractable.",
+  "_rebaseline_2026_06_16_4001_perplexity_diff_block": "PR #4001 own growth: perplexity-web.ts 939->1013 (+74 = parse the schematized API's RFC-6902 diff_block JSON-patch frames — applyMarkdownDiff + isAnswerTextUsage primary-usage lock + stop only on COMPLETED, not on a still-PENDING final flag — so streamed answers aren't empty, #3938 follow-up). Cohesive single-executor SSE-parsing logic; not separately extractable."
 }

open-sse/executors/perplexity-web.ts

@@ -163,6 +163,12 @@ function cleanResponse(text: string, strip = true): string {
 
 // ─── SSE types ──────────────────────────────────────────────────────────────
 
+interface PplxDiffPatch {
+  op?: string;
+  path?: string;
+  value?: unknown;
+}
+
 interface PplxBlock {
   intended_usage?: string;
   markdown_block?: {
@@ -171,6 +177,13 @@ interface PplxBlock {
     progress?: string;
     chunk_starting_offset?: number;
   };
+  // Schematized API (use_schematized_api) streams block updates as RFC-6902
+  // JSON-patch diffs against a target field (e.g. markdown_block) instead of
+  // sending the whole block each frame. `field` names the block being patched.
+  diff_block?: {
+    field?: string;
+    patches?: PplxDiffPatch[];
+  };
   web_result_block?: {
     web_results?: Array<{ url?: string; name?: string; snippet?: string }>;
   };
@@ -395,6 +408,40 @@ interface ContentChunk {
   done?: boolean;
 }
 
+// The schematized API delivers the answer text in blocks whose `intended_usage`
+// is either the aggregate `ask_text` or per-segment `ask_text_<n>_markdown`
+// (older builds used names merely containing "markdown"). All converge on the
+// same answer, so we lock onto a single primary usage to avoid double-counting.
+function isAnswerTextUsage(usage: string): boolean {
+  return usage === "ask_text" || /^ask_text_\d+_markdown$/.test(usage) || usage.includes("markdown");
+}
+
+// Reconstructed state for one answer-text block, built up from diff patches
+// (streaming) or a materialized markdown_block (final COMPLETED frame).
+interface MarkdownAccumulator {
+  chunks: string[];
+}
+
+// Apply a markdown_block diff_block patch set. Perplexity sends an initial
+// `{op:"replace", path:"", value:{chunks:[...]}}` then incremental
+// `{op:"add", path:"/chunks/<n>", value:"..."}` frames. We only need the
+// chunks array; joining it yields the cumulative answer text.
+function applyMarkdownDiff(acc: MarkdownAccumulator, patches: PplxDiffPatch[]): void {
+  for (const patch of patches) {
+    const path = patch.path ?? "";
+    if (path === "") {
+      const value = (patch.value ?? {}) as { chunks?: unknown };
+      acc.chunks = Array.isArray(value.chunks) ? value.chunks.map((c) => String(c)) : [];
+      continue;
+    }
+    const chunkMatch = /^\/chunks\/(\d+)$/.exec(path);
+    if (chunkMatch && typeof patch.value === "string") {
+      const idx = Number.parseInt(chunkMatch[1], 10);
+      acc.chunks[idx] = patch.value;
+    }
+  }
+}
+
 async function* extractContent(
   eventStream: ReadableStream<Uint8Array>,
   signal?: AbortSignal | null
@@ -403,6 +450,9 @@ async function* extractContent(
   let backendUuid: string | null = null;
   let seenLen = 0;
   const seenThinking = new Set<string>();
+  // Per-usage reconstructed answer-text blocks + the locked primary usage.
+  const mdState = new Map<string, MarkdownAccumulator>();
+  let primaryUsage: string | null = null;
 
   for await (const event of readPplxSseEvents(eventStream, signal)) {
     if (event.error_code || event.error_message) {
@@ -452,31 +502,52 @@ async function* extractContent(
         }
       }
 
-      // Content: markdown blocks
-      if (!usage.includes("markdown")) continue;
-      const mb = block.markdown_block;
-      if (!mb) continue;
-      const chunks = mb.chunks ?? [];
-      if (chunks.length === 0) continue;
+      // Content: answer-text blocks (schematized diff frames OR materialized
+      // markdown_block on the final COMPLETED frame).
+      if (!isAnswerTextUsage(usage)) continue;
+      let acc = mdState.get(usage);
+      if (!acc) {
+        acc = { chunks: [] };
+        mdState.set(usage, acc);
+      }
 
-      if (mb.progress === "DONE") {
-        fullAnswer = chunks.join("");
-      } else {
-        const chunkText = chunks.join("");
-        const cumulative = fullAnswer + chunkText;
-        if (cumulative.length > seenLen) {
-          const delta = cumulative.slice(seenLen);
-          fullAnswer = cumulative;
-          seenLen = cumulative.length;
-          yield { delta, answer: fullAnswer, backendUuid: backendUuid ?? undefined };
+      if (block.diff_block && Array.isArray(block.diff_block.patches)) {
+        applyMarkdownDiff(acc, block.diff_block.patches);
+      } else if (block.markdown_block) {
+        const mb = block.markdown_block;
+        if (Array.isArray(mb.chunks) && mb.chunks.length > 0) {
+          acc.chunks = mb.chunks.map((c) => String(c));
+        } else if (typeof mb.answer === "string" && mb.answer.length > 0) {
+          acc.chunks = [mb.answer];
         }
       }
+
+      // Prefer the aggregate `ask_text` block; otherwise lock the first seen.
+      if (usage === "ask_text") {
+        primaryUsage = "ask_text";
+      } else if (!primaryUsage) {
+        primaryUsage = usage;
+      }
     }
 
-    // Fallback: text field
-    if (blocks.length === 0 && event.text) {
+    // Emit at most one content delta per event, from the locked primary usage.
+    if (primaryUsage) {
+      const currentAnswer = (mdState.get(primaryUsage)?.chunks ?? []).join("");
+      if (currentAnswer.length > seenLen) {
+        const delta = currentAnswer.slice(seenLen);
+        fullAnswer = currentAnswer;
+        seenLen = currentAnswer.length;
+        yield { delta, answer: fullAnswer, backendUuid: backendUuid ?? undefined };
+      }
+    }
+
+    // Legacy fallback: a plain non-JSON `text` field with no structured blocks.
+    // The schematized API's `text` field is a JSON step-blob (not user-facing),
+    // so only use it when there are no answer-text blocks at all.
+    if (!primaryUsage && blocks.length === 0 && event.text) {
       const t = event.text.trim();
-      if (t.length > seenLen) {
+      const looksLikeJson = t.startsWith("{") || t.startsWith("[");
+      if (!looksLikeJson && t.length > seenLen) {
         const delta = t.slice(seenLen);
         fullAnswer = t;
         seenLen = t.length;
@@ -484,7 +555,10 @@ async function* extractContent(
       }
     }
 
-    if (event.final || event.status === "COMPLETED") break;
+    // Only stop on the terminal COMPLETED frame. A `final:true` flag can appear
+    // on a still-PENDING frame BEFORE the COMPLETED frame that materializes the
+    // full markdown_block — breaking on `final` there drops the answer.
+    if (event.status === "COMPLETED") break;
   }
 
   yield { delta: "", answer: fullAnswer, backendUuid: backendUuid ?? undefined, done: true };

tests/unit/perplexity-web.test.ts

@@ -242,8 +242,144 @@ test("Streaming: produces valid SSE chunks", async () => {
   }
 });
 
-// ─── Test: Thinking/reasoning content ───────────────────────────────────────
+// ─── Test: Schematized diff_block streaming (use_schematized_api) ───────────
 
+test("Schematized API: diff_block chunks reconstruct answer (non-streaming)", async () => {
+  // Mirrors the live www.perplexity.ai schematized API: the answer streams as
+  // RFC-6902 JSON-patch frames against markdown_block, a `final:true` flag
+  // arrives on a still-PENDING frame, then a COMPLETED frame materializes the
+  // full markdown_block. The parser must NOT stop on `final` and must apply
+  // the diff patches.
+  const pplxEvents = [
+    {
+      backend_uuid: "diff-uuid-1",
+      status: "PENDING",
+      blocks: [
+        {
+          intended_usage: "ask_text_0_markdown",
+          diff_block: {
+            field: "markdown_block",
+            patches: [
+              { op: "replace", path: "", value: { progress: "IN_PROGRESS", chunks: ["The "] } },
+            ],
+          },
+        },
+      ],
+    },
+    {
+      status: "PENDING",
+      blocks: [
+        {
+          intended_usage: "ask_text_0_markdown",
+          diff_block: { field: "markdown_block", patches: [{ op: "add", path: "/chunks/1", value: "answer " }] },
+        },
+      ],
+    },
+    {
+      status: "PENDING",
+      final: true,
+      blocks: [
+        {
+          intended_usage: "ask_text_0_markdown",
+          diff_block: { field: "markdown_block", patches: [{ op: "add", path: "/chunks/2", value: "is 42." }] },
+        },
+      ],
+    },
+    {
+      status: "COMPLETED",
+      final: true,
+      blocks: [
+        {
+          intended_usage: "ask_text_0_markdown",
+          markdown_block: { progress: "DONE", chunks: ["The answer is 42."], answer: "The answer is 42." },
+        },
+      ],
+    },
+  ];
+
+  const restore = mockFetch(200, pplxEvents);
+  try {
+    const executor = new PerplexityWebExecutor();
+    const result = await executor.execute({
+      model: "pplx-auto",
+      body: { messages: [{ role: "user", content: "what is the answer?" }], stream: false },
+      stream: false,
+      credentials: { apiKey: "test-cookie" },
+      signal: AbortSignal.timeout(10000),
+      log: null,
+    });
+
+    assert.equal(result.response.status, 200);
+    const json = JSON.parse(await result.response.text());
+    assert.equal(json.choices[0].message.content, "The answer is 42.");
+  } finally {
+    restore();
+  }
+});
+
+test("Schematized API: diff_block streams incremental deltas", async () => {
+  const pplxEvents = [
+    {
+      backend_uuid: "diff-uuid-2",
+      status: "PENDING",
+      blocks: [
+        {
+          intended_usage: "ask_text_0_markdown",
+          diff_block: { field: "markdown_block", patches: [{ op: "replace", path: "", value: { progress: "IN_PROGRESS", chunks: ["one, "] } }] },
+        },
+      ],
+    },
+    {
+      status: "PENDING",
+      blocks: [
+        {
+          intended_usage: "ask_text_0_markdown",
+          diff_block: { field: "markdown_block", patches: [{ op: "add", path: "/chunks/1", value: "two, " }] },
+        },
+      ],
+    },
+    {
+      status: "COMPLETED",
+      final: true,
+      blocks: [
+        {
+          intended_usage: "ask_text_0_markdown",
+          markdown_block: { progress: "DONE", chunks: ["one, two, three"], answer: "one, two, three" },
+        },
+      ],
+    },
+  ];
+
+  const restore = mockFetch(200, pplxEvents);
+  try {
+    const executor = new PerplexityWebExecutor();
+    const result = await executor.execute({
+      model: "pplx-auto",
+      body: { messages: [{ role: "user", content: "count" }], stream: true },
+      stream: true,
+      credentials: { apiKey: "test-cookie" },
+      signal: AbortSignal.timeout(10000),
+      log: null,
+    });
+
+    assert.equal(result.response.status, 200);
+    const text = await result.response.text();
+    let assembled = "";
+    for (const line of text.split("\n")) {
+      if (!line.startsWith("data: ")) continue;
+      const d = line.slice(6).trim();
+      if (d === "[DONE]") continue;
+      const o = JSON.parse(d);
+      const c = o.choices?.[0]?.delta?.content;
+      if (c) assembled += c;
+    }
+    assert.equal(assembled, "one, two, three");
+  } finally {
+    restore();
+  }
+});
+
+// ─── Test: Thinking/reasoning content ───────────────────────────────────────
 test("Streaming: thinking content emitted as reasoning_content", async () => {
   const pplxEvents = [
     {