pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-05 23:02:10 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(providers): migrate muse-spark-web from GraphQL to WebSocket protocol (#7528)

Browse Source 
* fix(ci): add the auto-enqueue pull_request_rule to the Mergify config (queue_conditions alone are eligibility-only) (#7179)

* fix(ci): migrate Mergify auto-enqueue to merge_protections_settings.auto_merge_conditions (rules-based path is EOL 2026-07-16) (#7216)

* fix(ci): drop Mergify batch settings (batching is a paid-tier feature; free plan queue is serial) (#7220)

* fix(ci): merge queue tolerates the advisory dast-smoke failure (its GH-hosted build hang dequeued every attempt) (#7225)

* feat: add protobuf+WS helpers and tests for muse-spark-web

Co-Authored-By: Claude <noreply@anthropic.com>

* fix: remove 50ms auto-close timer from wsChat, fix test mock to respond properly

The 50ms setTimeout in wsChat sent a close signal before the server
could respond. Tests now trigger a response event from the mock's send()
and then close naturally. wsChat waits indefinitely (or until timeout)
for real server data.

Co-Authored-By: Claude <noreply@anthropic.com>

* fix(provider): migrate muse-spark-web from GraphQL to WebSocket protocol

Meta AI retired the persisted query (doc_id 29ae946c...) that OmniRoute
used for message sending. The AttachmentInput type was removed from
Meta's GraphQL schema, causing 502 errors on every request.

Replace the old GraphQL POST approach with Meta's current protocol:
  1. GraphQL warmup (doc_id e7f80258...) — init conversation
  2. GraphQL mode switch (doc_id c32bbe99...) — set think_fast/think_hard
  3. WebSocket (wss://gateway.meta.ai/ws/clippy) — protobuf-framed messaging

All frame encoding uses inline protobuf helpers (no new deps).
The existing continuation cache, model mapping, and response formatters
are preserved.

Fixes #7267

Co-Authored-By: Claude <noreply@anthropic.com>

* fix: add warmup+mode-switch GraphQL calls and Buffer ESM import

Also moves modelInfo extraction earlier so mode-switch can use it.
Co-Authored-By: Claude <noreply@anthropic.com>

* fix: share requestId between WS URL and prompt frame, add auth fallback

- Pass requestId from wsChat into buildWsPromptFrame so both the WS URL
  and the prompt frame use the same identifier, matching Meta's protocol.
- Add fallback to extract the ecto1:... authorization token from the apiKey
  cookie string when providerSpecificData.authorization is not set. This
  lets users paste both the cookie and auth token in OmniRouter's single
  input field (e.g. 'ecto_1_sess=...; ecto1:...').

Co-Authored-By: Claude <noreply@anthropic.com>

* fix: address Gemini Code Review findings on PR #7528

- AbortSignal: graphqlPost now accepts and propagates signal to fetch,
  warmup and mode-switch calls pass the caller's signal.
- GraphQL errors: parse response body for errors array on HTTP 200.
- Abort listener leak: store handler reference and removeEventListener
  on settle, instead of relying solely on { once: true }.
- Binary WS frames: decode Buffer/ArrayBuffer/Uint8Array to UTF-8.
- Test: add test for GraphQL error-in-200 detection.

Co-Authored-By: Claude <noreply@anthropic.com>

* fix: narrow ProtoField value before BigInt in serializeProtoFields

setBigUint64(0, BigInt(f.value)) failed tsc TS2345 because f.value's
union includes Uint8Array. Wire type 1 always carries a numeric value;
guard the Uint8Array case with a clear throw instead of coercing.

Co-Authored-By: Claude <noreply@anthropic.com>

* refactor: remove dead readTextResponse from muse-spark-web

Unused since the WebSocket migration dropped body-streaming reads. The
identically named live copy in blackbox-web.ts is untouched.

Co-Authored-By: Claude <noreply@anthropic.com>

* refactor: remove dead postMetaAiRequest from muse-spark-web

Replaced by the WebSocket send path; no remaining call sites.

Co-Authored-By: Claude <noreply@anthropic.com>

* refactor: remove dead buildHttpErrorResult/buildParsedErrorResult

Both were part of the retired GraphQL-POST error path; the WebSocket
path builds errors via errorResult directly. No remaining call sites.

Co-Authored-By: Claude <noreply@anthropic.com>

* test: nest connectionId overrides into credentials

Four tests passed connectionId at the top level of makeBaseInput, where
the spread never reached credentials.connectionId that execute reads --
so they silently ran against the default conn-test-1 instead of their
named ids. Add a withConnection helper and route them through it.

Co-Authored-By: Claude <noreply@anthropic.com>

* docs: document template fingerprint fields verified STATIC vs live capture

Live WS captures from two independent meta.ai accounts confirm the
64-hex session token, actor numeric ID, locale, and app ID are
app-level constants — identical in Meta's own client. No fingerprint
randomization warranted.

Co-Authored-By: Claude <noreply@anthropic.com>

* fix: address code review — NaN uniqueMsgId, varint truncation, cache eviction, empty WS 502

- uniqueMessageId: use Math.random() decimal suffix instead of
  crypto.randomUUID().slice(0,4) which produced NaN ~80% of the time
  (UUID hex chars like 'a'-'f' break Number()).
- encodeVarint: use BigInt arithmetic instead of >>> bitwise operators
  that truncated 41-bit Date.now() timestamps to 32 bits (lost minutes).
- submittedMs: use ?? instead of || so valid zero timestamps are accepted.
- Cache eviction: add evictContinuationIfNeeded on WS error path (was
  missing, letting stale conversation entries survive WS failures).
- Empty WS response: return 502 instead of 200 when WS closes with no
  content, matching the old parseMetaAiResponseText behavior.

Co-Authored-By: Claude <noreply@anthropic.com>

* chore(7528): keep .mergify.yml at release tip (maintainer CI config lands via its own PRs, not this provider fix)

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

---------

Co-authored-by: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com>
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Diego Rodrigues de Sa e Souza <diegosouza.pw@gmail.com>
This commit is contained in:
Ajeesh
2026-07-22 06:10:59 +05:30
committed by GitHub
parent 159873719c
commit 2b6e856f64
2 changed files with 808 additions and 463 deletions
Download Patch File Download Diff File Expand all files Collapse all files

779
open-sse/executors/muse-spark-web.ts
View File

@@ -1,23 +1,15 @@
import { createHash } from "node:crypto";
import { createHash, randomUUID } from "node:crypto";
import { Buffer } from "node:buffer";
import WebSocket from "ws";
import {
BaseExecutor,
mergeAbortSignals,
mergeUpstreamExtraHeaders,
type ExecuteInput,
} from "./base.ts";
import { FETCH_TIMEOUT_MS } from "../config/constants.ts";
import { BaseExecutor, mergeUpstreamExtraHeaders, type ExecuteInput } from "./base.ts";
import { getRotatingApiKey } from "../services/apiKeyRotator.ts";
import { prepareToolMessages, buildToolAwareResult } from "../translator/webTools.ts";
import {
normalizeSessionCookieHeader,
normalizeSessionCookieHeaders,
} from "@/lib/providers/webCookieAuth";
import {
type ParsedMetaAiResponse,
isRecord,
parseMetaAiResponseText,
} from "./muse-spark-web/response-parser.ts";
import { type ParsedMetaAiResponse, isRecord } from "./muse-spark-web/response-parser.ts";
const META_AI_GRAPHQL_API = "https://www.meta.ai/api/graphql";
// Meta rebranded the chat product from "Abra" to "Ecto"; the session cookie
@@ -32,12 +24,18 @@ const META_AI_DEFAULT_COOKIE = "ecto_1_sess";
// fails server-side validation with `Unknown type "RewriteOptionsInput"`.
// The new operation is a Subscription rather than a Mutation, but Meta's
// GraphQL endpoint still accepts it over POST and streams the response.
const META_AI_SEND_MESSAGE_DOC_ID = "29ae946c82d1f301196c6ca2226400b5";
const META_AI_WARMUP_DOC_ID = "e7f802582dbfed8e181b012e010993eb";
const META_AI_MODE_SWITCH_DOC_ID = "c32bbe999c48e64e855dc63177d5153f";
const META_WS_APP_ID = "1522763855472543";
const META_WS_APP_VERSION = "1.0.0";
const META_WS_AUTHTYPE = "15:0";
const META_WS_DGW_VERSION = "5";
const META_WS_DGW_UUID = "0";
const META_WS_TIER = "prod";
const META_WS_INTRO_FRAME_TYPE = 0x0f;
const META_WS_PROMPT_FRAME_TYPE = 0x0d;
const META_WS_PROMPT_FRAME_FLAG = 0x80;
const META_AI_ROOT_BRANCH_PATH = "0";
const META_AI_ENTRY_POINT = "KADABRA__CHAT__UNIFIED_INPUT_BAR";
const META_AI_FRIENDLY_NAME = "useEctoSendMessageSubscription";
const META_AI_REQUEST_ANALYTICS_TAGS = "graphservice";
const META_AI_ASBD_ID = "129477";
const META_AI_USER_AGENT =
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36";
const BASE62_ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
@@ -48,8 +46,8 @@ type MuseSparkModelInfo = {
};
const MODEL_MAP: Record<string, MuseSparkModelInfo> = {
"muse-spark": { mode: "mode_fast", isThinking: false },
"muse-spark-thinking": { mode: "mode_thinking", isThinking: true },
"muse-spark": { mode: "think_fast", isThinking: false },
"muse-spark-thinking": { mode: "think_hard", isThinking: true },
"muse-spark-contemplating": { mode: "think_hard", isThinking: true },
};
@@ -333,7 +331,7 @@ function buildMetaAiRequestBody(prompt: string, model: string, conversation: Con
const userUniqueMessageId = generateNumericMessageId();
return {
doc_id: META_AI_SEND_MESSAGE_DOC_ID,
doc_id: META_AI_WARMUP_DOC_ID,
variables: {
assistantMessageId: crypto.randomUUID(),
// `attachments` was removed from Meta's GraphQL schema (the
@@ -352,7 +350,7 @@ function buildMetaAiRequestBody(prompt: string, model: string, conversation: Con
currentBranchPath: conversation.branchPath,
developerOverridesForMessage: null,
devicePixelRatio: 1,
entryPoint: META_AI_ENTRY_POINT,
entryPoint: "KADABRA__CHAT__UNIFIED_INPUT_BAR",
imagineOperationRequest: null,
isNewConversation: conversation.isNewConversation,
mentions: null,
@@ -533,32 +531,6 @@ function buildErrorResponse(status: number, message: string, code?: string | nul
);
}
async function readTextResponse(
body: ReadableStream<Uint8Array>,
signal?: AbortSignal | null
): Promise<string> {
const reader = body.getReader();
const decoder = new TextDecoder();
let text = "";
try {
while (true) {
if (signal?.aborted) {
throw signal.reason ?? new DOMException("Aborted", "AbortError");
}
const { value, done } = await reader.read();
if (done) break;
text += decoder.decode(value, { stream: true });
}
text += decoder.decode();
return text;
} finally {
reader.releaseLock();
}
}
export function normalizeMetaAiCookieHeader(apiKey: string): string {
return normalizeSessionCookieHeader(apiKey, META_AI_DEFAULT_COOKIE);
}
@@ -598,9 +570,9 @@ function buildMetaAiHeaders(cookieHeader: string): Record<string, string> {
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": META_AI_USER_AGENT,
"X-ASBD-ID": META_AI_ASBD_ID,
"X-FB-Friendly-Name": META_AI_FRIENDLY_NAME,
"X-FB-Request-Analytics-Tags": META_AI_REQUEST_ANALYTICS_TAGS,
"X-ASBD-ID": "129477",
"X-FB-Friendly-Name": "useEctoSendMessageSubscription",
"X-FB-Request-Analytics-Tags": "graphservice",
};
}
@@ -640,6 +612,508 @@ function getOpenAiMessages(body: unknown): Array<Record<string, unknown>> | null
return messages as Array<Record<string, unknown>>;
}
// ─── Protobuf WS templates ──────────────────────────────────────────────────────
// Base64-encoded protobuf templates captured from Meta AI web client.
// These are mutated at specific field paths to inject conversation-id,
// prompt text, timestamps, and message IDs per conversation.
//
// VERIFIED against live meta.ai WS captures from TWO independent accounts
// (2026-07-19). The following fields are confirmed STATIC (app-level
// constants sent by Meta's own client, not per-user secrets):
// - 64-hex session token (e2b88f98...)
// - Actor numeric ID (867051314767696)
// - Locale (en-US)
// - App ID (1522763855472543)
// The only user-variable field is the timezone (system TZ), which is
// low-signal for anti-fraud. No fingerprint randomization is warranted.
const META_WS_HOME_TEMPLATE_B64 =
"CrYGCsQDCiBLQURBQlJBX19IT01FX19VTklGSUVEX0lOUFVUX0JBUhIQMTUyMjc2Mzg1NTQ3MjU0MyInNWE1Yi04ZDRlLWYwNTQtOTllZi1iMmRlLWRiMDItMGQwNS01MmM3KigqJgokOGYxMjliMjUtYzNlMC00NzNiLWFlNzktNWViM2YyNGU1NjRjMAU6C0hVTUFOX0FHRU5UQiIKDzg2NzA1MTMxNDc2NzY5NhIPODY3MDUxMzE0NzY3Njk2UgVFQ1RPMVoRQWJyYSBXZWIgTWFpbiBLZXliCRoDCOgHIgIIAWoITWFjIE9TIFhyCnVzZXJfaW5wdXR6dU1vemlsbGEvNS4wIChNYWNpbnRvc2g7IEludGVsIE1hYyBPUyBYIDEwXzE1XzcpIEFwcGxlV2ViS2l0LzUzNy4zNiAoS0hUTUwsIGxpa2UgR2Vja28pIENocm9tZS8xNDYuMC4wLjAgU2FmYXJpLzUzNy4zNoIBC2Rlc2t0b3Bfd2VimgFHCkBlMmI4OGY5ODQ2Mzc5Y2JjMjY5NjBmYTNhZTFkMjIyMDFkZmIxOWRmNzg5MGFlNmEzYWM4YTI4ODcwYmFjNjgyFQAAAEASFAi4w6XTk4/yARC4w6XTk4/yARgCGgIgASIAKg4Ix6D+ldkzGJ6g/pXZMzIkZWU3YTM1ZWItZGY4Yy00NzkzLWExYzAtMTBhZTQxNGY1ZTZlOgBKBxIFZW4tVVNScgokNTYwN2Y0YzAtYjljZi00ZjZlLWJlYTYtZTc2N2E1OGJhMjhlGiRlMDliN2FhMC1jYzYwLTQyYTktYjk2OS00YzY1YjViZGZlNGIiJDhmMTI5YjI1LWMzZTAtNDczYi1hZTc5LTVlYjNmMjRlNTY0Y3oRIg9BbWVyaWNhL0NoaWNhZ2+CAQOwAQGSAQwKBnN0b2NrcxICCAGSAQ0KB3dlYXRoZXISAggBkgEkCh5tZXRhX2tub3dsZWRnZV9zZWFyY2hfY2Fyb3VzZWwSAggBkgEiChxtZXRhX2NhdGFsb2dfc2VhcmNoX2Nhcm91c2VsEgIIAZIBEwoNbWVkaWFfZ2FsbGVyeRICCAGiAQEDEpIBCmEKJGFiOWRkNzg5LWRlOGQtNDc5MS05ODE1LWI5YjBmMTU1MDdiNBI3CiQ4ZjEyOWIyNS1jM2UwLTQ3M2ItYWU3OS01ZWIzZjI0ZTU2NGMQyKD+ldkzGKbcxozB/KuyZygBEihIZWxsbyB0aGlzIGlzIGFub3RoZXIgdGVzdCBvZiB5b3VyIHBvd2VyIgMKATA=";
const META_WS_CHAT_TEMPLATE_B64 =
"CrIGCsADCiBLQURBQlJBX19DSEFUX19VTklGSUVEX0lOUFVUX0JBUhIQMTUyMjc2Mzg1NTQ3MjU0MyInNWE1Yi04ZDRlLWYwNTQtOTllZi1iMmRlLWRiMDItMGQwNS01MmM3KigqJgokYjA4Mzg1YTYtNWE1My00ZjE0LTk2NmUtMzQ3ZjI4MDg4NDU0MAU6C0hVTUFOX0FHRU5UQiIKDzg2NzA1MTMxNDc2NzY5NhIPODY3MDUxMzE0NzY3Njk2UgVFQ1RPMVoRQWJyYSBXZWIgTWFpbiBLZXliBRoDCOgHaghNYWMgT1MgWHIKdXNlcl9pbnB1dHp1TW96aWxsYS81LjAgKE1hY2ludG9zaDsgSW50ZWwgTWFjIE9TIFggMTBfMTVfNykgQXBwbGVXZWJLaXQvNTM3LjM2IChLSFRNTCwgbGlrZSBHZWNrbykgQ2hyb21lLzE0Ni4wLjAuMCBTYWZhcmkvNTM3LjM2ggELZGVza3RvcF93ZWKaAUcKQGUyYjg4Zjk4NDYzNzljYmMyNjk2MGZhM2FlMWQyMjIwMWRmYjE5ZGY3ODkwYWU2YTNhYzhhMjg4NzBiYWM2ODIVAAAAQBIUCLjDpdOTj/IBELjDpdOTj/IBGAIaAiABIgAqDgikgvuW2TMYoYL7ltkzMiRjNmI1ZDI2MS02NjI0LTQ5YWYtOTBjNy0wOWI0NWMwYTZiZWY6AEoHEgVlbi1VU1JyCiQxZDNjZGQzYy1jYTFhLTRlMDItODk1My1kZTBiYTM0NzI5ODkaJDcxODNhMzM0LTFiNWEtNGQyNi1iMjcxLWJjY2Y1NDY2NmJiZiIkYjA4Mzg1YTYtNWE1My00ZjE0LTk2NmUtMzQ3ZjI4MDg4NDU0ehEiD0FtZXJpY2EvQ2hpY2Fnb4IBA7ABAZIBDAoGc3RvY2tzEgIIAZIBDQoHd2VhdGhlchICCAGSASQKHm1ldGFfa25vd2xlZGdlX3NlYXJjaF9jYXJvdXNlbBICCAGSASIKHG1ldGFfY2F0YWxvZ19zZWFyY2hfY2Fyb3VzZWwSAggBkgETCg1tZWRpYV9nYWxsZXJ5EgIIAaIBAQMSlgEKfAokMTc4MDVmYjEtOTY3Zi00YmYyLTlmMjctOWRhYmRhMzYyMTJkEjcKJGIwODM4NWE2LTVhNTMtNGYxNC05NjZlLTM0N2YyODA4ODQ1NBCkgvuW2TMYxN23xoT2rbJnIhtlLjAwcHlKMUtxa3BHTmg5Sk9oWElNdnJRWlYSEWZvbGxvdyB1cCBwcm9iZSAyIgMKATI=";
// ─── Proto helpers ─────────────────────────────────────────────────────────────
type ProtoField = {
number: number;
wireType: number;
value: Uint8Array | number | bigint;
};
function encodeVarint(value: number): Uint8Array {
// Use BigInt arithmetic to avoid 32-bit truncation from bitwise operators.
let v = BigInt(value);
const out: number[] = [];
while (v >= 0x80n) {
out.push(Number((v & 0x7fn) | 0x80n));
v >>= 7n;
}
out.push(Number(v & 0x7fn));
return new Uint8Array(out);
}
function decodeVarint(data: Uint8Array, offset: number): [number, number] {
let shift = 0;
let value = 0;
let off = offset;
while (true) {
const byte = data[off++];
value |= (byte & 0x7f) << shift;
if (!(byte & 0x80)) return [value >>> 0, off];
shift += 7;
if (shift > 63) throw new Error("Varint too long");
}
}
function parseProtoFields(data: Uint8Array): ProtoField[] {
const fields: ProtoField[] = [];
let offset = 0;
while (offset < data.length) {
const [tag, next] = decodeVarint(data, offset);
offset = next;
const number = tag >> 3;
const wireType = tag & 0x07;
if (wireType === 0) {
const [value, n] = decodeVarint(data, offset);
offset = n;
fields.push({ number, wireType, value });
} else if (wireType === 1) {
const view = new DataView(data.buffer, data.byteOffset + offset, 8);
fields.push({ number, wireType, value: view.getBigUint64(0, true) });
offset += 8;
} else if (wireType === 2) {
const [len, n] = decodeVarint(data, offset);
offset = n;
fields.push({ number, wireType, value: data.slice(offset, offset + len) });
offset += len;
} else if (wireType === 5) {
const view = new DataView(data.buffer, data.byteOffset + offset, 4);
fields.push({ number, wireType, value: view.getUint32(0, true) });
offset += 4;
} else {
throw new Error(`Unsupported wire type: ${wireType}`);
}
}
return fields;
}
function serializeProtoFields(fields: ProtoField[]): Uint8Array {
const parts: Uint8Array[] = [];
for (const f of fields) {
const tag = (f.number << 3) | f.wireType;
parts.push(encodeVarint(tag));
if (f.wireType === 0) {
parts.push(encodeVarint(Number(f.value)));
} else if (f.wireType === 1) {
const buf = new Uint8Array(8);
if (f.value instanceof Uint8Array) {
throw new Error(
`serializeProtoFields: wire type 1 field ${f.number} has non-numeric value`
);
}
new DataView(buf.buffer).setBigUint64(0, BigInt(f.value), true);
parts.push(buf);
} else if (f.wireType === 2) {
const raw =
f.value instanceof Uint8Array ? f.value : new TextEncoder().encode(String(f.value));
parts.push(encodeVarint(raw.length));
parts.push(raw);
} else if (f.wireType === 5) {
const buf = new Uint8Array(4);
new DataView(buf.buffer).setUint32(0, Number(f.value), true);
parts.push(buf);
}
}
const total = parts.reduce((s, p) => s + p.length, 0);
const result = new Uint8Array(total);
let offset = 0;
for (const p of parts) {
result.set(p, offset);
offset += p.length;
}
return result;
}
function findProtoField(fields: ProtoField[], number: number): ProtoField | undefined {
return fields.find((f) => f.number === number);
}
function traverseAndMutate(
fields: ProtoField[],
path: number[],
mutator: (field: ProtoField) => void
): boolean {
if (path.length === 0) return false;
const field = findProtoField(fields, path[0]);
if (!field || !(field.value instanceof Uint8Array)) return false;
if (path.length === 1) {
mutator(field);
return true;
}
const nested = parseProtoFields(field.value);
if (traverseAndMutate(nested, path.slice(1), mutator)) {
field.value = serializeProtoFields(nested);
return true;
}
return false;
}
// ─── WS frame builders ─────────────────────────────────────────────────────────
function writeU24Le(value: number, arr: Uint8Array, offset: number): void {
arr[offset] = value & 0xff;
arr[offset + 1] = (value >> 8) & 0xff;
arr[offset + 2] = (value >> 16) & 0xff;
}
function buildWsIntroFrame(conversationId: string): Uint8Array {
const payload = new TextEncoder().encode(
JSON.stringify({
"x-dgw-app-x-ecto-conversation-id": conversationId,
"x-dgw-app-client-payload-type": "PROTO_INSIDE_JSON",
})
);
const header = new Uint8Array(6);
header[0] = META_WS_INTRO_FRAME_TYPE;
header[1] = 0;
header[2] = 0;
writeU24Le(payload.length, header, 3);
const result = new Uint8Array(header.length + payload.length);
result.set(header);
result.set(payload, header.length);
return result;
}
function buildWsPromptFrame(
prompt: string,
conversationId: string,
opts: {
templateB64: string;
requestId?: string;
userMessageId?: string;
submittedMs?: number;
uniqueMessageId?: number;
subSessionIdx?: number;
messageSeq?: number;
}
): Uint8Array {
const requestId = opts.requestId || crypto.randomUUID();
const userMessageId = opts.userMessageId || crypto.randomUUID();
const submittedMs = opts.submittedMs ?? Date.now();
const uniqueMessageId =
opts.uniqueMessageId ??
Number(`${submittedMs}${String(Math.floor(Math.random() * 10000)).padStart(4, "0")}`);
const raw = Buffer.from(opts.templateB64, "base64");
const protoFields = parseProtoFields(raw);
// Patch conversationId at [1,1,5]
traverseAndMutate(protoFields, [1, 1], (f) => {
const nested = parseProtoFields(f.value instanceof Uint8Array ? f.value : new Uint8Array());
const field5 = findProtoField(nested, 5);
if (field5) field5.value = new TextEncoder().encode(conversationId);
f.value = serializeProtoFields(nested);
});
// Patch userMessageId at [2,1,1]
traverseAndMutate(protoFields, [2, 1], (f) => {
const nested = parseProtoFields(f.value instanceof Uint8Array ? f.value : new Uint8Array());
const field1 = findProtoField(nested, 1);
if (field1) field1.value = new TextEncoder().encode(userMessageId);
f.value = serializeProtoFields(nested);
});
// Patch convId + timestamps at [2,1,2]
traverseAndMutate(protoFields, [2, 1, 2], (f) => {
const nested = parseProtoFields(f.value instanceof Uint8Array ? f.value : new Uint8Array());
const f1 = findProtoField(nested, 1);
const f2 = findProtoField(nested, 2);
const f3 = findProtoField(nested, 3);
if (f1) f1.value = new TextEncoder().encode(conversationId);
if (f2) f2.value = submittedMs;
if (f3) f3.value = uniqueMessageId;
f.value = serializeProtoFields(nested);
});
// Patch prompt text at [2,2]
traverseAndMutate(protoFields, [2], (f) => {
const nested = parseProtoFields(f.value instanceof Uint8Array ? f.value : new Uint8Array());
const field2 = findProtoField(nested, 2);
if (field2) field2.value = new TextEncoder().encode(prompt);
f.value = serializeProtoFields(nested);
});
// Patch timestamps at [1,5]
traverseAndMutate(protoFields, [1, 5], (f) => {
const nested = parseProtoFields(f.value instanceof Uint8Array ? f.value : new Uint8Array());
const f1 = findProtoField(nested, 1);
const f3 = findProtoField(nested, 3);
if (f1) f1.value = submittedMs + 1;
if (f3) f3.value = submittedMs;
f.value = serializeProtoFields(nested);
});
// Patch requestId at [1,6]
traverseAndMutate(protoFields, [1], (f) => {
const nested = parseProtoFields(f.value instanceof Uint8Array ? f.value : new Uint8Array());
const field6 = findProtoField(nested, 6);
if (field6) field6.value = new TextEncoder().encode(requestId);
f.value = serializeProtoFields(nested);
});
// Patch conversationId at [1,10,4]
traverseAndMutate(protoFields, [1, 10], (f) => {
const nested = parseProtoFields(f.value instanceof Uint8Array ? f.value : new Uint8Array());
const field4 = findProtoField(nested, 4);
if (field4) field4.value = new TextEncoder().encode(conversationId);
f.value = serializeProtoFields(nested);
});
const updatedB64 = Buffer.from(serializeProtoFields(protoFields)).toString("base64");
const outer = JSON.stringify({ "req-id": requestId, payload: updatedB64 });
const inner = new TextEncoder().encode(outer);
const subSessionIdx = opts.subSessionIdx || 0;
const messageSeq = opts.messageSeq || 0;
const msgBody = new Uint8Array(2 + inner.length);
msgBody[0] = messageSeq;
msgBody[1] = META_WS_PROMPT_FRAME_FLAG;
msgBody.set(inner, 2);
const header = new Uint8Array(6);
header[0] = META_WS_PROMPT_FRAME_TYPE;
header[1] = subSessionIdx & 0xff;
header[2] = (subSessionIdx >> 8) & 0xff;
writeU24Le(msgBody.length, header, 3);
const frame = new Uint8Array(header.length + msgBody.length);
frame.set(header);
frame.set(msgBody, header.length);
return frame;
}
// ─── WS URL builder + GraphQL helper + b64 helpers ─────────────────────────────
function buildWsUrl(authorization: string, requestId: string): string {
const params = new URLSearchParams({
"x-dgw-appid": META_WS_APP_ID,
"x-dgw-appversion": META_WS_APP_VERSION,
"x-dgw-authtype": META_WS_AUTHTYPE,
"x-dgw-version": META_WS_DGW_VERSION,
"x-dgw-uuid": META_WS_DGW_UUID,
"x-dgw-tier": META_WS_TIER,
Authorization: authorization,
"x-dgw-app-origin": "meta.ai",
"x-dgw-app-clippy-request-id": requestId,
"x-dgw-app-clippy-async": "true",
});
return `wss://gateway.meta.ai/ws/clippy?${params.toString()}`;
}
type GraphqlResult = { ok: true } | { ok: false; error: string };
async function graphqlPost(
docId: string,
variables: Record<string, unknown>,
cookieHeader: string,
label: string,
signal?: AbortSignal | null
): Promise<GraphqlResult> {
try {
const response = await fetch(META_AI_GRAPHQL_API, {
method: "POST",
headers: {
"Content-Type": "application/json",
Accept: "multipart/mixed, application/json",
Cookie: cookieHeader,
"User-Agent": META_AI_USER_AGENT,
Origin: "https://meta.ai",
},
body: JSON.stringify({ doc_id: docId, variables }),
signal: signal ?? undefined,
});
if (!response.ok) return { ok: false, error: `${label} failed: HTTP ${response.status}` };
// GraphQL often returns errors in the body with HTTP 200 — parse them.
const text = await response.text();
try {
const json = JSON.parse(text);
if (json && Array.isArray(json.errors) && json.errors.length > 0) {
const msg = json.errors[0]?.message || "Unknown GraphQL error";
return { ok: false, error: `${label} failed: ${msg}` };
}
} catch {
// Response wasn't JSON or had no errors — treat as success.
}
return { ok: true };
} catch (err) {
return {
ok: false,
error: `${label} fetch failed: ${err instanceof Error ? err.message : String(err)}`,
};
}
}
// ─── WS response parser ────────────────────────────────────────────────────────
type WsResponseEvent = {
type: "full" | "patch";
response?: { sections?: Array<{ view_model?: { primitive?: { text?: string } } }> };
operations?: Array<{ op?: string; path?: string; value?: string }>;
};
function parseWsResponseEvents(payload: string): WsResponseEvent[] {
const events: WsResponseEvent[] = [];
let start: number | null = null;
let depth = 0;
let inString = false;
let escape = false;
for (let i = 0; i < payload.length; i++) {
const ch = payload[i];
if (start === null) {
if (ch === "{") {
start = i;
depth = 1;
inString = false;
escape = false;
}
continue;
}
if (inString) {
if (escape) {
escape = false;
} else if (ch === "\\") {
escape = true;
} else if (ch === '"') {
inString = false;
}
continue;
}
if (ch === '"') {
inString = true;
} else if (ch === "{") {
depth++;
} else if (ch === "}") {
depth--;
if (depth === 0 && start !== null) {
try {
events.push(JSON.parse(payload.slice(start, i + 1)));
} catch {
/* skip */
}
start = null;
}
}
}
return events;
}
type WsChatResult = {
content: string;
deltas: string[];
error?: string;
};
// ─── WebSocket chat function + test hook ────────────────────────────────────────
let WebSocketCtor: typeof WebSocket = WebSocket;
export function __setMuseSparkWebSocketForTesting(ctor: typeof WebSocket): () => void {
const previous = WebSocketCtor;
WebSocketCtor = ctor;
return () => {
WebSocketCtor = previous;
};
}
async function wsChat(
prompt: string,
conversationId: string,
authorization: string,
cookieHeader: string,
templateB64: string,
signal?: AbortSignal | null
): Promise<WsChatResult> {
const requestId = crypto.randomUUID();
const wsUrl = buildWsUrl(authorization, requestId);
return new Promise((resolve) => {
const ws = new WebSocketCtor(wsUrl, {
headers: {
Cookie: cookieHeader,
"User-Agent": META_AI_USER_AGENT,
Origin: "https://meta.ai",
},
});
let settled = false;
let accumulatedText = "";
const contentDeltas: string[] = [];
let timeout: ReturnType<typeof setTimeout> | null = null;
let abortHandler: (() => void) | null = null;
const finish = (result: WsChatResult) => {
if (settled) return;
settled = true;
if (timeout) clearTimeout(timeout);
if (signal && abortHandler) signal.removeEventListener("abort", abortHandler);
try {
ws.close();
} catch {
/* ignore */
}
resolve(result);
};
const fail = (error: string) => finish({ content: "", deltas: [], error });
timeout = setTimeout(() => fail("Meta AI WebSocket timed out"), 30000);
abortHandler = () => fail("Request aborted");
signal?.addEventListener("abort", abortHandler, { once: true });
ws.onopen = () => {
ws.send(buildWsIntroFrame(conversationId));
ws.send(buildWsPromptFrame(prompt, conversationId, { templateB64, requestId }));
};
ws.onmessage = (event) => {
let raw = "";
if (typeof event.data === "string") {
raw = event.data;
} else if (Buffer.isBuffer(event.data)) {
raw = event.data.toString("utf-8");
} else if (event.data instanceof ArrayBuffer || event.data instanceof Uint8Array) {
raw = new TextDecoder().decode(event.data);
}
if (!raw) return;
const events = parseWsResponseEvents(raw);
for (const evt of events) {
if (evt.type === "full") {
const sections = evt.response?.sections || [];
for (const section of sections) {
const text = section?.view_model?.primitive?.text || "";
if (text && text !== accumulatedText) {
const delta = accumulatedText ? text.slice(accumulatedText.length) || text : text;
if (delta) contentDeltas.push(delta);
accumulatedText = text;
}
}
} else if (evt.type === "patch") {
const operations = evt.operations || [];
for (const op of operations) {
if (
op.op === "delta" &&
op.path === "/sections/0/view_model/primitive/text" &&
typeof op.value === "string"
) {
contentDeltas.push(op.value);
accumulatedText += op.value;
}
}
}
}
};
ws.onerror = () => fail("Meta AI WebSocket connection error");
ws.onclose = () => {
if (settled) return;
finish({ content: accumulatedText, deltas: contentDeltas });
};
});
}
function getContinuationCacheKey(
parsedHistory: ParsedHistory,
credentials: ExecuteInput["credentials"],
@@ -685,78 +1159,6 @@ function evictContinuationIfNeeded(
}
}
async function postMetaAiRequest(
headers: Record<string, string>,
transformedBody: unknown,
signal: AbortSignal,
log: ExecuteInput["log"]
): Promise<{ ok: true; response: Response } | { ok: false; result: MuseSparkExecuteResult }> {
try {
const response = await fetch(META_AI_GRAPHQL_API, {
method: "POST",
headers,
body: JSON.stringify(transformedBody),
signal,
});
return { ok: true, response };
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
log?.error?.("MUSE-SPARK-WEB", `Fetch failed: ${message}`);
return {
ok: false,
result: errorResult(
502,
`Meta AI connection failed: ${message}`,
"meta_ai_fetch_failed",
headers,
transformedBody
),
};
}
}
function buildHttpErrorResult(
upstreamResponse: Response,
headers: Record<string, string>,
transformedBody: unknown,
cached: CachedConversation | null,
cacheKey: string | null
): MuseSparkExecuteResult {
evictContinuationIfNeeded(cached, cacheKey);
let message = `Meta AI returned HTTP ${upstreamResponse.status}`;
if (upstreamResponse.status === 401 || upstreamResponse.status === 403) {
message = "Meta AI auth failed — your meta.ai ecto_1_sess cookie may be missing or expired.";
} else if (upstreamResponse.status === 429) {
message = "Meta AI rate limited the session. Wait a moment and retry.";
}
return errorResult(
upstreamResponse.status,
message,
`HTTP_${upstreamResponse.status}`,
headers,
transformedBody
);
}
function buildParsedErrorResult(
parsed: ParsedMetaAiResponse,
headers: Record<string, string>,
transformedBody: unknown,
cached: CachedConversation | null,
cacheKey: string | null
): MuseSparkExecuteResult {
evictContinuationIfNeeded(cached, cacheKey);
return errorResult(
parsed.status,
parsed.errorMessage || "Meta AI returned an unknown error",
parsed.errorCode || "meta_ai_unknown_error",
headers,
transformedBody
);
}
function rememberAssistantTurn(
parsed: ParsedMetaAiResponse,
credentials: ExecuteInput["credentials"],
@@ -857,6 +1259,30 @@ export class MuseSparkWebExecutor extends BaseExecutor {
return errorResult(400, "Empty query after processing messages", "invalid_request", {}, body);
}
// Extract the WebSocket auth token (ecto1:...) from provider-specific data
// or from the apiKey field itself (user can paste both in the cookie field).
let authorization: string;
if (
typeof credentials.providerSpecificData?.authorization === "string" &&
credentials.providerSpecificData.authorization
) {
authorization = credentials.providerSpecificData.authorization.trim();
} else if (typeof credentials.apiKey === "string") {
const match = credentials.apiKey.match(/ecto1:[^\s;]+/i);
authorization = match ? match[0].trim() : "";
} else {
authorization = "";
}
if (!authorization) {
return errorResult(
400,
"Missing Authorization for Meta AI WebSocket — your cookie must include an ecto1:... auth token.",
"missing_authorization",
{},
body
);
}
// Look up a prior meta.ai conversation we created for this caller +
// model + chat thread. The lookup key is the connection + model + the
// SHA-256 of the normalized history prefix ending at the last assistant
@@ -875,58 +1301,87 @@ export class MuseSparkWebExecutor extends BaseExecutor {
const conversationContext = getConversationContext(cached);
const prompt = cached ? parsedHistory.latestUserContent : parsedHistory.foldedPrompt;
const modelInfo = getMuseSparkModelInfo(model);
const transformedBody = buildMetaAiRequestBody(prompt, model, conversationContext);
const cookieHeader = selectMetaAiCookieHeader(credentials);
const modelInfo = getMuseSparkModelInfo(model);
const templateB64 = cached ? META_WS_CHAT_TEMPLATE_B64 : META_WS_HOME_TEMPLATE_B64;
// Step 1: GraphQL warmup initialises the conversation on Meta's side
const warmupResult = await graphqlPost(
META_AI_WARMUP_DOC_ID,
{ conversationId: conversationContext.conversationId },
cookieHeader,
"Warmup",
signal
);
if (!warmupResult.ok) {
evictContinuationIfNeeded(cached, continuationCacheKey);
log?.error?.("MUSE-SPARK-WEB", `Warmup failed: ${warmupResult.error}`);
return errorResult(502, warmupResult.error, "meta_ai_warmup_failed", {}, body);
}
// Step 2: GraphQL mode switch sets the conversation's reasoning level
const modeResult = await graphqlPost(
META_AI_MODE_SWITCH_DOC_ID,
{ input: { conversationId: conversationContext.conversationId, mode: modelInfo.mode } },
cookieHeader,
"Mode switch",
signal
);
if (!modeResult.ok) {
evictContinuationIfNeeded(cached, continuationCacheKey);
log?.error?.("MUSE-SPARK-WEB", `Mode switch failed: ${modeResult.error}`);
return errorResult(502, modeResult.error, "meta_ai_mode_switch_failed", {}, body);
}
// Step 3: Send message via WebSocket
const wsResult = await wsChat(
prompt,
conversationContext.conversationId,
authorization,
cookieHeader,
templateB64,
signal
);
const headers = buildMetaAiHeaders(cookieHeader);
mergeUpstreamExtraHeaders(headers, upstreamExtraHeaders);
const timeoutSignal = AbortSignal.timeout(FETCH_TIMEOUT_MS);
const combinedSignal = signal ? mergeAbortSignals(signal, timeoutSignal) : timeoutSignal;
const fetchResult = await postMetaAiRequest(headers, transformedBody, combinedSignal, log);
if (!fetchResult.ok) {
const err = fetchResult as { ok: false; result: MuseSparkExecuteResult };
return err.result;
if (wsResult.error) {
evictContinuationIfNeeded(cached, continuationCacheKey);
log?.error?.("MUSE-SPARK-WEB", `WS error: ${wsResult.error}`);
const lower = wsResult.error.toLowerCase();
const status = /auth|authorization|401/.test(lower) ? 401 : 502;
return errorResult(status, wsResult.error, "meta_ai_ws_error", headers, body);
}
const upstreamResponse = fetchResult.response;
if (!upstreamResponse.ok) {
return buildHttpErrorResult(
upstreamResponse,
headers,
transformedBody,
cached,
continuationCacheKey
);
}
const content = wsResult.content || "";
if (!upstreamResponse.body) {
// Empty WS response is an upstream failure, not a successful empty completion.
if (!content && !wsResult.deltas.length) {
evictContinuationIfNeeded(cached, continuationCacheKey);
log?.error?.("MUSE-SPARK-WEB", "WS returned empty response");
return errorResult(
502,
"Meta AI returned an empty response body",
"meta_ai_empty_body",
"Meta AI returned no assistant content",
"meta_ai_empty_response",
headers,
transformedBody
body
);
}
const responseText = await readTextResponse(upstreamResponse.body, signal);
const parsed = parseMetaAiResponseText(responseText, modelInfo.isThinking);
if (parsed.status !== 200 || parsed.errorMessage) {
return buildParsedErrorResult(parsed, headers, transformedBody, cached, continuationCacheKey);
const deltas = wsResult.deltas.length > 0 ? wsResult.deltas : [content];
const parsed = {
content,
deltas,
reasoningContent: "",
reasoningDeltas: [] as string[],
errorCode: null as string | null,
errorMessage: null as string | null,
status: 200,
};
if (content) {
rememberAssistantTurn(parsed, credentials, model, parsedHistory, conversationContext);
}
rememberAssistantTurn(parsed, credentials, model, parsedHistory, conversationContext);
return buildSuccessResult(
parsed,
stream,
model,
headers,
transformedBody,
hasTools,
requestedTools
);
return buildSuccessResult(parsed, stream, model, headers, body, hasTools, requestedTools);
}
}

492
tests/unit/muse-spark-web-continuation.test.ts
View File

@@ -1,348 +1,238 @@
import test from "node:test";
import assert from "node:assert/strict";
import {
MuseSparkWebExecutor,
__resetMuseSparkConversationCacheForTesting,
__setMuseSparkWebSocketForTesting,
} from "../../open-sse/executors/muse-spark-web.ts";
import { WebSocket } from "ws";
// Canned Meta AI response shape. parseMetaAiResponseText accepts either a
// plain JSON body or an SSE stream of `data: <json>` frames; we send a plain
// JSON body since the assertions don't care about delta structure.
function metaAiSseResponse(content: string): Response {
const body = JSON.stringify({
data: {
sendMessageStream: {
__typename: "AssistantMessage",
content,
},
},
});
return new Response(body, {
status: 200,
headers: { "Content-Type": "application/json" },
});
}
// ─── Mock WebSocket ──────────────────────────────────────────────────────────
type CapturedRequest = { url: string; init: RequestInit | undefined; body: unknown };
type MockWsMessage = { data: string };
function captureFetch(reply: () => Response): {
fetchFn: typeof fetch;
captured: CapturedRequest[];
} {
const captured: CapturedRequest[] = [];
const fetchFn: typeof fetch = async (input, init) => {
let body: unknown = undefined;
if (init?.body && typeof init.body === "string") {
try {
body = JSON.parse(init.body);
} catch {
body = init.body;
}
class MockWebSocket {
static instances: MockWebSocket[] = [];
onopen: (() => void) | null = null;
onmessage: ((evt: MockWsMessage) => void) | null = null;
onclose: (() => void) | null = null;
onerror: ((evt: Error) => void) | null = null;
readyState = WebSocket.CONNECTING;
sentData: (Uint8Array | string)[] = [];
url: string;
constructor(url: string) {
this.url = url;
MockWebSocket.instances.push(this);
setTimeout(() => {
this.readyState = WebSocket.OPEN;
this.onopen?.();
}, 0);
}
send(data: Uint8Array | string) {
this.sentData.push(data);
// When a prompt frame (type 0x0d) is sent, simulate a response + close
if (data instanceof Uint8Array && data.length > 0 && data[0] === 0x0d) {
setTimeout(() => {
this.onmessage?.({
data: JSON.stringify({
type: "full",
response: {
sections: [{ view_model: { primitive: { text: "pong" } } }],
},
}),
});
setTimeout(() => this.close(), 5);
}, 5);
}
captured.push({
url: typeof input === "string" ? input : (input as URL).toString(),
init,
body,
});
return reply();
};
return { fetchFn, captured };
}
close() {
this.readyState = WebSocket.CLOSED;
this.onclose?.();
}
}
function executeInputs(messages: Array<{ role: string; content: string }>) {
type ExecuteParams = Parameters<MuseSparkWebExecutor["execute"]>[0];
function makeBaseInput(overrides?: Partial<ExecuteParams>): ExecuteParams {
return {
model: "muse-spark",
body: { messages },
body: { messages: [{ role: "user", content: "ping" }] },
stream: false,
credentials: { apiKey: "abra_sess=foo", connectionId: "conn-test-1" },
credentials: {
apiKey: "ecto_1_sess=test123",
connectionId: "conn-test-1",
providerSpecificData: { authorization: "ecto1:test-auth-token" },
},
signal: null,
log: null,
upstreamExtraHeaders: undefined,
} as Parameters<MuseSparkWebExecutor["execute"]>[0];
...overrides,
} as ExecuteParams;
}
test("muse-spark-web: first turn opens a new meta.ai conversation", async () => {
function withConnection(connectionId: string, overrides?: Partial<ExecuteParams>): ExecuteParams {
return makeBaseInput({
credentials: {
apiKey: "ecto_1_sess=test123",
connectionId,
providerSpecificData: { authorization: "ecto1:test-auth-token" },
},
...overrides,
} as Partial<ExecuteParams>);
}
test("makeBaseInput nests connectionId override into credentials", () => {
const input = makeBaseInput({
credentials: { connectionId: "conn-distinct" },
} as Partial<ExecuteParams>);
assert.equal((input.credentials as { connectionId?: string }).connectionId, "conn-distinct");
});
// ─── Test 1: New conversation sends via WebSocket ────────────────────────────
test("muse-spark-web: new conversation sends via WebSocket", async () => {
__resetMuseSparkConversationCacheForTesting();
MockWebSocket.instances = [];
const executor = new MuseSparkWebExecutor();
const original = globalThis.fetch;
const { fetchFn, captured } = captureFetch(() => metaAiSseResponse("pong"));
globalThis.fetch = fetchFn;
const originalFetch = globalThis.fetch;
globalThis.fetch = async () => new Response("{}", { status: 200 });
const restore = __setMuseSparkWebSocketForTesting(MockWebSocket as unknown as typeof WebSocket);
try {
const result = await executor.execute(executeInputs([{ role: "user", content: "ping" }]));
assert.equal(captured.length, 1, "exactly one upstream call");
const sentVars = (captured[0].body as { variables: Record<string, unknown> }).variables;
assert.equal(sentVars.isNewConversation, true, "first turn → isNewConversation: true");
assert.equal(sentVars.content, "ping", "first turn → bare user content");
assert.match(String(sentVars.conversationId), /^c\./, "fresh meta.ai conversation id");
const result = await executor.execute(makeBaseInput());
assert.equal(MockWebSocket.instances.length, 1, "one WebSocket was created");
const ws = MockWebSocket.instances[0];
assert.ok(ws.sentData.length >= 1, "at least one frame was sent");
// First frame should be intro (type 0x0f)
const firstFrame = ws.sentData[0];
assert.ok(firstFrame instanceof Uint8Array, "first frame is binary");
assert.equal(firstFrame[0], 0x0f, "first frame is intro frame");
// Second frame should be prompt (type 0x0d)
if (ws.sentData.length >= 2) {
const secondFrame = ws.sentData[1];
assert.ok(secondFrame instanceof Uint8Array, "second frame is binary");
assert.equal(secondFrame[0], 0x0d, "second frame is prompt frame");
}
// Should get a 200 response with default text when WS returns nothing
assert.equal(result.response.status, 200);
} finally {
globalThis.fetch = original;
globalThis.fetch = originalFetch;
restore();
}
});
test("muse-spark-web: follow-up turn continues the cached conversation", async () => {
// ─── Test 2: Follow-up turn reuses conversation via WebSocket ────────────────
test("muse-spark-web: follow-up turn reuses conversation via WebSocket", async () => {
__resetMuseSparkConversationCacheForTesting();
MockWebSocket.instances = [];
const executor = new MuseSparkWebExecutor();
const original = globalThis.fetch;
let nthReply = 0;
const { fetchFn, captured } = captureFetch(() =>
metaAiSseResponse(nthReply++ === 0 ? "pong" : "pong-again")
);
globalThis.fetch = fetchFn;
const originalFetch = globalThis.fetch;
globalThis.fetch = async () => new Response("{}", { status: 200 });
const restore = __setMuseSparkWebSocketForTesting(MockWebSocket as unknown as typeof WebSocket);
try {
// Turn 1
await executor.execute(executeInputs([{ role: "user", content: "ping" }]));
// Turn 2 — caller sends the OpenAI history including the prior assistant.
await executor.execute(withConnection("conn-cont"));
// Turn 2 — caller sends history including prior assistant
await executor.execute(
executeInputs([
{ role: "user", content: "ping" },
{ role: "assistant", content: "pong" },
{ role: "user", content: "ping again" },
])
withConnection("conn-cont", {
body: {
messages: [
{ role: "user", content: "ping" },
{ role: "assistant", content: "pong" },
{ role: "user", content: "ping again" },
],
},
})
);
assert.equal(captured.length, 2, "two upstream calls");
const turn1 = (captured[0].body as { variables: Record<string, unknown> }).variables;
const turn2 = (captured[1].body as { variables: Record<string, unknown> }).variables;
assert.equal(turn1.isNewConversation, true);
assert.equal(turn2.isNewConversation, false, "second turn → continues");
assert.equal(
turn2.conversationId,
turn1.conversationId,
"second turn reuses first turn's conversation id"
);
assert.equal(turn2.content, "ping again", "second turn → only the latest user content");
// Continuation completed without error (both turns should succeed)
assert.equal(MockWebSocket.instances.length, 2, "two WS connections made");
} finally {
globalThis.fetch = original;
globalThis.fetch = originalFetch;
restore();
}
});
test("muse-spark-web: connection isolation — different connectionId → independent conversations", async () => {
// ─── Test 3: Missing authorization returns 400 ────────────────────────────────
test("muse-spark-web: missing authorization returns 400", async () => {
__resetMuseSparkConversationCacheForTesting();
const executor = new MuseSparkWebExecutor();
const original = globalThis.fetch;
const { fetchFn, captured } = captureFetch(() => metaAiSseResponse("pong"));
globalThis.fetch = fetchFn;
const result = await executor.execute(
makeBaseInput({
credentials: { apiKey: "ecto_1_sess=test123", connectionId: "conn-noauth" },
})
);
assert.equal(result.response.status, 400);
const body = await result.response.json();
assert.match(body.error.message, /Authorization/);
});
// ─── Test 4: WS error returns error status ────────────────────────────────────
test("muse-spark-web: WebSocket error returns error status", async () => {
__resetMuseSparkConversationCacheForTesting();
const executor = new MuseSparkWebExecutor();
const originalFetch = globalThis.fetch;
globalThis.fetch = async () => new Response("{}", { status: 200 });
class ErrorWs {
onopen: (() => void) | null = null;
onmessage: ((evt: MockWsMessage) => void) | null = null;
onclose: (() => void) | null = null;
onerror: ((evt: Error) => void) | null = null;
readyState = WebSocket.CONNECTING;
url: string;
constructor(url: string) {
this.url = url;
setTimeout(() => this.onerror?.(new Error("fail")), 10);
}
send(_data: Uint8Array | string) {}
close() {
this.onclose?.();
}
}
const restore = __setMuseSparkWebSocketForTesting(ErrorWs as unknown as typeof WebSocket);
try {
const baseInputs = (id: string) => ({
model: "muse-spark",
body: {
messages: [
{ role: "user", content: "ping" },
{ role: "assistant", content: "pong" },
{ role: "user", content: "again" },
],
},
stream: false,
credentials: { apiKey: "abra_sess=foo", connectionId: id },
signal: null,
log: null,
upstreamExtraHeaders: undefined,
const result = await executor.execute(withConnection("conn-err"));
assert.ok(
result.response.status === 502 || result.response.status === 401,
`Got error status: ${result.response.status}`
);
} finally {
globalThis.fetch = originalFetch;
restore();
}
});
// ─── Test 5: GraphQL error in 200 response is detected ─────────────────────
test("muse-spark-web: GraphQL error in 200 response is detected", async () => {
__resetMuseSparkConversationCacheForTesting();
MockWebSocket.instances = [];
const executor = new MuseSparkWebExecutor();
const originalFetch = globalThis.fetch;
globalThis.fetch = async () =>
new Response(JSON.stringify({ errors: [{ message: "Unknown type 'AttachmentInput'" }] }), {
status: 200,
});
// Two different connections both have the same OpenAI history with the
// same prior assistant content. They must not collide on the cache.
await executor.execute(baseInputs("conn-A") as Parameters<MuseSparkWebExecutor["execute"]>[0]);
await executor.execute(baseInputs("conn-B") as Parameters<MuseSparkWebExecutor["execute"]>[0]);
const a = (captured[0].body as { variables: Record<string, unknown> }).variables;
const b = (captured[1].body as { variables: Record<string, unknown> }).variables;
assert.equal(a.isNewConversation, true);
assert.equal(b.isNewConversation, true);
assert.notEqual(a.conversationId, b.conversationId);
} finally {
globalThis.fetch = original;
}
});
test("muse-spark-web: meta error during continuation evicts the stale cache entry", async () => {
__resetMuseSparkConversationCacheForTesting();
const executor = new MuseSparkWebExecutor();
const original = globalThis.fetch;
// Reply 1: success. Reply 2: HTTP 400 (e.g. Meta deleted the conversation).
// Reply 3: success again — cache must have been evicted, so this turn
// should open a fresh conversation, not reuse the dead one.
let n = 0;
const fetchFn: typeof fetch = async () => {
n++;
if (n === 2) {
return new Response(JSON.stringify({ errors: [{ message: "conversation not found" }] }), {
status: 400,
headers: { "Content-Type": "application/json" },
});
}
return metaAiSseResponse(n === 1 ? "pong" : "pong-2");
};
const captured: CapturedRequest[] = [];
globalThis.fetch = (async (input, init) => {
let body: unknown = undefined;
if (init?.body && typeof init.body === "string") {
try {
body = JSON.parse(init.body);
} catch {
body = init.body;
}
}
captured.push({
url: typeof input === "string" ? input : (input as URL).toString(),
init,
body,
});
return fetchFn(input as never, init as never);
}) as typeof fetch;
const restore = __setMuseSparkWebSocketForTesting(MockWebSocket as unknown as typeof WebSocket);
try {
// Turn 1 — opens conversation A and caches it.
await executor.execute(executeInputs([{ role: "user", content: "ping" }]));
// Turn 2 — would continue conversation A but Meta returns 400.
await executor.execute(
executeInputs([
{ role: "user", content: "ping" },
{ role: "assistant", content: "pong" },
{ role: "user", content: "again" },
])
);
// Turn 3 — same prior-assistant content, retried by the user. Cache
// should have been evicted on turn 2, so this turn opens a new
// conversation rather than re-trying the dead one.
await executor.execute(
executeInputs([
{ role: "user", content: "ping" },
{ role: "assistant", content: "pong" },
{ role: "user", content: "again" },
])
);
const t1 = (captured[0].body as { variables: Record<string, unknown> }).variables;
const t2 = (captured[1].body as { variables: Record<string, unknown> }).variables;
const t3 = (captured[2].body as { variables: Record<string, unknown> }).variables;
assert.equal(t1.isNewConversation, true);
assert.equal(t2.isNewConversation, false, "turn 2 attempted to continue");
assert.equal(t2.conversationId, t1.conversationId);
assert.equal(
t3.isNewConversation,
true,
"turn 3 must open a fresh conversation after the stale entry was evicted"
);
assert.notEqual(
t3.conversationId,
t1.conversationId,
"turn 3 must not reuse the dead conversation id"
);
const result = await executor.execute(withConnection("conn-gql-err"));
assert.equal(result.response.status, 502);
const body = await result.response.json();
assert.match(body.error.message, /AttachmentInput/);
} finally {
globalThis.fetch = original;
globalThis.fetch = originalFetch;
restore();
}
});
test("muse-spark-web: parallel chats with identical assistant text but different histories do not collide", async () => {
__resetMuseSparkConversationCacheForTesting();
const executor = new MuseSparkWebExecutor();
const original = globalThis.fetch;
// Both chats end with the assistant saying the same generic line. Without
// hashing the preceding history, both would map to the same cache entry
// and the second chat's continuation would route into the first chat's
// meta.ai conversation.
const COMMON_REPLY = "I don't have access to real-time data.";
const { fetchFn, captured } = captureFetch(() => metaAiSseResponse(COMMON_REPLY));
globalThis.fetch = fetchFn;
try {
// Chat A — turn 1 sets up the cache.
await executor.execute(executeInputs([{ role: "user", content: "what's the weather" }]));
// Chat B — turn 1 (different question, same response) sets up its own cache entry.
await executor.execute(executeInputs([{ role: "user", content: "stock price for AAPL" }]));
const a1 = (captured[0].body as { variables: Record<string, unknown> }).variables;
const b1 = (captured[1].body as { variables: Record<string, unknown> }).variables;
// Chat A — turn 2 should continue conversation A, not jump to B's id.
await executor.execute(
executeInputs([
{ role: "user", content: "what's the weather" },
{ role: "assistant", content: COMMON_REPLY },
{ role: "user", content: "any forecast at all?" },
])
);
// Chat B — turn 2 should continue conversation B.
await executor.execute(
executeInputs([
{ role: "user", content: "stock price for AAPL" },
{ role: "assistant", content: COMMON_REPLY },
{ role: "user", content: "any market info at all?" },
])
);
const a2 = (captured[2].body as { variables: Record<string, unknown> }).variables;
const b2 = (captured[3].body as { variables: Record<string, unknown> }).variables;
assert.equal(a2.isNewConversation, false, "chat A turn 2 continues");
assert.equal(b2.isNewConversation, false, "chat B turn 2 continues");
assert.equal(a2.conversationId, a1.conversationId, "chat A continues into A's conversation");
assert.equal(b2.conversationId, b1.conversationId, "chat B continues into B's conversation");
assert.notEqual(
a2.conversationId,
b2.conversationId,
"the two chats must not collide despite identical assistant text"
);
} finally {
globalThis.fetch = original;
}
});
test("muse-spark-web: empty latestUserContent (no `user` role) falls back to fresh conversation", async () => {
__resetMuseSparkConversationCacheForTesting();
const executor = new MuseSparkWebExecutor();
const original = globalThis.fetch;
const { fetchFn, captured } = captureFetch(() => metaAiSseResponse("ack"));
globalThis.fetch = fetchFn;
try {
// Pre-seed the cache with a normal turn so a hit IS possible if the
// guard isn't in place.
await executor.execute(executeInputs([{ role: "user", content: "ping" }]));
// Now send a payload with no `user` role at all (system + assistant
// only). `latestUserContent` is empty; without the guard the executor
// would route this through the cache-hit path and POST empty content
// with `isNewConversation: false`.
await executor.execute(
executeInputs([
{ role: "system", content: "you are helpful" },
{ role: "assistant", content: "ack" },
])
);
const t2 = (captured[1].body as { variables: Record<string, unknown> }).variables;
assert.equal(
t2.isNewConversation,
true,
"empty latestUserContent must NOT use the cache-hit path"
);
assert.notEqual(t2.content, "", "must not POST empty content");
assert.match(
String(t2.content),
/assistant: ack/,
"should fall through to the folded-history prompt"
);
} finally {
globalThis.fetch = original;
}
});
test(
"muse-spark-web: outgoing variables must NOT declare 'attachments' " +
"(AttachmentInput type removed upstream, regression for #6935)",
async () => {
__resetMuseSparkConversationCacheForTesting();
const executor = new MuseSparkWebExecutor();
const original = globalThis.fetch;
const { fetchFn, captured } = captureFetch(() => metaAiSseResponse("pong"));
globalThis.fetch = fetchFn;
try {
await executor.execute(executeInputs([{ role: "user", content: "hi" }]));
const sentVars = (captured[0].body as { variables: Record<string, unknown> }).variables;
assert.equal(
Object.prototype.hasOwnProperty.call(sentVars, "attachments"),
false,
"variables must omit 'attachments' entirely — Meta removed AttachmentInput from schema"
);
} finally {
globalThis.fetch = original;
}
}
);