From 05b131188483670beda7faa2579ba67cfa8b36f8 Mon Sep 17 00:00:00 2001 From: "Jeyhun F. Aslanov" <64644942+jeyhunfaslanov@users.noreply.github.com> Date: Thu, 13 Aug 2026 19:38:11 +0400 Subject: [PATCH] fix(sse): extract perplexity-web answers from workflow_block (#10259) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Perplexity moved the answer text out of `markdown_block` into `workflow_block` (`intended_usage: "workflow_root"`), streaming it as RFC-6902 patches whose `field` is `"workflow_block"` and whose paths address `/steps//items//payload/text_payload/chunks/`. `extractContent` recognised neither shape. Two independent guards dropped every answer frame: - `isAnswerTextUsage("workflow_root")` is false, so the block loop `continue`d before any accumulation. - the diff guard skipped every patch whose `field !== "markdown_block"`. The stream therefore ran to `COMPLETED` with an empty accumulator and the executor surfaced `Provider returned empty content` (502) even though the upstream SSE carried the full answer. Every model was affected — the carrying block is model-independent — so the provider was unusable. Adds `workflow_block` to `PplxBlock`, an `applyWorkflowDiff` patch applier for the streaming path, and `applyWorkflowBlock` for a materialized block on the terminal frame. Answer tracks are keyed per step+item so concurrent items cannot overwrite each other's chunk indices, and only `variant: "answer"` payloads are accumulated — search queries, sources and "thinking" items stay out of the message. Fixtures in the regression test are trimmed from a live capture (pplx-auto, mode=copilot); replaying the full 96 KB capture through the patched extractor yields the complete 247-char answer over 7 incremental deltas, against an empty string before the fix. Co-authored-by: Jeyhun F. Aslanov --- open-sse/executors/perplexity-web/protocol.ts | 170 +++++++++++ .../perplexity-web-workflow-block.test.ts | 265 ++++++++++++++++++ 2 files changed, 435 insertions(+) create mode 100644 tests/unit/perplexity-web-workflow-block.test.ts diff --git a/open-sse/executors/perplexity-web/protocol.ts b/open-sse/executors/perplexity-web/protocol.ts index 0afc778e99..11e81f05d3 100644 --- a/open-sse/executors/perplexity-web/protocol.ts +++ b/open-sse/executors/perplexity-web/protocol.ts @@ -148,6 +148,36 @@ export interface PplxBlock { }>; goals?: Array<{ description?: string }>; }; + // Workflow API (`intended_usage: "workflow_root"`). Perplexity moved the answer + // text here from markdown_block: it now arrives as one WORKFLOW_ITEM_TEXT item + // whose `text_payload.variant` is "answer", nested under a workflow step. Other + // variants ("thinking") and item types (queries, sources) are not answer text. + workflow_block?: PplxWorkflowBlock; +} + +export interface PplxWorkflowTextPayload { + text?: string; + chunks?: string[]; + variant?: string; + is_streaming?: boolean; +} + +export interface PplxWorkflowItem { + type?: string; + variant?: string; + payload?: { text_payload?: PplxWorkflowTextPayload }; +} + +export interface PplxWorkflowStep { + status?: string; + title?: string; + tool_name?: string; + items?: PplxWorkflowItem[]; +} + +export interface PplxWorkflowBlock { + status?: string; + steps?: PplxWorkflowStep[]; } export interface PplxUpsellInformation { @@ -427,6 +457,134 @@ export function applyMarkdownDiff(acc: MarkdownAccumulator, patches: PplxDiffPat } } +/** Answer-text items carry this `variant`; "thinking" and friends are not answer text. */ +const WORKFLOW_ANSWER_VARIANT = "answer"; + +/** + * mdState key for one workflow answer item. Keyed per step+item so the + * `/chunks/` indices of two concurrent items can never overwrite each other. + */ +function workflowUsageKey(stepIdx: number, itemIdx: number): string { + return `workflow_root:${stepIdx}:${itemIdx}`; +} + +function isAnswerItem(item: PplxWorkflowItem | undefined): boolean { + if (!item) return false; + const payloadVariant = item.payload?.text_payload?.variant; + return (payloadVariant ?? item.variant) === WORKFLOW_ANSWER_VARIANT; +} + +/** + * Seed an accumulator from a materialized answer item. Chunks win over `text`: + * the terminal frame can carry a `text` that lags the chunk track (same + * precedence markdown_block already uses for `chunks` over `answer`). + */ +function seedFromAnswerItem(acc: MarkdownAccumulator, item: PplxWorkflowItem): void { + const tp = item.payload?.text_payload; + if (!tp) return; + if (Array.isArray(tp.chunks) && tp.chunks.length > 0) { + acc.chunks = tp.chunks.map((c) => String(c)); + } else if (typeof tp.text === "string" && tp.text.length > 0) { + acc.chunks = [tp.text]; + } +} + +function ensureAcc(mdState: Map, key: string): MarkdownAccumulator { + let acc = mdState.get(key); + if (!acc) { + acc = { chunks: [] }; + mdState.set(key, acc); + } + return acc; +} + +/** + * Apply a `field: "workflow_block"` diff patch set. + * + * Live shapes (Aug 2026 capture, pplx-auto / mode=copilot): + * {op:"add", path:"/steps/1", value:{items:[…]}} + * {op:"add", path:"/steps/0/items/1", value:{…}} + * {op:"add", path:"/steps/1/items/0/payload/text_payload/chunks/2", value:"…"} + * {op:"replace", path:"/steps/1/items/0/payload/text_payload/text", value:"…"} + * + * Only answer-variant items are accumulated; step/status patches are ignored. + */ +export function applyWorkflowDiff( + mdState: Map, + patches: PplxDiffPatch[] +): void { + for (const patch of patches) { + const path = patch.path ?? ""; + + // Whole step materialized — pick up every answer item it carries. + const stepMatch = /^\/steps\/(\d+)$/.exec(path); + if (stepMatch) { + const stepIdx = Number.parseInt(stepMatch[1], 10); + const step = (patch.value ?? {}) as PplxWorkflowStep; + (step.items ?? []).forEach((item, itemIdx) => { + if (!isAnswerItem(item)) return; + seedFromAnswerItem(ensureAcc(mdState, workflowUsageKey(stepIdx, itemIdx)), item); + }); + continue; + } + + // Single item appended to an existing step. + const itemMatch = /^\/steps\/(\d+)\/items\/(\d+)$/.exec(path); + if (itemMatch) { + const item = (patch.value ?? {}) as PplxWorkflowItem; + if (!isAnswerItem(item)) continue; + const key = workflowUsageKey( + Number.parseInt(itemMatch[1], 10), + Number.parseInt(itemMatch[2], 10) + ); + seedFromAnswerItem(ensureAcc(mdState, key), item); + continue; + } + + // Incremental chunk append — the streaming hot path. + const chunkMatch = /^\/steps\/(\d+)\/items\/(\d+)\/payload\/text_payload\/chunks\/(\d+)$/.exec( + path + ); + if (chunkMatch && typeof patch.value === "string") { + const key = workflowUsageKey( + Number.parseInt(chunkMatch[1], 10), + Number.parseInt(chunkMatch[2], 10) + ); + // Only extend a track already seeded by an answer item: a chunk patch + // carries no variant, so an unseeded key could be a "thinking" track. + const acc = mdState.get(key); + if (!acc) continue; + acc.chunks[Number.parseInt(chunkMatch[3], 10)] = patch.value; + continue; + } + + // Terminal `text` materialization — only used when no chunks arrived. + const textMatch = /^\/steps\/(\d+)\/items\/(\d+)\/payload\/text_payload\/text$/.exec(path); + if (textMatch && typeof patch.value === "string" && patch.value.length > 0) { + const key = workflowUsageKey( + Number.parseInt(textMatch[1], 10), + Number.parseInt(textMatch[2], 10) + ); + const acc = mdState.get(key); + if (!acc || acc.chunks.join("").length > 0) continue; + acc.chunks = [patch.value]; + } + } +} + +/** Accumulate every answer item of a materialized workflow_block. */ +export function applyWorkflowBlock( + mdState: Map, + workflow: PplxWorkflowBlock +): void { + (workflow.steps ?? []).forEach((step, stepIdx) => { + (step.items ?? []).forEach((item, itemIdx) => { + if (!isAnswerItem(item)) return; + seedFromAnswerItem(ensureAcc(mdState, workflowUsageKey(stepIdx, itemIdx)), item); + }); + }); +} + /** * Extract the assistant answer from the COMPLETED frame's `text` step-blob. * @@ -646,6 +804,18 @@ export async function* extractContent( } } + // Content: workflow_block answer items. Perplexity migrated the answer text + // here from markdown_block, so this must run BEFORE the isAnswerTextUsage + // gate — the carrying usage is "workflow_root", which that gate rejects. + if (block.workflow_block) { + applyWorkflowBlock(mdState, block.workflow_block); + continue; + } + if (block.diff_block?.field === "workflow_block") { + applyWorkflowDiff(mdState, block.diff_block.patches ?? []); + continue; + } + // Content: answer-text blocks (schematized diff frames OR materialized // markdown_block on the final COMPLETED frame). if (!isAnswerTextUsage(usage)) continue; diff --git a/tests/unit/perplexity-web-workflow-block.test.ts b/tests/unit/perplexity-web-workflow-block.test.ts new file mode 100644 index 0000000000..e7fc094b17 --- /dev/null +++ b/tests/unit/perplexity-web-workflow-block.test.ts @@ -0,0 +1,265 @@ +// Perplexity moved the answer text out of `markdown_block` into `workflow_block` +// (`intended_usage: "workflow_root"`), streaming it as RFC-6902 patches whose +// `field` is `"workflow_block"` and whose paths address +// `/steps//items//payload/text_payload/chunks/`. +// +// `extractContent` recognised neither shape: `isAnswerTextUsage("workflow_root")` +// is false, and the diff guard skipped every `field !== "markdown_block"` patch. +// The stream therefore completed with an empty accumulator and the executor +// surfaced "Provider returned empty content" while the upstream answer was +// present in the SSE all along. +// +// Fixtures below are trimmed from a live capture (pplx-auto, mode=copilot). + +import test from "node:test"; +import assert from "node:assert/strict"; + +const { extractContent } = await import("../../open-sse/executors/perplexity-web/protocol.ts"); + +function sseStream(events: unknown[]): ReadableStream { + const encoder = new TextEncoder(); + const chunks = events.map((e) => `event: message\r\ndata: ${JSON.stringify(e)}\r\n\r\n`); + chunks.push("event: end_of_stream\r\n\r\n"); + const body = chunks.join(""); + return new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(body)); + controller.close(); + }, + }); +} + +async function collect(events: unknown[]) { + let answer = ""; + const deltas: string[] = []; + for await (const chunk of extractContent(sseStream(events))) { + if (chunk.error) throw new Error(chunk.error); + if (typeof chunk.delta === "string") deltas.push(chunk.delta); + if (typeof chunk.answer === "string") answer = chunk.answer; + } + return { answer, deltas }; +} + +const ANSWER_PART_1 = "- The Caspian Sea is the world's largest inland"; +const ANSWER_PART_2 = " body of water by area, spanning about 371,000 square kilometers"; +const ANSWER_PART_3 = " (143,200 square miles).[1][2]"; +const FULL_ANSWER = ANSWER_PART_1 + ANSWER_PART_2 + ANSWER_PART_3; + +// `add /steps/1` seeds the answer item; later `add …/chunks/` frames append. +const STREAMING_EVENTS = [ + { + status: "PENDING", + backend_uuid: "76cf494e-2663-43e8-9382-5f37129707bb", + blocks: [ + { + intended_usage: "workflow_root", + diff_block: { + field: "workflow_block", + patches: [ + { op: "replace", path: "/status", value: "WORKFLOW_EXECUTING_STEPS" }, + { + op: "add", + path: "/steps/1", + value: { + status: "WORKFLOW_PENDING", + title: "", + items: [ + { + id: "45d63f9870ab45bc9bd10e8a264a6fc4", + type: "WORKFLOW_ITEM_TEXT", + payload: { + text_payload: { + text: "", + chunks: [ANSWER_PART_1], + variant: "answer", + is_streaming: true, + }, + }, + variant: "answer", + }, + ], + }, + }, + ], + }, + }, + ], + }, + { + status: "PENDING", + blocks: [ + { + intended_usage: "workflow_root", + diff_block: { + field: "workflow_block", + patches: [ + { + op: "add", + path: "/steps/1/items/0/payload/text_payload/chunks/1", + value: ANSWER_PART_2, + }, + ], + }, + }, + ], + }, + { + status: "COMPLETED", + final_sse_message: true, + blocks: [ + { + intended_usage: "workflow_root", + diff_block: { + field: "workflow_block", + patches: [ + { op: "replace", path: "/status", value: "WORKFLOW_COMPLETED" }, + { + op: "add", + path: "/steps/1/items/0/payload/text_payload/chunks/2", + value: ANSWER_PART_3, + }, + { + op: "replace", + path: "/steps/1/items/0/payload/text_payload/is_streaming", + value: false, + }, + ], + }, + }, + ], + }, +]; + +test("extractContent reconstructs the answer from workflow_block diff patches", async () => { + const { answer, deltas } = await collect(STREAMING_EVENTS); + + assert.equal(answer, FULL_ANSWER); + assert.equal(deltas.join(""), FULL_ANSWER, "deltas must concatenate to the full answer"); + assert.ok(deltas.length > 1, "streaming must emit incremental deltas, not one final blob"); +}); + +// A reconnect (or a truncated diff track) can deliver only the materialized +// workflow_block on the terminal frame — the answer must still be recovered. +test("extractContent reads a materialized workflow_block on the COMPLETED frame", async () => { + const { answer } = await collect([ + { + status: "COMPLETED", + final_sse_message: true, + blocks: [ + { + intended_usage: "workflow_root", + workflow_block: { + status: "WORKFLOW_COMPLETED", + steps: [ + { + status: "WORKFLOW_COMPLETED", + title: "Searching the web", + tool_name: "search_web", + items: [{ type: "WORKFLOW_ITEM_QUERIES", payload: { queries_payload: {} } }], + }, + { + status: "WORKFLOW_COMPLETED", + title: "", + items: [ + { + type: "WORKFLOW_ITEM_TEXT", + payload: { + text_payload: { + text: FULL_ANSWER, + chunks: [ANSWER_PART_1, ANSWER_PART_2, ANSWER_PART_3], + variant: "answer", + is_streaming: false, + }, + }, + variant: "answer", + }, + ], + }, + ], + }, + }, + ], + }, + ]); + + assert.equal(answer, FULL_ANSWER); +}); + +// Non-answer workflow items (search queries, sources) must never leak into the +// assistant message. +test("extractContent ignores non-answer workflow items", async () => { + const { answer } = await collect([ + { + status: "COMPLETED", + final_sse_message: true, + blocks: [ + { + intended_usage: "workflow_root", + workflow_block: { + status: "WORKFLOW_COMPLETED", + steps: [ + { + items: [ + { + type: "WORKFLOW_ITEM_TEXT", + payload: { + text_payload: { + text: "Searching the web for facts", + chunks: ["Searching the web for facts"], + variant: "thinking", + }, + }, + variant: "thinking", + }, + { + type: "WORKFLOW_ITEM_TEXT", + payload: { + text_payload: { + text: FULL_ANSWER, + chunks: [FULL_ANSWER], + variant: "answer", + }, + }, + variant: "answer", + }, + ], + }, + ], + }, + }, + ], + }, + ]); + + assert.equal(answer, FULL_ANSWER); +}); + +// Guard the pre-existing markdown_block path against regression from the fix. +test("extractContent still handles legacy markdown_block diffs", async () => { + const { answer } = await collect([ + { + status: "PENDING", + blocks: [ + { + intended_usage: "ask_text", + diff_block: { + field: "markdown_block", + patches: [{ op: "replace", path: "", value: { chunks: [ANSWER_PART_1] } }], + }, + }, + ], + }, + { + status: "COMPLETED", + final_sse_message: true, + blocks: [ + { + intended_usage: "ask_text", + markdown_block: { chunks: [ANSWER_PART_1, ANSWER_PART_2, ANSWER_PART_3] }, + }, + ], + }, + ]); + + assert.equal(answer, FULL_ANSWER); +});