From 7bcd750d72f56eaec1bd457565b68db840ba8ae6 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:53:19 -0300 Subject: [PATCH] refactor(executors): extract pure wire protocol from perplexity-web (#6014) Extract the pure Perplexity wire protocol (consts, SSE stream types, SSE parsing, OpenAI<->Perplexity message translation, request/query builders, content extraction, sseChunk) verbatim into the leaf perplexity-web/protocol.ts. Host imports back the 10 symbols it uses; everything module-private (no re-export). Session cache, TLS fetch, auth, and the executor class stay in the host. Host 1028 -> 534 LOC. Byte-identical bodies (verbatim), leaf imports only randomUUID (no host import, no cycle). Adds a split-guard; consumer tests stay green (perplexity-web 26, streaming-tools-5927 2, tls-client 6, key-validation-models 2). --- open-sse/executors/perplexity-web.ts | 507 +----------------- open-sse/executors/perplexity-web/protocol.ts | 498 +++++++++++++++++ .../perplexity-web-executor-split.test.ts | 34 ++ 3 files changed, 544 insertions(+), 495 deletions(-) create mode 100644 open-sse/executors/perplexity-web/protocol.ts create mode 100644 tests/unit/perplexity-web-executor-split.test.ts diff --git a/open-sse/executors/perplexity-web.ts b/open-sse/executors/perplexity-web.ts index 2e3ea7dea5..d5c693cff3 100644 --- a/open-sse/executors/perplexity-web.ts +++ b/open-sse/executors/perplexity-web.ts @@ -16,73 +16,18 @@ import { import { prepareToolMessages } from "../translator/webTools.ts"; import { buildToolModeResponse } from "./chatgptWebTools.ts"; import { sanitizeErrorMessage } from "../utils/error.ts"; - -const PPLX_SSE_ENDPOINT = "https://www.perplexity.ai/rest/sse/perplexity_ask"; -// Perplexity's current request schema version (sent in params.version). Perplexity rejects -// stale versions with HTTP 400 — keep this in lockstep with the website's payload. -const PPLX_API_VERSION = "2.18"; -// Block use-cases the current web client advertises. The schematized API (use_schematized_api) -// validates the request shape, so this must be present (mirrors the browser request body). -const PPLX_SUPPORTED_BLOCK_USE_CASES = [ - "answer_modes", - "media_items", - "knowledge_cards", - "inline_entity_cards", - "place_widgets", - "finance_widgets", - "sports_widgets", - "news_widgets", - "shopping_widgets", - "jobs_widgets", - "search_result_widgets", - "inline_images", - "inline_assets", - "placeholder_cards", - "diff_blocks", - "inline_knowledge_cards", - "entity_group_v2", - "refinement_filters", - "canvas_mode", - "maps_preview", - "answer_tabs", - "price_comparison_widgets", - "preserve_latex", - "generic_onboarding_widgets", - "in_context_suggestions", - "pending_followups", - "inline_claims", - "unified_assets", - "workflow_steps", - "background_agents", -]; -// Firefox 148 — must match the `firefox_148` TLS profile used by perplexityTlsClient. -// A mismatched UA vs TLS fingerprint is itself a Cloudflare bot signal (issue #2459). -const PPLX_USER_AGENT = - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:148.0) Gecko/20100101 Firefox/148.0"; - -const MODEL_MAP: Record = { - "pplx-auto": ["concise", "pplx_pro"], - "pplx-sonar": ["copilot", "experimental"], - "pplx-gpt": ["copilot", "gpt54"], - "pplx-gemini": ["copilot", "gemini31pro_high"], - "pplx-sonnet": ["copilot", "claude46sonnet"], - "pplx-opus": ["copilot", "claude46opus"], - "pplx-nemotron": ["copilot", "nv_nemotron_3_super"], -}; - -const THINKING_MAP: Record = { - "pplx-gpt": "gpt54_thinking", - "pplx-sonnet": "claude46sonnetthinking", - "pplx-opus": "claude46opusthinking", -}; - -const CITATION_RE = /\[\d+\]/g; -const GROK_TAG_RE = /]*>.*?<\/grok:[^>]*>/gs; -const GROK_SELF_RE = /]*\/>/g; -const XML_DECL_RE = /<[?]xml[^?]*[?]>/g; -const RESPONSE_TAG_RE = /<\/?response\b[^>]*>/gi; -const MULTI_SPACE = / {2,}/g; -const MULTI_NL = /\n{3,}/g; +import { + PPLX_SSE_ENDPOINT, + PPLX_USER_AGENT, + MODEL_MAP, + THINKING_MAP, + cleanResponse, + parseOpenAIMessages, + buildPplxRequestBody, + buildQuery, + extractContent, + sseChunk, +} from "./perplexity-web/protocol.ts"; // ─── Session continuity ───────────────────────────────────────────────────── @@ -145,434 +90,6 @@ function sessionStore( } } -// ─── Helpers ──────────────────────────────────────────────────────────────── - -function cleanResponse(text: string, strip = true): string { - let t = text; - t = t.replace(XML_DECL_RE, ""); - t = t.replace(CITATION_RE, ""); - t = t.replace(GROK_TAG_RE, ""); - t = t.replace(GROK_SELF_RE, ""); - t = t.replace(RESPONSE_TAG_RE, ""); - if (strip) { - t = t.replace(MULTI_SPACE, " "); - t = t.replace(MULTI_NL, "\n\n"); - t = t.trim(); - } - return t; -} - -// ─── SSE types ────────────────────────────────────────────────────────────── - -interface PplxDiffPatch { - op?: string; - path?: string; - value?: unknown; -} - -interface PplxBlock { - intended_usage?: string; - markdown_block?: { - answer?: string; - chunks?: string[]; - 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 }>; - }; - plan_block?: { - steps?: Array<{ - step_type?: string; - search_web_content?: { queries?: Array<{ query?: string }> }; - read_results_content?: { urls?: string[] }; - }>; - goals?: Array<{ description?: string }>; - }; -} - -interface PplxStreamEvent { - status?: string; - final?: boolean; - text?: string; - blocks?: PplxBlock[]; - backend_uuid?: string; - web_results?: Array<{ url?: string; name?: string }>; - error_code?: string; - error_message?: string; - display_model?: string; -} - -// ─── SSE parsing ──────────────────────────────────────────────────────────── - -async function* readPplxSseEvents( - body: ReadableStream, - signal?: AbortSignal | null -): AsyncGenerator { - const reader = body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; - let dataLines: string[] = []; - - function flush(): PplxStreamEvent | null | "done" { - if (dataLines.length === 0) return null; - const payload = dataLines.join("\n"); - dataLines = []; - const trimmed = payload.trim(); - if (!trimmed || trimmed === "[DONE]") return "done"; - try { - return JSON.parse(trimmed) as PplxStreamEvent; - } catch { - return null; - } - } - - try { - while (true) { - if (signal?.aborted) return; - const { value, done } = await reader.read(); - if (done) break; - buffer += decoder.decode(value, { stream: true }); - - while (true) { - const idx = buffer.indexOf("\n"); - if (idx < 0) break; - const rawLine = buffer.slice(0, idx); - buffer = buffer.slice(idx + 1); - const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; - - if (line === "") { - const parsed = flush(); - if (parsed === "done") return; - if (parsed) yield parsed; - continue; - } - if (line.startsWith("data:")) { - dataLines.push(line.slice(5).trimStart()); - } - if (line === "event: end_of_stream") { - return; - } - } - } - - buffer += decoder.decode(); - if (buffer.trim().startsWith("data:")) { - dataLines.push(buffer.trim().slice(5).trimStart()); - } - const tail = flush(); - if (tail && tail !== "done") yield tail; - } finally { - reader.releaseLock(); - } -} - -// ─── OpenAI → Perplexity translation ──────────────────────────────────────── - -interface ParsedMessages { - systemMsg: string; - history: Array<{ role: string; content: string }>; - currentMsg: string; -} - -function parseOpenAIMessages(messages: Array>): ParsedMessages { - let systemMsg = ""; - const history: Array<{ role: string; content: string }> = []; - - for (const msg of messages) { - let role = String(msg.role || "user"); - if (role === "developer") role = "system"; - - let content = ""; - if (typeof msg.content === "string") { - content = msg.content; - } else if (Array.isArray(msg.content)) { - content = (msg.content as Array>) - .filter((c) => c.type === "text") - .map((c) => String(c.text || "")) - .join(" "); - } - if (!content.trim()) continue; - - if (role === "system") { - systemMsg += content + "\n"; - } else if (role === "user" || role === "assistant") { - history.push({ role, content }); - } - } - - let currentMsg = ""; - if (history.length > 0 && history[history.length - 1].role === "user") { - currentMsg = history.pop()!.content; - } - - return { systemMsg, history, currentMsg }; -} - -function buildPplxRequestBody( - query: string, - dslQuery: string, - mode: string, - modelPref: string, - followUpUuid: string | null, - requestId: string -): Record { - const tz = typeof Intl !== "undefined" ? Intl.DateTimeFormat().resolvedOptions().timeZone : "UTC"; - - // Mirrors the current www.perplexity.ai/rest/sse/perplexity_ask request body. Perplexity's - // schematized API validates this shape; an outdated version or missing required fields → HTTP 400. - const params: Record = { - attachments: [], - language: "en-US", - timezone: tz, - search_focus: "internet", - sources: ["web"], - frontend_uuid: requestId, - mode, - model_preference: modelPref, - is_related_query: false, - is_sponsored: false, - frontend_context_uuid: crypto.randomUUID(), - prompt_source: "user", - query_source: "home", - is_incognito: true, - local_search_enabled: false, - use_schematized_api: true, - send_back_text_in_streaming_api: false, - supported_block_use_cases: PPLX_SUPPORTED_BLOCK_USE_CASES, - client_coordinates: null, - mentions: [], - dsl_query: dslQuery && dslQuery.trim() ? dslQuery : query, - skip_search_enabled: true, - is_nav_suggestions_disabled: false, - source: "default", - always_search_override: false, - override_no_search: false, - client_search_results_cache_key: requestId, - should_ask_for_mcp_tool_confirmation: true, - browser_agent_allow_once_from_toggle: false, - force_enable_browser_agent: false, - supported_features: ["browser_agent_permission_banner_v1.1"], - extended_context: false, - version: PPLX_API_VERSION, - rum_session_id: crypto.randomUUID(), - }; - - // Only present on follow-ups (matches the browser, which omits it for a fresh query). - if (followUpUuid) { - params.last_backend_uuid = followUpUuid; - } - - return { - query_str: query, - params, - }; -} - -function buildQuery(parsed: ParsedMessages, followUpUuid: string | null): string { - if (followUpUuid) return parsed.currentMsg; - - const obj: Record = {}; - if (parsed.systemMsg.trim()) { - obj.instructions = [ - parsed.systemMsg.trim(), - "You have built-in web search. Answer questions directly using search results.", - ]; - } - if (parsed.history.length > 0) { - obj.history = parsed.history; - } - if (parsed.currentMsg) { - obj.query = parsed.currentMsg; - } else if (parsed.history.length === 0) { - obj.query = ""; - } - const json = JSON.stringify(obj); - return json.length > 96000 ? json.slice(-96000) : json; -} - -// ─── Content extraction ───────────────────────────────────────────────────── - -interface ContentChunk { - delta?: string; - answer?: string; - backendUuid?: string; - thinking?: string; - error?: string; - 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__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/", 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, - signal?: AbortSignal | null -): AsyncGenerator { - let fullAnswer = ""; - let backendUuid: string | null = null; - let seenLen = 0; - const seenThinking = new Set(); - // Per-usage reconstructed answer-text blocks + the locked primary usage. - const mdState = new Map(); - let primaryUsage: string | null = null; - - for await (const event of readPplxSseEvents(eventStream, signal)) { - if (event.error_code || event.error_message) { - yield { - error: event.error_message || `Perplexity error: ${event.error_code}`, - done: true, - }; - return; - } - - if (event.backend_uuid) backendUuid = event.backend_uuid; - - const blocks = event.blocks ?? []; - for (const block of blocks) { - const usage = block.intended_usage ?? ""; - - // Thinking: search steps - if (usage === "pro_search_steps" && block.plan_block?.steps) { - for (const step of block.plan_block.steps) { - if (step.step_type === "SEARCH_WEB") { - for (const q of step.search_web_content?.queries ?? []) { - const qr = q.query ?? ""; - if (qr && !seenThinking.has(qr)) { - seenThinking.add(qr); - yield { thinking: `Searching: ${qr}`, backendUuid: backendUuid ?? undefined }; - } - } - } else if (step.step_type === "READ_RESULTS") { - for (const u of (step.read_results_content?.urls ?? []).slice(0, 3)) { - if (u && !seenThinking.has(u)) { - seenThinking.add(u); - yield { thinking: `Reading: ${u}`, backendUuid: backendUuid ?? undefined }; - } - } - } - } - } - - // Thinking: plan goals - if (usage === "plan" && block.plan_block?.goals) { - for (const goal of block.plan_block.goals) { - const desc = goal.description ?? ""; - if (desc && !seenThinking.has(desc)) { - seenThinking.add(desc); - yield { thinking: desc, backendUuid: backendUuid ?? undefined }; - } - } - } - - // 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 (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; - } - } - - // 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(); - const looksLikeJson = t.startsWith("{") || t.startsWith("["); - if (!looksLikeJson && t.length > seenLen) { - const delta = t.slice(seenLen); - fullAnswer = t; - seenLen = t.length; - yield { delta, answer: fullAnswer, backendUuid: backendUuid ?? undefined }; - } - } - - // 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 }; -} - -// ─── OpenAI SSE format ────────────────────────────────────────────────────── - -function sseChunk(data: unknown): string { - return `data: ${JSON.stringify(data)}\n\n`; -} - function buildStreamingResponse( eventStream: ReadableStream, model: string, diff --git a/open-sse/executors/perplexity-web/protocol.ts b/open-sse/executors/perplexity-web/protocol.ts new file mode 100644 index 0000000000..102ef3f9b9 --- /dev/null +++ b/open-sse/executors/perplexity-web/protocol.ts @@ -0,0 +1,498 @@ +// Pure Perplexity wire protocol: consts, types, SSE parsing, request/query building, +// content extraction. Extracted verbatim from perplexity-web.ts. No host state/fetch/auth. +import { randomUUID } from "crypto"; + +export const PPLX_SSE_ENDPOINT = "https://www.perplexity.ai/rest/sse/perplexity_ask"; +// Perplexity's current request schema version (sent in params.version). Perplexity rejects +// stale versions with HTTP 400 — keep this in lockstep with the website's payload. +export const PPLX_API_VERSION = "2.18"; +// Block use-cases the current web client advertises. The schematized API (use_schematized_api) +// validates the request shape, so this must be present (mirrors the browser request body). +export const PPLX_SUPPORTED_BLOCK_USE_CASES = [ + "answer_modes", + "media_items", + "knowledge_cards", + "inline_entity_cards", + "place_widgets", + "finance_widgets", + "sports_widgets", + "news_widgets", + "shopping_widgets", + "jobs_widgets", + "search_result_widgets", + "inline_images", + "inline_assets", + "placeholder_cards", + "diff_blocks", + "inline_knowledge_cards", + "entity_group_v2", + "refinement_filters", + "canvas_mode", + "maps_preview", + "answer_tabs", + "price_comparison_widgets", + "preserve_latex", + "generic_onboarding_widgets", + "in_context_suggestions", + "pending_followups", + "inline_claims", + "unified_assets", + "workflow_steps", + "background_agents", +]; +// Firefox 148 — must match the `firefox_148` TLS profile used by perplexityTlsClient. +// A mismatched UA vs TLS fingerprint is itself a Cloudflare bot signal (issue #2459). +export const PPLX_USER_AGENT = + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:148.0) Gecko/20100101 Firefox/148.0"; + +export const MODEL_MAP: Record = { + "pplx-auto": ["concise", "pplx_pro"], + "pplx-sonar": ["copilot", "experimental"], + "pplx-gpt": ["copilot", "gpt54"], + "pplx-gemini": ["copilot", "gemini31pro_high"], + "pplx-sonnet": ["copilot", "claude46sonnet"], + "pplx-opus": ["copilot", "claude46opus"], + "pplx-nemotron": ["copilot", "nv_nemotron_3_super"], +}; + +export const THINKING_MAP: Record = { + "pplx-gpt": "gpt54_thinking", + "pplx-sonnet": "claude46sonnetthinking", + "pplx-opus": "claude46opusthinking", +}; + +export const CITATION_RE = /\[\d+\]/g; +export const GROK_TAG_RE = /]*>.*?<\/grok:[^>]*>/gs; +export const GROK_SELF_RE = /]*\/>/g; +export const XML_DECL_RE = /<[?]xml[^?]*[?]>/g; +export const RESPONSE_TAG_RE = /<\/?response\b[^>]*>/gi; +export const MULTI_SPACE = / {2,}/g; +export const MULTI_NL = /\n{3,}/g; + +// ─── Helpers ──────────────────────────────────────────────────────────────── + +export function cleanResponse(text: string, strip = true): string { + let t = text; + t = t.replace(XML_DECL_RE, ""); + t = t.replace(CITATION_RE, ""); + t = t.replace(GROK_TAG_RE, ""); + t = t.replace(GROK_SELF_RE, ""); + t = t.replace(RESPONSE_TAG_RE, ""); + if (strip) { + t = t.replace(MULTI_SPACE, " "); + t = t.replace(MULTI_NL, "\n\n"); + t = t.trim(); + } + return t; +} + +// ─── SSE types ────────────────────────────────────────────────────────────── + +export interface PplxDiffPatch { + op?: string; + path?: string; + value?: unknown; +} + +export interface PplxBlock { + intended_usage?: string; + markdown_block?: { + answer?: string; + chunks?: string[]; + 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 }>; + }; + plan_block?: { + steps?: Array<{ + step_type?: string; + search_web_content?: { queries?: Array<{ query?: string }> }; + read_results_content?: { urls?: string[] }; + }>; + goals?: Array<{ description?: string }>; + }; +} + +export interface PplxStreamEvent { + status?: string; + final?: boolean; + text?: string; + blocks?: PplxBlock[]; + backend_uuid?: string; + web_results?: Array<{ url?: string; name?: string }>; + error_code?: string; + error_message?: string; + display_model?: string; +} + +// ─── SSE parsing ──────────────────────────────────────────────────────────── + +export async function* readPplxSseEvents( + body: ReadableStream, + signal?: AbortSignal | null +): AsyncGenerator { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let dataLines: string[] = []; + + function flush(): PplxStreamEvent | null | "done" { + if (dataLines.length === 0) return null; + const payload = dataLines.join("\n"); + dataLines = []; + const trimmed = payload.trim(); + if (!trimmed || trimmed === "[DONE]") return "done"; + try { + return JSON.parse(trimmed) as PplxStreamEvent; + } catch { + return null; + } + } + + try { + while (true) { + if (signal?.aborted) return; + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + + while (true) { + const idx = buffer.indexOf("\n"); + if (idx < 0) break; + const rawLine = buffer.slice(0, idx); + buffer = buffer.slice(idx + 1); + const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; + + if (line === "") { + const parsed = flush(); + if (parsed === "done") return; + if (parsed) yield parsed; + continue; + } + if (line.startsWith("data:")) { + dataLines.push(line.slice(5).trimStart()); + } + if (line === "event: end_of_stream") { + return; + } + } + } + + buffer += decoder.decode(); + if (buffer.trim().startsWith("data:")) { + dataLines.push(buffer.trim().slice(5).trimStart()); + } + const tail = flush(); + if (tail && tail !== "done") yield tail; + } finally { + reader.releaseLock(); + } +} + +// ─── OpenAI → Perplexity translation ──────────────────────────────────────── + +export interface ParsedMessages { + systemMsg: string; + history: Array<{ role: string; content: string }>; + currentMsg: string; +} + +export function parseOpenAIMessages(messages: Array>): ParsedMessages { + let systemMsg = ""; + const history: Array<{ role: string; content: string }> = []; + + for (const msg of messages) { + let role = String(msg.role || "user"); + if (role === "developer") role = "system"; + + let content = ""; + if (typeof msg.content === "string") { + content = msg.content; + } else if (Array.isArray(msg.content)) { + content = (msg.content as Array>) + .filter((c) => c.type === "text") + .map((c) => String(c.text || "")) + .join(" "); + } + if (!content.trim()) continue; + + if (role === "system") { + systemMsg += content + "\n"; + } else if (role === "user" || role === "assistant") { + history.push({ role, content }); + } + } + + let currentMsg = ""; + if (history.length > 0 && history[history.length - 1].role === "user") { + currentMsg = history.pop()!.content; + } + + return { systemMsg, history, currentMsg }; +} + +export function buildPplxRequestBody( + query: string, + dslQuery: string, + mode: string, + modelPref: string, + followUpUuid: string | null, + requestId: string +): Record { + const tz = typeof Intl !== "undefined" ? Intl.DateTimeFormat().resolvedOptions().timeZone : "UTC"; + + // Mirrors the current www.perplexity.ai/rest/sse/perplexity_ask request body. Perplexity's + // schematized API validates this shape; an outdated version or missing required fields → HTTP 400. + const params: Record = { + attachments: [], + language: "en-US", + timezone: tz, + search_focus: "internet", + sources: ["web"], + frontend_uuid: requestId, + mode, + model_preference: modelPref, + is_related_query: false, + is_sponsored: false, + frontend_context_uuid: crypto.randomUUID(), + prompt_source: "user", + query_source: "home", + is_incognito: true, + local_search_enabled: false, + use_schematized_api: true, + send_back_text_in_streaming_api: false, + supported_block_use_cases: PPLX_SUPPORTED_BLOCK_USE_CASES, + client_coordinates: null, + mentions: [], + dsl_query: dslQuery && dslQuery.trim() ? dslQuery : query, + skip_search_enabled: true, + is_nav_suggestions_disabled: false, + source: "default", + always_search_override: false, + override_no_search: false, + client_search_results_cache_key: requestId, + should_ask_for_mcp_tool_confirmation: true, + browser_agent_allow_once_from_toggle: false, + force_enable_browser_agent: false, + supported_features: ["browser_agent_permission_banner_v1.1"], + extended_context: false, + version: PPLX_API_VERSION, + rum_session_id: crypto.randomUUID(), + }; + + // Only present on follow-ups (matches the browser, which omits it for a fresh query). + if (followUpUuid) { + params.last_backend_uuid = followUpUuid; + } + + return { + query_str: query, + params, + }; +} + +export function buildQuery(parsed: ParsedMessages, followUpUuid: string | null): string { + if (followUpUuid) return parsed.currentMsg; + + const obj: Record = {}; + if (parsed.systemMsg.trim()) { + obj.instructions = [ + parsed.systemMsg.trim(), + "You have built-in web search. Answer questions directly using search results.", + ]; + } + if (parsed.history.length > 0) { + obj.history = parsed.history; + } + if (parsed.currentMsg) { + obj.query = parsed.currentMsg; + } else if (parsed.history.length === 0) { + obj.query = ""; + } + const json = JSON.stringify(obj); + return json.length > 96000 ? json.slice(-96000) : json; +} + +// ─── Content extraction ───────────────────────────────────────────────────── + +export interface ContentChunk { + delta?: string; + answer?: string; + backendUuid?: string; + thinking?: string; + error?: string; + 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__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. +export 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). +export 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/", value:"..."}` frames. We only need the +// chunks array; joining it yields the cumulative answer text. +export 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; + } + } +} + +export async function* extractContent( + eventStream: ReadableStream, + signal?: AbortSignal | null +): AsyncGenerator { + let fullAnswer = ""; + let backendUuid: string | null = null; + let seenLen = 0; + const seenThinking = new Set(); + // Per-usage reconstructed answer-text blocks + the locked primary usage. + const mdState = new Map(); + let primaryUsage: string | null = null; + + for await (const event of readPplxSseEvents(eventStream, signal)) { + if (event.error_code || event.error_message) { + yield { + error: event.error_message || `Perplexity error: ${event.error_code}`, + done: true, + }; + return; + } + + if (event.backend_uuid) backendUuid = event.backend_uuid; + + const blocks = event.blocks ?? []; + for (const block of blocks) { + const usage = block.intended_usage ?? ""; + + // Thinking: search steps + if (usage === "pro_search_steps" && block.plan_block?.steps) { + for (const step of block.plan_block.steps) { + if (step.step_type === "SEARCH_WEB") { + for (const q of step.search_web_content?.queries ?? []) { + const qr = q.query ?? ""; + if (qr && !seenThinking.has(qr)) { + seenThinking.add(qr); + yield { thinking: `Searching: ${qr}`, backendUuid: backendUuid ?? undefined }; + } + } + } else if (step.step_type === "READ_RESULTS") { + for (const u of (step.read_results_content?.urls ?? []).slice(0, 3)) { + if (u && !seenThinking.has(u)) { + seenThinking.add(u); + yield { thinking: `Reading: ${u}`, backendUuid: backendUuid ?? undefined }; + } + } + } + } + } + + // Thinking: plan goals + if (usage === "plan" && block.plan_block?.goals) { + for (const goal of block.plan_block.goals) { + const desc = goal.description ?? ""; + if (desc && !seenThinking.has(desc)) { + seenThinking.add(desc); + yield { thinking: desc, backendUuid: backendUuid ?? undefined }; + } + } + } + + // 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 (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; + } + } + + // 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(); + const looksLikeJson = t.startsWith("{") || t.startsWith("["); + if (!looksLikeJson && t.length > seenLen) { + const delta = t.slice(seenLen); + fullAnswer = t; + seenLen = t.length; + yield { delta, answer: fullAnswer, backendUuid: backendUuid ?? undefined }; + } + } + + // 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 }; +} + +// ─── OpenAI SSE format ────────────────────────────────────────────────────── + +export function sseChunk(data: unknown): string { + return `data: ${JSON.stringify(data)}\n\n`; +} diff --git a/tests/unit/perplexity-web-executor-split.test.ts b/tests/unit/perplexity-web-executor-split.test.ts new file mode 100644 index 0000000000..9348704a63 --- /dev/null +++ b/tests/unit/perplexity-web-executor-split.test.ts @@ -0,0 +1,34 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { dirname, join } from "node:path"; + +// Split-guard for the perplexity-web executor protocol extraction. +// The pure wire protocol (consts, types, SSE parsing, request/query building, content +// extraction) lives in perplexity-web/protocol.ts (no host state/fetch/auth). Host imports +// back the symbols it uses; everything is module-private (no re-export). +const HERE = dirname(fileURLToPath(import.meta.url)); +const EXE = join(HERE, "../../open-sse/executors"); +const HOST = join(EXE, "perplexity-web.ts"); +const LEAF = join(EXE, "perplexity-web/protocol.ts"); + +test("leaf hosts the protocol helpers and does not import the host", () => { + const src = readFileSync(LEAF, "utf8"); + for (const sym of ["cleanResponse", "buildPplxRequestBody", "extractContent", "sseChunk"]) { + assert.match(src, new RegExp(`export (async function\\*?|function\\*?|const) ${sym}\\b`)); + } + assert.doesNotMatch(src, /from "\.\.\/perplexity-web\.ts"/); +}); + +test("host imports the protocol helpers back from the leaf", () => { + const host = readFileSync(HOST, "utf8"); + assert.match(host, /from "\.\/perplexity-web\/protocol\.ts"/); +}); + +test("cleanResponse strips citations and sseChunk formats a chunk", async () => { + const { cleanResponse, sseChunk } = + await import("../../open-sse/executors/perplexity-web/protocol.ts"); + assert.equal(typeof cleanResponse("hello", true), "string"); + assert.match(sseChunk({ a: 1 }), /^data: \{"a":1\}\n\n$/); +});