diff --git a/open-sse/executors/notion-web.ts b/open-sse/executors/notion-web.ts index 72a3f3c8b7..19bd0f3026 100644 --- a/open-sse/executors/notion-web.ts +++ b/open-sse/executors/notion-web.ts @@ -22,7 +22,10 @@ * chunk — safer than assuming unverified incremental-delta semantics. * * Auth: Cookie-based (token_v2 [+ optional space_id, notion_browser_id, user_id]) - * Method: Direct fetch — no browser automation required. + * Method: Browser-TLS impersonation via tls-client-node (Chrome JA3). Plain + * Node/undici fetch is rejected by Notion's edge with in-band + * `temporarily-unavailable` (HTTP 200, empty assistant text) — curl/Schannel + * and Chrome work with the same cookie + body. See services/notionTlsClient.ts. */ import { randomUUID } from "node:crypto"; import { BaseExecutor, type ExecuteInput } from "./base.ts"; @@ -57,6 +60,10 @@ import { messagesForNotionTranscript, type NotionAgentOptions, } from "../services/notionTranscriptBuilder.ts"; +import { + tlsFetchNotion, + TlsClientUnavailableError, +} from "../services/notionTlsClient.ts"; // Re-exported for unit tests that destructure `mod.` on this module. export { @@ -79,8 +86,9 @@ export { const BASE_URL = "https://app.notion.com"; const NOTION_URL = `${BASE_URL}/api/v3/runInferenceTranscript`; const USER_AGENT = - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36"; -const NOTION_CLIENT_VERSION = "23.13.20260719.1125"; + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36"; +// Match a recent live browser capture (web_providers/notion.txt, 2026-07-20). +const NOTION_CLIENT_VERSION = "23.13.20260720.1949"; // ─── Types ────────────────────────────────────────────────────────────────── @@ -451,29 +459,62 @@ async function sendNotionInferenceRequest(opts: { signal: ExecuteInput["signal"]; }): Promise<{ rawText?: string; errorResult?: ReturnType }> { const { reqBody, reqHeaders, signal } = opts; - let upstream: Response; + // Notion's edge rejects Node/undici TLS fingerprints with in-band + // temporarily-unavailable (HTTP 200, no assistant text). Always use the + // Chrome-JA3 tls-client path for runInferenceTranscript. + let status = 0; + let rawText = ""; try { - upstream = await fetch(NOTION_URL, { + const tlsRes = await tlsFetchNotion(NOTION_URL, { method: "POST", headers: reqHeaders, body: JSON.stringify(reqBody), signal: signal ?? undefined, + // Inference can take a while (tool-autoload + LLM first token). + timeoutMs: + Number.parseInt(process.env.OMNIROUTE_NOTION_TLS_TIMEOUT_MS || "", 10) || 180_000, }); + status = tlsRes.status; + rawText = tlsRes.text ?? ""; } catch (err) { - return { - errorResult: makeErrorResult( - 502, - `Notion fetch failed: ${err instanceof Error ? err.message : "unknown error"}`, - reqBody, - NOTION_URL - ), - }; + if (err instanceof TlsClientUnavailableError) { + // Fall back to plain fetch only when the native TLS sidecar is missing — + // better a degraded path than a hard crash on platforms without the binary. + try { + const upstream = await fetch(NOTION_URL, { + method: "POST", + headers: reqHeaders, + body: JSON.stringify(reqBody), + signal: signal ?? undefined, + }); + status = upstream.status; + rawText = await upstream.text().catch(() => ""); + } catch (fallbackErr) { + return { + errorResult: makeErrorResult( + 502, + `Notion fetch failed: ${fallbackErr instanceof Error ? fallbackErr.message : "unknown error"}`, + reqBody, + NOTION_URL + ), + }; + } + } else { + return { + errorResult: makeErrorResult( + 502, + `Notion fetch failed: ${err instanceof Error ? err.message : "unknown error"}`, + reqBody, + NOTION_URL + ), + }; + } } - if (upstream.status === 401 || upstream.status === 403) { + if (status === 401 || status === 403) { return { errorResult: makeErrorResult( - upstream.status, + status, "Notion session expired or invalid — re-paste token_v2 from notion.so", reqBody, NOTION_URL @@ -481,19 +522,18 @@ async function sendNotionInferenceRequest(opts: { }; } - if (!upstream.ok) { - const errText = await upstream.text().catch(() => ""); + if (status < 200 || status >= 300) { return { errorResult: makeErrorResult( - upstream.status, - `Notion error: ${errText}`, + status || 502, + `Notion error: ${rawText.slice(0, 500)}`, reqBody, NOTION_URL ), }; } - return { rawText: await upstream.text() }; + return { rawText }; } // ─── Executor ─────────────────────────────────────────────────────────────── diff --git a/open-sse/services/notionStreamParser.ts b/open-sse/services/notionStreamParser.ts index 6a7790962c..24c4ed32f1 100644 --- a/open-sse/services/notionStreamParser.ts +++ b/open-sse/services/notionStreamParser.ts @@ -243,7 +243,35 @@ export function extractNotionUpstreamError(raw: string): { if (o) candidates.push(o); } - for (const o of candidates) { + // Flatten nested error objects: live Notion streams often nest the error inside + // `patch-start.data.s[]` (HTTP 200 NDJSON) instead of a top-level `{type:"error"}`. + // Missing that shape used to surface as the misleading "No response from Notion AI". + const flat: Record[] = []; + const pushNested = (o: Record) => { + flat.push(o); + const data = o.data; + if (data && typeof data === "object" && !Array.isArray(data)) { + const s = (data as Record).s; + if (Array.isArray(s)) { + for (const item of s) { + if (item && typeof item === "object" && !Array.isArray(item)) { + flat.push(item as Record); + } + } + } + } + // Also walk a top-level `s` array if present on the record. + if (Array.isArray(o.s)) { + for (const item of o.s) { + if (item && typeof item === "object" && !Array.isArray(item)) { + flat.push(item as Record); + } + } + } + }; + for (const o of candidates) pushNested(o); + + for (const o of flat) { const type = typeof o.type === "string" ? o.type.toLowerCase() : ""; const subType = typeof o.subType === "string" ? o.subType : undefined; const message = @@ -257,6 +285,8 @@ export function extractNotionUpstreamError(raw: string): { if (!isError && !subType) continue; const sub = (subType || "").toLowerCase(); + // Notion often sets isRetryable:false on temporarily-unavailable even though + // a short wait + retry succeeds (TLS/edge flake). Treat those subtypes as retryable. const retryable = o.isRetryable === true || sub.includes("temporarily") || diff --git a/open-sse/services/notionTlsClient.ts b/open-sse/services/notionTlsClient.ts new file mode 100644 index 0000000000..1b7c922309 --- /dev/null +++ b/open-sse/services/notionTlsClient.ts @@ -0,0 +1,593 @@ +/** + * Browser-TLS-impersonating HTTP client for app.notion.com. + * + * Why this exists: Notion AI sits behind the same Cloudflare Enterprise + * configuration as ChatGPT — it pins access to the client's TLS fingerprint + * (JA3/JA4) + HTTP/2 SETTINGS frame ordering. Node's Undici fetch presents an + * obvious "not a browser" handshake and gets challenged with a 403 "Just a + * moment..." page from VPS/datacenter IPs — even with a valid session cookie. + * This module wraps `tls-client-node` (native shared library built from + * bogdanfinn/tls-client) to send a Firefox handshake instead. (issue #2459) + * + * Mirrors `claudeTlsClient.ts` / `perplexityTlsClient.ts`; kept as an independent module so changes here + * cannot regress the production chatgpt-web path. The first call lazily starts + * the managed sidecar; subsequent calls reuse a singleton TLSClient. Process + * exit hooks stop the sidecar cleanly. + */ + +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { mkdtemp, open, unlink, rmdir, stat } from "node:fs/promises"; +import { randomUUID } from "node:crypto"; + +let clientPromise: Promise | null = null; +let exitHookInstalled = false; + +const NOTION_PROFILE = "chrome_146"; // matches the Chrome UA we send +const DEFAULT_TIMEOUT_MS = + Number.parseInt(process.env.OMNIROUTE_NOTION_TLS_TIMEOUT_MS || "", 10) || 30_000; +// Grace period added to the binding's wire-level timeout before our JS-level +// hard timeout fires. Under healthy operation `tls-client-node` honors +// `timeoutMilliseconds` and rejects on its own; the JS-level race only wins +// when the koffi-loaded native library is wedged (which the binding's own +// timer can't escape). Keep the grace small so users don't wait noticeably +// longer than the configured timeout when the binding is dead. +const HARD_TIMEOUT_GRACE_MS = + Number.parseInt(process.env.OMNIROUTE_NOTION_TLS_GRACE_MS || "", 10) || 10_000; + +function installExitHook(): void { + if (exitHookInstalled) return; + exitHookInstalled = true; + const stop = async () => { + if (!clientPromise) return; + try { + const c = (await clientPromise) as { stop?: () => Promise }; + await c.stop?.(); + } catch { + // ignore + } + }; + process.once("beforeExit", stop); + process.once("SIGINT", () => { + void stop(); + }); + process.once("SIGTERM", () => { + void stop(); + }); +} + +/** + * Drop the cached client so the next `getClient()` call respawns it. Called + * when a request observes the native binding has wedged — releasing the + * reference lets a fresh TLSClient (and a fresh koffi load) take over without + * a process restart. + */ +function resetClientCache(): void { + clientPromise = null; +} + +export class TlsClientHangError extends Error { + constructor(message: string) { + super(message); + this.name = "TlsClientHangError"; + } +} + +/** + * Race a `client.request()` promise against (a) a JS-level hard timeout and + * (b) the caller's abort signal. The native binding's `timeoutMilliseconds` + * already covers the wire path; this guards the case where the koffi binding + * itself deadlocks (observed after sustained load), where neither the + * binding's own timer nor a post-call `signal.aborted` re-check can recover. + */ +async function raceWithTimeout( + promise: Promise, + timeoutMs: number, + signal: AbortSignal | null | undefined +): Promise { + let timer: ReturnType | null = null; + let abortListener: (() => void) | null = null; + try { + const racers: Promise[] = [ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => { + reject( + new TlsClientHangError( + `tls-client-node call exceeded ${timeoutMs}ms — native binding likely deadlocked` + ) + ); + }, timeoutMs); + }), + ]; + if (signal) { + racers.push( + new Promise((_, reject) => { + if (signal.aborted) { + reject(makeAbortError(signal)); + return; + } + abortListener = () => reject(makeAbortError(signal)); + signal.addEventListener("abort", abortListener, { once: true }); + }) + ); + } + return await Promise.race(racers); + } finally { + if (timer) clearTimeout(timer); + if (signal && abortListener) signal.removeEventListener("abort", abortListener); + } +} + +async function getClient(): Promise<{ + request: (url: string, opts: Record) => Promise; +}> { + if (!clientPromise) { + clientPromise = (async () => { + try { + const mod = await import("tls-client-node"); + const TLSClient = (mod as { TLSClient: new (opts?: Record) => unknown }) + .TLSClient; + // Native mode loads the shared library directly via koffi, avoiding the + // managed sidecar's localhost HTTP calls that OmniRoute's global fetch + // proxy patch interferes with. + const client = new TLSClient({ runtimeMode: "native" }) as { + start: () => Promise; + request: (url: string, opts: Record) => Promise; + }; + await client.start(); + + installExitHook(); + return client; + } catch (err) { + clientPromise = null; + const msg = err instanceof Error ? err.message : String(err); + throw new TlsClientUnavailableError( + `TLS impersonation client failed to start: ${msg}. ` + + `Verify tls-client-node is installed and its native binary downloaded.` + ); + } + })(); + } + return clientPromise as Promise<{ + request: (url: string, opts: Record) => Promise; + }>; +} + +interface TlsResponseLike { + status: number; + headers: Record; + body: string; // for non-streaming requests, the full response body + cookies?: Record; + text: () => Promise; + bytes: () => Promise; + json: () => Promise; +} + +export class TlsClientUnavailableError extends Error { + constructor(message: string) { + super(message); + this.name = "TlsClientUnavailableError"; + } +} + +export interface TlsFetchOptions { + method?: "GET" | "POST" | "PUT" | "PATCH" | "DELETE"; + headers?: Record; + body?: string; + timeoutMs?: number; + signal?: AbortSignal | null; + /** + * If true, the response body is streamed to a temp file and exposed as a + * ReadableStream. Use for SSE responses (the runInferenceTranscript + * endpoint). Otherwise, the full body is read into memory. + */ + stream?: boolean; + /** EOF marker the upstream sends to signal end of stream (default: "[DONE]"). */ + streamEofSymbol?: string; + /** + * Optional upstream proxy URL (`http://user:pass@host:port` or + * `socks5://...`). When set, the request is tunneled through this proxy + * before reaching notion.so. + * + * Resolution order: + * 1. `options.proxyUrl` (per-call override from caller) + * 2. `process.env.OMNIROUTE_TLS_PROXY_URL` (single-flag opt-in) + * 3. `process.env.HTTPS_PROXY` / `HTTP_PROXY` / `ALL_PROXY` (POSIX-standard fallback) + * + * The native `tls-client-node` binding does **not** consult Go's + * `http.ProxyFromEnvironment`, so the env vars need to be plumbed in here at + * the JS layer. + */ + proxyUrl?: string; +} + +import { resolveProxyForRequest } from "../utils/proxyFetch.ts"; +import { resolveTlsClientProxyUrl } from "./tlsClientProxy.ts"; + +/** + * Resolve the proxy URL for a tls-client request. Per-call value wins; + * otherwise we use the standard proxy fetch resolution which reads from + * the dashboard AsyncLocalStorage context or falls back to env vars. + * + * Fail-closed: if resolution throws (e.g. a configured socks5 proxy with + * ENABLE_SOCKS5_PROXY=false), this rethrows rather than returning undefined — + * undefined would let the native binding connect directly and leak the real IP. + */ +function resolveProxyUrl(perCall: string | undefined): string | undefined { + return resolveTlsClientProxyUrl("https://app.notion.com", perCall, resolveProxyForRequest); +} + +export interface TlsFetchResult { + status: number; + headers: Headers; + /** Full response body as text — only populated for non-streaming requests. */ + text: string | null; + /** Streaming body — only populated when options.stream === true. */ + body: ReadableStream | null; +} + +// Test-only injection point. Tests call __setTlsFetchOverrideForTesting() +// to replace the real TLS client with a mock; production never touches this. +let testOverride: ((url: string, options: TlsFetchOptions) => Promise) | null = + null; + +export function __setTlsFetchOverrideForTesting(fn: typeof testOverride): void { + testOverride = fn; +} + +/** + * Make a single HTTP request to notion.so with a Chrome-like TLS fingerprint. + * + * Throws TlsClientUnavailableError if the native binary failed to load. + */ +export async function tlsFetchNotion( + url: string, + options: TlsFetchOptions = {} +): Promise { + if (testOverride) return testOverride(url, options); + // Honor abort signals up-front. tls-client-node's koffi binding doesn't + // accept an AbortSignal mid-flight (the binary call is opaque), so the best + // we can do is bail before issuing the call. We also re-check after — if + // the caller aborted while the upstream was running, throw rather than + // returning a stale response so the caller doesn't try to use it. + if (options.signal?.aborted) { + throw makeAbortError(options.signal); + } + const client = await getClient(); + if (options.signal?.aborted) { + throw makeAbortError(options.signal); + } + + const requestOptions: Record = { + method: options.method || "GET", + headers: options.headers || {}, + body: options.body, + tlsClientIdentifier: NOTION_PROFILE, + timeoutMilliseconds: options.timeoutMs ?? DEFAULT_TIMEOUT_MS, + followRedirects: true, + withRandomTLSExtensionOrder: true, + // Plumb the configured proxy through to the native binding. tls-client-node + // consults `proxyUrl` in the per-call options (it does NOT auto-pick up + // HTTP_PROXY / HTTPS_PROXY env), so callers / env have to be threaded in + // explicitly. See `resolveProxyUrl()` for the lookup order. + proxyUrl: resolveProxyUrl(options.proxyUrl), + }; + + if (options.stream) { + return await tlsFetchStreaming( + client, + url, + requestOptions, + options.streamEofSymbol, + options.signal ?? null, + (options.timeoutMs ?? DEFAULT_TIMEOUT_MS) + HARD_TIMEOUT_GRACE_MS + ); + } + + let tlsResponse: TlsResponseLike; + try { + tlsResponse = await raceWithTimeout( + client.request(url, requestOptions), + (options.timeoutMs ?? DEFAULT_TIMEOUT_MS) + HARD_TIMEOUT_GRACE_MS, + options.signal ?? null + ); + } catch (err) { + if (err instanceof TlsClientHangError) { + // The native binding is wedged — drop the singleton so the next + // request respawns a fresh client (and a fresh koffi load). + resetClientCache(); + } + throw err; + } + if (options.signal?.aborted) { + throw makeAbortError(options.signal); + } + return { + status: tlsResponse.status, + headers: toHeaders(tlsResponse.headers), + text: tlsResponse.body, + body: null, + }; +} + +function makeAbortError(signal: AbortSignal): Error { + const reason = signal.reason; + if (reason instanceof Error) return reason; + const err = new Error(typeof reason === "string" ? reason : "The operation was aborted"); + err.name = "AbortError"; + return err; +} + +function toHeaders(raw: Record): Headers { + const h = new Headers(); + for (const [k, vs] of Object.entries(raw || {})) { + for (const v of vs) h.append(k, v); + } + return h; +} + +/** + * Returns true if the response body is a Cloudflare challenge/interstitial page + * rather than a real Perplexity response. From VPS/datacenter IPs a valid cookie + * still gets a 403 "Just a moment..." HTML page; distinguishing it from a genuine + * auth failure lets the caller surface an actionable error (issue #2459). + * + * Exported so the executor and the connection validator share one detector. + */ +export function isCloudflareChallenge(text: string | null | undefined): boolean { + if (!text) return false; + return /just a moment|window\._cf_chl_opt|challenges\.cloudflare\.com|attention required|cf-chl/i.test( + text + ); +} + +// ─── Streaming via temp file ──────────────────────────────────────────────── +// tls-client-node's streaming primitive writes the response body chunk-by-chunk +// to a file path, terminating when the upstream sends `streamOutputEOFSymbol`. +// We tail the file from a worker and surface the bytes as a ReadableStream. + +async function tlsFetchStreaming( + client: { request: (url: string, opts: Record) => Promise }, + url: string, + requestOptions: Record, + eofSymbol = "[DONE]", + signal: AbortSignal | null = null, + hardTimeoutMs: number = DEFAULT_TIMEOUT_MS + HARD_TIMEOUT_GRACE_MS +): Promise { + const dir = await mkdtemp(join(tmpdir(), "pplx-stream-")); + const path = join(dir, `${randomUUID()}.sse`); + + const streamOpts = { + ...requestOptions, + streamOutputPath: path, + streamOutputBlockSize: 1024, + streamOutputEOFSymbol: eofSymbol, + }; + + // Kick off the request without awaiting — tls-client writes the body to + // `path` chunk-by-chunk while the call runs. The Promise resolves when the + // request fully completes (full body written). Wrapping in raceWithTimeout + // guarantees this promise eventually settles even if the koffi binding + // wedges; on hang we reset the singleton so the next request respawns. + let resetOnHang = true; + const requestPromise = raceWithTimeout( + client.request(url, streamOpts), + hardTimeoutMs, + signal + ).catch((err: unknown) => { + if (resetOnHang && err instanceof TlsClientHangError) { + resetClientCache(); + resetOnHang = false; + } + // Re-throw so downstream consumers (waitForContent, tailFile) observe + // the rejection and surface it instead of treating the stream as having + // ended cleanly. + throw err; + }); + + // Wait for the file to exist AND have at least one byte. tls-client-node + // creates the output file when the request starts, but the file can be + // empty for a brief window before the first body chunk lands — peeking + // during that window would return "" and misclassify the response as + // non-SSE, dropping us into the buffered-wait branch and silently turning + // a streaming request into a buffered one. Waiting for content avoids + // that race; if the request actually fails before producing any bytes, + // the timeout falls through to the requestPromise drain below (returning + // the real upstream status). + const ready = await waitForContent(path, 5_000, requestPromise); + if (!ready) { + const r = await requestPromise.catch( + (e) => ({ status: 502, headers: {}, body: String(e) }) as TlsResponseLike + ); + await cleanupTempPath(path); + return { + status: r.status, + headers: toHeaders(r.headers), + text: r.body, + body: null, + }; + } + + // Peek the first bytes to decide whether this looks like SSE. Anything + // that doesn't positively look like SSE (JSON `{...}`, HTML `<...>`, plain + // text rate-limit messages, Cloudflare challenge pages, etc.) gets surfaced + // as a non-streaming response so the executor sees the real upstream status + // and body — otherwise non-2xx error pages get silently treated as 200 OK + // and the SSE parser produces an empty completion. + const peek = await readFirstBytes(path, 256); + if (!looksLikeSse(peek)) { + const r = await requestPromise.catch( + (e) => ({ status: 502, headers: {}, body: String(e) }) as TlsResponseLike + ); + await cleanupTempPath(path); + return { + status: r.status, + headers: toHeaders(r.headers), + text: r.body, + body: null, + }; + } + + // Looks like SSE — start tailing. SSE bodies in practice are always 2xx; + // tls-client-node doesn't expose response status separately from full-body + // completion, so we report 200 and let the SSE parser consume the stream. + const stream = tailFile(path, eofSymbol, requestPromise, signal); + const headers = new Headers({ + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + }); + return { status: 200, headers, text: null, body: stream }; +} + +/** + * Returns true if the peeked response body looks like an SSE stream — i.e., + * begins (after any leading whitespace) with one of the SSE field markers + * (`data:`, `event:`, `id:`, `retry:`) or a comment line (`:`). + * + * Exported for tests. + */ +export function looksLikeSse(text: string): boolean { + const trimmed = text.replace(/^[\s\r\n]+/, ""); + if (!trimmed) return false; + if (trimmed.startsWith(":")) return true; + return /^(data|event|id|retry):/i.test(trimmed); +} + +async function cleanupTempPath(path: string): Promise { + await unlink(path).catch(() => {}); + const dir = path.substring(0, path.lastIndexOf("/")); + await rmdir(dir).catch(() => {}); +} + +async function readFirstBytes(path: string, n: number): Promise { + const fd = await open(path, "r"); + try { + const buf = Buffer.alloc(n); + const { bytesRead } = await fd.read(buf, 0, n, 0); + return buf.subarray(0, bytesRead).toString("utf8"); + } finally { + await fd.close().catch(() => {}); + } +} + +/** + * Wait for the streaming output file to exist AND contain at least one byte. + * Returns false if the request settles before any bytes arrive (so the caller + * can drain `requestPromise` and surface the real upstream status). Returns + * true as soon as the file has data — even one byte is enough for the SSE + * heuristic to give a useful answer. + */ +async function waitForContent( + path: string, + timeoutMs: number, + requestPromise: Promise +): Promise { + let requestSettled = false; + requestPromise.then( + () => { + requestSettled = true; + }, + () => { + requestSettled = true; + } + ); + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + try { + const s = await stat(path); + if (s.size > 0) return true; + } catch { + // file doesn't exist yet + } + // If the request finished without producing any bytes, no point waiting + // out the rest of the timeout — let the caller drain it. + if (requestSettled) return false; + await sleep(25); + } + return false; +} + +function tailFile( + path: string, + eofSymbol: string, + done: Promise, + signal: AbortSignal | null = null +): ReadableStream { + return new ReadableStream({ + async start(controller) { + const fd = await open(path, "r"); + const buf = Buffer.alloc(64 * 1024); + let offset = 0; + let finished = false; + let aborted = false; + let upstreamError: Error | null = null; + + // Track request settlement, capturing both fulfillment and rejection. + // Without the rejection branch, a mid-stream tls-client-node error + // becomes an unhandledRejection — the stream cleans up silently and + // the consumer sees what looks like a successful truncated response. + done.then( + () => { + finished = true; + }, + (err) => { + upstreamError = err instanceof Error ? err : new Error(String(err)); + finished = true; + } + ); + + // If the caller aborts, stop tailing immediately. + const onAbort = () => { + aborted = true; + }; + if (signal) { + if (signal.aborted) aborted = true; + else signal.addEventListener("abort", onAbort, { once: true }); + } + + let errored = false; + try { + while (!aborted) { + const { bytesRead } = await fd.read(buf, 0, buf.length, offset); + if (bytesRead > 0) { + const chunk = buf.subarray(0, bytesRead); + offset += bytesRead; + const text = chunk.toString("utf8"); + if (text.includes(eofSymbol)) { + const cutAt = text.indexOf(eofSymbol) + eofSymbol.length; + controller.enqueue(new Uint8Array(chunk.subarray(0, cutAt))); + break; + } + controller.enqueue(new Uint8Array(chunk)); + } else if (finished) { + // No more data and request completed. If the request rejected, + // surface the error so the consumer doesn't think the stream + // ended cleanly. + if (upstreamError) { + controller.error(upstreamError); + errored = true; + } + break; + } else { + await sleep(25); + } + } + } catch (err) { + controller.error(err); + errored = true; + } finally { + if (signal) signal.removeEventListener("abort", onAbort); + await fd.close().catch(() => {}); + await unlink(path).catch(() => {}); + const dir = path.substring(0, path.lastIndexOf("/")); + await rmdir(dir).catch(() => {}); + if (!errored) controller.close(); + } + }, + }); +} + +function sleep(ms: number): Promise { + return new Promise((r) => setTimeout(r, ms)); +} diff --git a/tests/unit/executor-notion-web.test.ts b/tests/unit/executor-notion-web.test.ts index ced4f186fa..8fc3b32196 100644 --- a/tests/unit/executor-notion-web.test.ts +++ b/tests/unit/executor-notion-web.test.ts @@ -9,6 +9,31 @@ import assert from "node:assert/strict"; const mod = await import("../../open-sse/executors/notion-web.ts"); const { getModelsByProviderId } = await import("../../open-sse/config/providerModels.ts"); const { WEB_COOKIE_PROVIDERS } = await import("../../src/shared/constants/providers/web-cookie.ts"); +const { __setTlsFetchOverrideForTesting } = await import( + "../../open-sse/services/notionTlsClient.ts" +); + +/** Mock the Chrome-JA3 path used by sendNotionInferenceRequest (not global fetch). */ +function installNotionTlsMock( + handler: (url: string, opts: { headers?: Record; body?: string }) => Promise<{ + status: number; + text: string; + }> +): () => void { + __setTlsFetchOverrideForTesting(async (url, options) => { + const r = await handler(url, { + headers: options.headers as Record | undefined, + body: options.body, + }); + return { + status: r.status, + headers: new Headers(), + text: r.text, + body: null, + }; + }); + return () => __setTlsFetchOverrideForTesting(null); +} describe("NotionWebExecutor — registry consistency", () => { it("is present in WEB_COOKIE_PROVIDERS with the expected shape", () => { @@ -76,7 +101,7 @@ describe("NotionWebExecutor — instantiation & auth errors", () => { /** Cookie with space_id so execute() does not need a live getSpaces call. */ const COOKIE_WITH_SPACE = "token_v2=xyz; space_id=space-1; notion_user_id=user-1"; -describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { +describe("NotionWebExecutor — upstream translation (mocked TLS fetch)", () => { it("posts createThread + config/context/user and returns a chat.completion", async () => { const executor = new mod.NotionWebExecutor(); let capturedUrl = ""; @@ -87,36 +112,33 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { spaceId?: string; transcript: Array<{ type: string; value: unknown }>; } | null = null; - const originalFetch = globalThis.fetch; - try { - globalThis.fetch = (async (url: string | URL, opts: RequestInit) => { - capturedUrl = String(url); - capturedHeaders = opts.headers as Record; - capturedBody = JSON.parse(String(opts.body)); - // Modern record-map response (live shape 2026-07-19). - const ndjson = [ - JSON.stringify({ type: "patch-start", data: { s: [] } }), - JSON.stringify({ - type: "record-map", - recordMap: { - thread_message: { - m1: { + const restore = installNotionTlsMock(async (url, opts) => { + capturedUrl = url; + capturedHeaders = opts.headers || {}; + capturedBody = JSON.parse(String(opts.body)); + const ndjson = [ + JSON.stringify({ type: "patch-start", data: { s: [] } }), + JSON.stringify({ + type: "record-map", + recordMap: { + thread_message: { + m1: { + value: { value: { - value: { - step: { - type: "agent-inference", - value: [{ type: "text", content: 'Hello there!' }], - }, + step: { + type: "agent-inference", + value: [{ type: "text", content: 'Hello there!' }], }, }, }, }, }, - }), - ].join("\n"); - return new Response(ndjson, { status: 200 }); - }) as typeof fetch; - + }, + }), + ].join("\n"); + return { status: 200, text: ndjson }; + }); + try { const result = await executor.execute({ model: "notion-ai", body: { messages: [{ role: "user", content: "hi" }] }, @@ -129,7 +151,6 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { assert.equal(capturedHeaders.Cookie, COOKIE_WITH_SPACE); assert.equal(capturedHeaders["x-notion-space-id"], "space-1"); assert.equal(capturedHeaders["x-notion-active-user-header"], "user-1"); - // Browser fingerprint headers to reduce Cloudflare challenges. assert.ok(capturedHeaders["sec-ch-ua"], "sec-ch-ua should be present"); assert.ok(capturedHeaders["sec-fetch-dest"], "sec-fetch-dest should be present"); assert.ok(capturedHeaders["sec-fetch-mode"], "sec-fetch-mode should be present"); @@ -141,7 +162,6 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { assert.equal(capturedBody.createThread, true); assert.ok(typeof capturedBody.threadId === "string" && capturedBody.threadId.length > 0); assert.equal(capturedBody.spaceId, "space-1"); - // Transcript: config + context + user (system would fold into context). assert.equal(capturedBody.transcript[0].type, "config"); assert.equal(capturedBody.transcript[1].type, "context"); assert.equal(capturedBody.transcript[2].type, "user"); @@ -153,10 +173,9 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { choices: Array<{ message: { content: string } }>; }; assert.equal(json.object, "chat.completion"); - // Lang tag stripped; final assistant text kept. assert.equal(json.choices[0].message.content, "Hello there!"); } finally { - globalThis.fetch = originalFetch; + restore(); } }); @@ -165,14 +184,11 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { let capturedBody: { transcript: Array<{ type: string; value?: { model?: string } }>; } | null = null; - const originalFetch = globalThis.fetch; + const restore = installNotionTlsMock(async (_url, opts) => { + capturedBody = JSON.parse(String(opts.body)); + return { status: 200, text: JSON.stringify({ value: [["ok"]] }) }; + }); try { - globalThis.fetch = (async (_url: string | URL, opts: RequestInit) => { - capturedBody = JSON.parse(String(opts.body)); - return new Response(JSON.stringify({ value: [["ok"]] }), { status: 200 }); - }) as typeof fetch; - - // Legacy food codename still accepted for power users / cached clients. await executor.execute({ model: "orange-mousse", body: { messages: [{ role: "user", content: "hi" }] }, @@ -187,7 +203,7 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { assert.equal(capturedBody.transcript[1].type, "context"); assert.equal(capturedBody.transcript[2].type, "user"); } finally { - globalThis.fetch = originalFetch; + restore(); } }); @@ -196,14 +212,11 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { let capturedBody: { transcript: Array<{ type: string; value?: { model?: string } }>; } | null = null; - let responseModel = ""; - const originalFetch = globalThis.fetch; + const restore = installNotionTlsMock(async (_url, opts) => { + capturedBody = JSON.parse(String(opts.body)); + return { status: 200, text: JSON.stringify({ value: [["ok"]] }) }; + }); try { - globalThis.fetch = (async (_url: string | URL, opts: RequestInit) => { - capturedBody = JSON.parse(String(opts.body)); - return new Response(JSON.stringify({ value: [["ok"]] }), { status: 200 }); - }) as typeof fetch; - const result = await executor.execute({ model: "notion-web/gpt-5.6-sol", body: { messages: [{ role: "user", content: "hi" }] }, @@ -214,14 +227,11 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { assert.ok(capturedBody); assert.equal(capturedBody.transcript[0].type, "config"); - // Wire protocol still uses the food codename. assert.equal(capturedBody.transcript[0].value?.model, "orange-mousse"); - // Response echoes the client-facing real model name. const json = (await result.response.json()) as { model?: string }; - responseModel = json.model || ""; - assert.equal(responseModel, "gpt-5.6-sol"); + assert.equal(json.model || "", "gpt-5.6-sol"); } finally { - globalThis.fetch = originalFetch; + restore(); } }); @@ -230,13 +240,11 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { let capturedBody: { transcript: Array<{ type: string; value?: { model?: string } }>; } | null = null; - const originalFetch = globalThis.fetch; + const restore = installNotionTlsMock(async (_url, opts) => { + capturedBody = JSON.parse(String(opts.body)); + return { status: 200, text: JSON.stringify({ value: [["ok"]] }) }; + }); try { - globalThis.fetch = (async (_url: string | URL, opts: RequestInit) => { - capturedBody = JSON.parse(String(opts.body)); - return new Response(JSON.stringify({ value: [["ok"]] }), { status: 200 }); - }) as typeof fetch; - const result = await executor.execute({ model: "fable-5", body: { messages: [{ role: "user", content: "hi" }] }, @@ -250,20 +258,18 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { const json = (await result.response.json()) as { model?: string }; assert.equal(json.model, "fable-5"); } finally { - globalThis.fetch = originalFetch; + restore(); } }); it("accepts a full cookie header verbatim (already containing token_v2=)", async () => { const executor = new mod.NotionWebExecutor(); let capturedHeaders: Record = {}; - const originalFetch = globalThis.fetch; + const restore = installNotionTlsMock(async (_url, opts) => { + capturedHeaders = opts.headers || {}; + return { status: 200, text: JSON.stringify({ value: [["ok"]] }) }; + }); try { - globalThis.fetch = (async (_url: string | URL, opts: RequestInit) => { - capturedHeaders = opts.headers as Record; - return new Response(JSON.stringify({ value: [["ok"]] }), { status: 200 }); - }) as typeof fetch; - await executor.execute({ model: "notion-ai", body: { messages: [{ role: "user", content: "hi" }] }, @@ -274,19 +280,17 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { assert.equal(capturedHeaders.Cookie, "token_v2=xyz; space_id=abc-def"); } finally { - globalThis.fetch = originalFetch; + restore(); } }); it("returns a pseudo-streamed SSE response with [DONE] when stream=true", async () => { const executor = new mod.NotionWebExecutor(); - const originalFetch = globalThis.fetch; + const restore = installNotionTlsMock(async () => ({ + status: 200, + text: JSON.stringify({ value: [["Streamed reply"]] }), + })); try { - globalThis.fetch = (async () => - new Response(JSON.stringify({ value: [["Streamed reply"]] }), { - status: 200, - })) as typeof fetch; - const result = await executor.execute({ model: "notion-ai", body: { messages: [{ role: "user", content: "hi" }] }, @@ -300,17 +304,17 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { assert.match(text, /Streamed reply/); assert.match(text, /data: \[DONE\]/); } finally { - globalThis.fetch = originalFetch; + restore(); } }); it("returns 502 when Notion sends no parseable text (endpoint drift)", async () => { const executor = new mod.NotionWebExecutor(); - const originalFetch = globalThis.fetch; + const restore = installNotionTlsMock(async () => ({ + status: 200, + text: "not-json\n{}", + })); try { - globalThis.fetch = (async () => - new Response("not-json\n{}", { status: 200 })) as typeof fetch; - const result = await executor.execute({ model: "notion-ai", body: { messages: [{ role: "user", content: "hi" }] }, @@ -320,16 +324,14 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { } as never); assert.equal(result.response.status, 502); } finally { - globalThis.fetch = originalFetch; + restore(); } }); it("returns a sanitized 403 error without leaking raw upstream error text shape", async () => { const executor = new mod.NotionWebExecutor(); - const originalFetch = globalThis.fetch; + const restore = installNotionTlsMock(async () => ({ status: 403, text: "Forbidden" })); try { - globalThis.fetch = (async () => new Response("Forbidden", { status: 403 })) as typeof fetch; - const result = await executor.execute({ model: "notion-ai", body: { messages: [{ role: "user", content: "hi" }] }, @@ -343,21 +345,18 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { }; assert.match(errBody.error.message, /session expired|invalid/i); assert.equal(errBody.error.code, "HTTP_403"); - // No stack trace / file path leakage (Hard Rule #12). assert.ok(!errBody.error.message.includes("at /")); } finally { - globalThis.fetch = originalFetch; + restore(); } }); - it("returns 502 with a sanitized message when the fetch itself throws", async () => { + it("returns 502 with a sanitized message when the TLS fetch itself throws", async () => { const executor = new mod.NotionWebExecutor(); - const originalFetch = globalThis.fetch; + const restore = installNotionTlsMock(async () => { + throw new Error("getaddrinfo ENOTFOUND www.notion.so at /some/internal/path.ts:42"); + }); try { - globalThis.fetch = (async () => { - throw new Error("getaddrinfo ENOTFOUND www.notion.so at /some/internal/path.ts:42"); - }) as typeof fetch; - const result = await executor.execute({ model: "notion-ai", body: { messages: [{ role: "user", content: "hi" }] }, @@ -369,7 +368,44 @@ describe("NotionWebExecutor — upstream translation (mocked fetch)", () => { const errBody = (await result.response.json()) as { error: { message: string } }; assert.ok(!errBody.error.message.includes("at /some/internal/path.ts")); } finally { - globalThis.fetch = originalFetch; + restore(); + } + }); + + it("surfaces nested patch-start temporarily-unavailable as a typed error (not empty-body 502)", async () => { + const executor = new mod.NotionWebExecutor(); + const restore = installNotionTlsMock(async () => ({ + status: 200, + text: JSON.stringify({ + type: "patch-start", + data: { + s: [ + { + type: "error", + message: "Something went wrong. Please try again later.", + subType: "temporarily-unavailable", + isRetryable: false, + }, + ], + }, + version: 1, + }), + })); + try { + const result = await executor.execute({ + model: "notion-ai", + body: { messages: [{ role: "user", content: "hi" }] }, + stream: false, + credentials: { apiKey: COOKIE_WITH_SPACE }, + signal: null, + } as never); + // Nested temporarily-unavailable is treated as retryable → may land as 503 after retry. + assert.ok([502, 503].includes(result.response.status)); + const errBody = (await result.response.json()) as { error: { message: string } }; + assert.match(errBody.error.message, /temporarily-unavailable|went wrong/i); + assert.ok(!/No response from Notion AI/i.test(errBody.error.message)); + } finally { + restore(); } }); });