diff --git a/open-sse/executors/notion-web.ts b/open-sse/executors/notion-web.ts index 364bbb4769..30475b8ef3 100644 --- a/open-sse/executors/notion-web.ts +++ b/open-sse/executors/notion-web.ts @@ -7,9 +7,12 @@ * (notion2api / Notion2API-go, cited in issue #6758): a `token_v2` session * cookie posted to `POST /api/v3/runInferenceTranscript`. * - * Live capture (2026-07-19) against a Business workspace confirmed the - * contract that actually works: - * - createThread: true + a fresh threadId (createThread:false → ValidationError 400) + * Live capture (2026-07-19 / 2026-07-20) against a Business workspace confirmed: + * - First turn: createThread: true + a fresh threadId + * (createThread:false without a known threadId → ValidationError 400) + * - Follow-ups: createThread: false + the SAME threadId + full transcript + * (OpenAI multi-turn messages[] maps to one Notion AI chat; a new UUID + * every request forces a new chat and breaks agent/tool continuity) * - transcript starts with config + context, then user/assistant steps * - x-notion-space-id + x-notion-active-user-header required * - response is NDJSON patch-start / patch / record-map (not legacy rich-text @@ -30,6 +33,44 @@ import { resolveNotionCodename, resolveNotionRuntimeWorkspace, } from "../services/notionWebModels.ts"; +import { + __resetNotionThreadSessionsForTests, + conversationPrefixBeforeLastUser, + extractNotionMessageText, + hashNotionConversation, + notionThreadMarkConfirmed, + notionThreadMarkCreateAttempted, + notionThreadSessionLookup, + notionThreadSessionStore, + readClientThreadId, + resolveNotionThreadBinding, + type NotionMessage, +} from "../services/notionThreadSessions.ts"; +import { + extractNotionUpstreamError, + parseNotionInferenceStream, + sanitizeNotionAssistantText, +} from "../services/notionStreamParser.ts"; +import { + buildNotionTranscript, + messagesForNotionTranscript, + type NotionAgentOptions, +} from "../services/notionTranscriptBuilder.ts"; + +// Re-exported for unit tests that destructure `mod.` on this module. +export { + __resetNotionThreadSessionsForTests, + buildNotionTranscript, + conversationPrefixBeforeLastUser, + extractNotionUpstreamError, + hashNotionConversation, + notionThreadSessionLookup, + notionThreadSessionStore, + parseNotionInferenceStream, + resolveNotionThreadBinding, + notionThreadMarkCreateAttempted, + sanitizeNotionAssistantText, +}; // ─── Constants ────────────────────────────────────────────────────────────── @@ -42,14 +83,12 @@ const NOTION_CLIENT_VERSION = "23.13.20260719.1125"; // ─── Types ────────────────────────────────────────────────────────────────── -interface NotionMessage { - role: string; - content: string; -} - interface NotionRequestBody { messages?: NotionMessage[]; model?: string; + /** Optional client-supplied Notion thread continuity (also via X-Notion-Thread-Id). */ + notion_thread_id?: string; + thread_id?: string; } // ─── Helpers — credential resolution ─────────────────────────────────────── @@ -138,357 +177,6 @@ function extractUserIdFromCookie(cookie: string): string { return extractNotionUserIdFromCookie(cookie); } -function isoNow(): string { - // Millisecond precision matches the browser client. - return new Date().toISOString().replace(/\.\d{3}Z$/, (m) => m); // keep ms + Z -} - -// ─── Helpers — request/response translation ──────────────────────────────── - -/** - * Build a Notion `runInferenceTranscript` transcript array from OpenAI-style - * chat messages. - * - * Live contract (verified 2026-07-19): - * - Leading `config` (workflow + optional model food-codename) - * - Leading `context` (spaceId / userId / surface / timezone) - * - User turns as `type: "user"` (legacy `human` also works with createThread, - * but `user` matches the current web client) - * - Assistant turns as `agent-inference` text parts - */ -function buildNotionConfigStep(model: string): Record { - const configValue: Record = { - type: "workflow", - useWebSearch: false, - searchScopes: [{ type: "everything" }], - modelFromUser: Boolean(model), - enableAgentAutomations: false, - enableAgentIntegrations: false, - enableCustomAgents: false, - enableDatabaseAgents: false, - enableUserSessionContext: false, - isCustomAgent: false, - }; - if (model) configValue.model = model; - return { id: randomUUID(), type: "config", value: configValue }; -} - -function buildNotionContextValue(opts: { - spaceId?: string; - userId?: string; - now: string; -}): Record { - const contextValue: Record = { - timezone: "UTC", - surface: "ai_module", - currentDatetime: opts.now, - }; - if (opts.spaceId) contextValue.spaceId = opts.spaceId; - if (opts.userId) contextValue.userId = opts.userId; - return contextValue; -} - -/** - * Normalize OpenAI-style message content to a plain string. - * Accepts a string or content-parts array (`{ type:"text", text }` / `{ text }`). - * Previously only string content was accepted — array-shaped system/user messages - * (common from agent clients) were silently dropped, so system/jailbreak/agentic - * injects never reached Notion when any message used parts. - */ -function extractNotionMessageText(content: unknown): string { - if (typeof content === "string") return content; - if (!Array.isArray(content)) return ""; - const parts: string[] = []; - for (const p of content) { - if (typeof p === "string") { - if (p) parts.push(p); - continue; - } - if (!p || typeof p !== "object") continue; - const o = p as Record; - if (typeof o.text === "string" && o.text) parts.push(o.text); - else if (typeof o.content === "string" && o.content) parts.push(o.content); - } - return parts.join("\n"); -} - -/** Converts one OpenAI-style message into a transcript step, or `null` when it - * was folded into the context (system prompts). */ -function buildNotionMessageStep( - m: NotionMessage, - contextValue: Record, - opts: { userId?: string; now: string } -): Record | null { - // Accept string OR content-parts array (agent clients often send parts). - const text = extractNotionMessageText((m as { content?: unknown })?.content); - if (!text || text.length === 0) return null; - const role = (m.role || "").toLowerCase(); - - if (role === "system") { - // Fold system prompts into context instructions rather than a separate step. - const existing = typeof contextValue.instructions === "string" ? contextValue.instructions : ""; - contextValue.instructions = existing ? `${existing}\n${text}` : text; - return null; - } - - if (role === "assistant") { - return { - id: randomUUID(), - type: "agent-inference", - value: [{ type: "text", content: text }], - }; - } - - // user (and anything else treated as user) - const userStep: Record = { - id: randomUUID(), - type: "user", - value: [[text]], - createdAt: opts.now, - }; - if (opts.userId) userStep.userId = opts.userId; - return userStep; -} - -export function buildNotionTranscript( - messages: NotionMessage[], - opts: { - notionModel?: string; - spaceId?: string; - userId?: string; - } = {} -): Array> { - const trimmedModel = typeof opts.notionModel === "string" ? opts.notionModel.trim() : ""; - const model = trimmedModel && trimmedModel !== "notion-ai" ? trimmedModel : ""; - const now = isoNow(); - - const contextValue = buildNotionContextValue({ spaceId: opts.spaceId, userId: opts.userId, now }); - const entries: Array> = [ - buildNotionConfigStep(model), - { id: randomUUID(), type: "context", value: contextValue }, - ]; - - for (const m of messages) { - const step = buildNotionMessageStep(m, contextValue, { userId: opts.userId, now }); - if (step) entries.push(step); - } - return entries; -} - -/** Strip Notion's `` prefix and similar noise from answers. */ -export function sanitizeNotionAssistantText(text: string): string { - if (!text) return ""; - let clean = text.replace(/^\uFEFF/, "").trim(); - // Self-closing or paired lang tags at the start (and anywhere). - clean = clean.replace(/<\/?lang\b[^>]*\/?>/gi, ""); - clean = clean.replace(/<\/lang>/gi, ""); - // Incomplete leading ")) return ""; - return clean.trim(); -} - -/** Extract plain text from Notion's rich-text tuple value: `[[text, marks?]]`. */ -function extractRichText(value: unknown): string { - if (!Array.isArray(value)) return ""; - return value - .map((segment) => (Array.isArray(segment) && typeof segment[0] === "string" ? segment[0] : "")) - .join(""); -} - -function extractAgentInferenceText(value: unknown): string { - if (!Array.isArray(value)) return ""; - const parts: string[] = []; - for (const item of value) { - if (!item || typeof item !== "object" || Array.isArray(item)) continue; - const part = item as Record; - const t = typeof part.type === "string" ? part.type.toLowerCase() : ""; - if (t === "text" && typeof part.content === "string" && part.content) { - parts.push(part.content); - } - } - return parts.join(""); -} - -/** Unwraps `thread_message[key].value.value.step` from a Notion record-map entry. */ -function extractThreadMessageStep(msg: unknown): Record | null { - if (!msg || typeof msg !== "object") return null; - const valueWrapper = (msg as Record).value; - if (!valueWrapper || typeof valueWrapper !== "object") return null; - const inner = (valueWrapper as Record).value; - if (!inner || typeof inner !== "object") return null; - const step = (inner as Record).step; - if (!step || typeof step !== "object") return null; - return step as Record; -} - -/** Extracts the text carried by a single thread-message step, or "" if none. */ -function extractStepText(stepObj: Record): string { - const stepType = typeof stepObj.type === "string" ? stepObj.type : ""; - if (stepType === "agent-inference") { - return extractAgentInferenceText(stepObj.value); - } - if (stepType === "markdown-chat" && typeof stepObj.value === "string") { - return stepObj.value; - } - return ""; -} - -function extractFromRecordMap(recordMap: unknown): string { - if (!recordMap || typeof recordMap !== "object" || Array.isArray(recordMap)) return ""; - const tm = (recordMap as Record).thread_message; - if (!tm || typeof tm !== "object" || Array.isArray(tm)) return ""; - let best = ""; - for (const msg of Object.values(tm as Record)) { - const stepObj = extractThreadMessageStep(msg); - if (!stepObj) continue; - const text = extractStepText(stepObj); - if (text && text.length >= best.length) best = text; - } - return best; -} - -/** - * Parse Notion's NDJSON `runInferenceTranscript` response body. - * Supports: - * 1. Legacy rich-text tuples on `value` (cumulative snapshots) - * 2. Modern patch-start / patch streams (text / markdown-chat ops) - * 3. Terminal record-map with agent-inference steps (authoritative final) - */ -/** Accumulator threaded through {@link parseNotionInferenceStream}'s line parsing. */ -type NotionStreamState = { - lastLegacy: string; - lastPatchFinal: string; - lastIncremental: string; - lastRecordMap: string; -}; - -/** Applies one `patch` op (full text-part append / step append / incremental string) to state. */ -/** Full agent-inference text-part append: `o:"a", p:".../value/-"`. */ -function applyNotionValuePartAppend(v: unknown, state: NotionStreamState): void { - if (!v || typeof v !== "object" || Array.isArray(v)) return; - const part = v as Record; - if (part.type === "text" && typeof part.content === "string" && part.content) { - state.lastPatchFinal = part.content; - } - if (part.type === "markdown-chat" && typeof part.value === "string" && part.value) { - state.lastPatchFinal = part.value; - } -} - -/** Step append with markdown-chat / agent-inference: `o:"a", p:".../s/-"`. */ -function applyNotionStepAppend(v: unknown, state: NotionStreamState): void { - if (!v || typeof v !== "object" || Array.isArray(v)) return; - const step = v as Record; - if (step.type === "markdown-chat" && typeof step.value === "string" && step.value) { - state.lastPatchFinal = step.value; - } - if (step.type === "agent-inference") { - const text = extractAgentInferenceText(step.value); - if (text) state.lastPatchFinal = text; - } -} - -function applyNotionPatchOp(rawOp: unknown, state: NotionStreamState): void { - if (!rawOp || typeof rawOp !== "object") return; - const op = rawOp as Record; - const o = typeof op.o === "string" ? op.o : ""; - const p = typeof op.p === "string" ? op.p : ""; - const v = op.v; - - if (o === "a" && p.endsWith("/value/-")) { - applyNotionValuePartAppend(v, state); - } else if (o === "a" && p.endsWith("/s/-")) { - applyNotionStepAppend(v, state); - } else if ((o === "x" || o === "p") && p.includes("/value") && typeof v === "string" && v) { - // Incremental string patches - state.lastIncremental += v; - } -} - -/** Applies one parsed NDJSON record (markdown-chat / agent-inference / patch / record-map / legacy). */ -function applyNotionStreamRecord(rec: Record, state: NotionStreamState): void { - const type = typeof rec.type === "string" ? rec.type : ""; - - // 1) Direct markdown-chat event - if (type === "markdown-chat" && typeof rec.value === "string" && rec.value) { - state.lastPatchFinal = rec.value; - return; - } - - // 2) Direct agent-inference event - if (type === "agent-inference") { - const text = extractAgentInferenceText(rec.value); - if (text) state.lastPatchFinal = text; - return; - } - - // 3) Patch stream - if (type === "patch" && Array.isArray(rec.v)) { - for (const rawOp of rec.v) applyNotionPatchOp(rawOp, state); - return; - } - - // 4) record-map terminal - if (type === "record-map" || rec.recordMap) { - const text = extractFromRecordMap(rec.recordMap || rec); - if (text) state.lastRecordMap = text; - return; - } - - // 5) Legacy rich-text value (cumulative) - const rich = extractRichText(rec.value); - if (rich) state.lastLegacy = rich; -} - -/** Parses one raw NDJSON line (trims / strips SSE `data:` prefix / JSON-parses) into state. */ -function applyNotionStreamLine(rawLine: string, state: NotionStreamState): void { - const line = rawLine.trim(); - if (!line || line === "[DONE]") return; - // Strip optional SSE "data:" prefix if a proxy rewrote it. - const payloadLine = line.startsWith("data:") ? line.slice(5).trim() : line; - if (!payloadLine) return; - - let record: unknown; - try { - record = JSON.parse(payloadLine); - } catch { - return; - } - if (!record || typeof record !== "object" || Array.isArray(record)) return; - applyNotionStreamRecord(record as Record, state); -} - -/** - * Parse Notion's NDJSON `runInferenceTranscript` response body. - * Supports: - * 1. Legacy rich-text tuples on `value` (cumulative snapshots) - * 2. Modern patch-start / patch streams (text / markdown-chat ops) - * 3. Terminal record-map with agent-inference steps (authoritative final) - */ -export function parseNotionInferenceStream(raw: string): string { - if (!raw) return ""; - const state: NotionStreamState = { - lastLegacy: "", - lastPatchFinal: "", - lastIncremental: "", - lastRecordMap: "", - }; - - for (const rawLine of raw.split("\n")) { - applyNotionStreamLine(rawLine, state); - } - - const candidates = [ - state.lastRecordMap, - state.lastPatchFinal, - state.lastIncremental, - state.lastLegacy, - ] - .map(sanitizeNotionAssistantText) - .filter(Boolean); - // Prefer the longest non-empty candidate; record-map usually wins. - return candidates.sort((a, b) => b.length - a.length)[0] || ""; -} /** * Notion's undocumented inference API does not return token usage. @@ -501,7 +189,7 @@ export function estimateNotionUsage( content: string ): { prompt_tokens: number; completion_tokens: number; total_tokens: number; estimated: true } { const promptText = (messages || []) - .map((m) => (typeof m?.content === "string" ? m.content : "")) + .map((m) => extractNotionMessageText(m?.content)) .join("\n"); // ~4 chars/token (English-ish); at least 1 when there is any text. const prompt_tokens = promptText ? Math.max(1, Math.ceil(promptText.length / 4)) : 0; @@ -514,24 +202,39 @@ export function estimateNotionUsage( }; } -function chatCompletionResponse(content: string, model: string, messages?: NotionMessage[]) { +function chatCompletionResponse( + content: string, + model: string, + messages?: NotionMessage[], + threadId?: string +) { + const id = threadId ? `chatcmpl-notion-${threadId}` : `chatcmpl-notion-${Date.now()}`; return new Response( JSON.stringify({ - id: `chatcmpl-notion-${Date.now()}`, + id, object: "chat.completion", created: Math.floor(Date.now() / 1000), model, choices: [{ index: 0, message: { role: "assistant", content }, finish_reason: "stop" }], usage: estimateNotionUsage(messages, content), + // Non-standard but useful for clients that want to pin continuity explicitly + notion_thread_id: threadId || undefined, }), - { status: 200, headers: { "Content-Type": "application/json" } } + { + status: 200, + headers: { + "Content-Type": "application/json", + ...(threadId ? { "X-Notion-Thread-Id": threadId } : {}), + }, + } ); } -function pseudoStreamResponse(content: string, model: string) { +function pseudoStreamResponse(content: string, model: string, threadId?: string) { const encoder = new TextEncoder(); + const id = threadId ? `chatcmpl-notion-${threadId}` : `chatcmpl-notion-${Date.now()}`; const chunk = (delta: string, finishReason: string | null) => ({ - id: `chatcmpl-notion-${Date.now()}`, + id, object: "chat.completion.chunk", created: Math.floor(Date.now() / 1000), model, @@ -551,6 +254,7 @@ function pseudoStreamResponse(content: string, model: string) { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", Connection: "keep-alive", + ...(threadId ? { "X-Notion-Thread-Id": threadId } : {}), }, }); } @@ -582,32 +286,47 @@ async function resolveExecuteWorkspace( return { spaceId, userId }; } -/** Live-verified working shape (createThread:false without threadId → 400 ValidationError). */ -function buildNotionCreateThreadRequestBody(opts: { +/** + * Live-verified shape: + * - First turn: createThread true + new threadId + * - Follow-up: createThread false + same threadId (false without threadId → 400) + */ +function buildNotionInferenceRequestBody(opts: { spaceId: string; userId: string; threadId: string; transcript: unknown; + createThread: boolean; + agent?: NotionAgentOptions; }): Record { - const { spaceId, threadId, transcript } = opts; + const { spaceId, threadId, transcript, createThread, agent } = opts; + const isCustom = Boolean(agent?.workflowId); + const workflowId = agent?.workflowId || ""; + // Follow-ups: isPartialTranscript true matches open-source Notion bridges and + // avoids re-validating the entire prior transcript (a source of transient errors). + const isFollowUp = !createThread; return { traceId: randomUUID(), spaceId, threadId, - createThread: true, - generateTitle: true, + createThread, + // Only generate a title when starting a new Notion AI chat + generateTitle: createThread, asPatchResponse: true, - isPartialTranscript: false, + patchResponseVersion: 2, + isPartialTranscript: isFollowUp, saveAllThreadOperations: true, - setUnreadState: true, - createdSource: "ai_module", + setUnreadState: createThread, + createdSource: isCustom ? "custom_agent" : "ai_module", threadType: "workflow", + supportsCustomAgentNudgeTranscriptStep: true, + isUserInAnySalesAssistedSpace: false, + isSpaceSalesAssisted: false, transcript, - threadParentPointer: { - table: "space", - id: spaceId, - spaceId, - }, + // Default AI is parented by the workspace; custom agents by the workflow id. + threadParentPointer: isCustom + ? { table: "workflow", id: workflowId, spaceId } + : { table: "space", id: spaceId, spaceId }, debugOverrides: { annotationInferences: {}, cachedInferences: {}, @@ -621,14 +340,21 @@ function buildNotionExecuteHeaders(opts: { cookie: string; spaceId: string; userId: string; + agent?: NotionAgentOptions; }): Record { + const isCustom = Boolean(opts.agent?.workflowId); + // Browser uses /agent/?wfv=chat for custom agents. + const agentPathId = (opts.agent?.workflowId || "").replace(/-/g, ""); + const referer = isCustom && agentPathId + ? `${BASE_URL}/agent/${agentPathId}?wfv=chat` + : `${BASE_URL}/ai`; const reqHeaders: Record = { "Content-Type": "application/json", "User-Agent": USER_AGENT, Accept: "application/x-ndjson", Cookie: opts.cookie, Origin: BASE_URL, - Referer: `${BASE_URL}/ai`, + Referer: referer, "notion-client-version": NOTION_CLIENT_VERSION, "notion-audit-log-platform": "web", "x-notion-space-id": opts.spaceId, @@ -639,6 +365,81 @@ function buildNotionExecuteHeaders(opts: { return reqHeaders; } +/** Normalize a pasted workflow/agent id (with or without dashes). */ +export function normalizeNotionWorkflowId(raw: string | undefined | null): string { + const s = String(raw || "").trim(); + if (!s) return ""; + // URL path segment …/agent/?… or bare hex + const fromUrl = s.match(/\/agent\/([a-f0-9-]{20,})/i); + let id = fromUrl ? fromUrl[1]! : s; + id = id.replace(/[^a-f0-9-]/gi, ""); + // Insert dashes if 32 hex chars (no dashes) + const hex = id.replace(/-/g, ""); + if (/^[a-f0-9]{32}$/i.test(hex)) { + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`.toLowerCase(); + } + // Already UUID-like + if (/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(id)) { + return id.toLowerCase(); + } + return id; +} + +/** + * Read custom-agent workflow id + optional context page from credentials. + * Sources (priority): providerSpecificData → cookie pairs on apiKey + * (`workflow_id=…`, `notion_workflow_id=…`, `context_page_id=…`). + */ +export function resolveNotionAgentOptions( + credentials: ExecuteInput["credentials"], + cookie: string +): NotionAgentOptions { + const ps = credentials?.providerSpecificData; + const workflowFromPs = + readProviderSpecificString(ps, [ + "workflowId", + "workflow_id", + "notionWorkflowId", + "notion_workflow_id", + "agentId", + "agent_id", + ]) || ""; + const pageFromPs = + readProviderSpecificString(ps, [ + "contextPageId", + "context_page_id", + "notionContextPageId", + ]) || ""; + + const readCookie = (name: string): string => { + const m = cookie.match(new RegExp(`(?:^|;\\s*)${name}=([^;]+)`, "i")); + if (!m) return ""; + const raw = m[1]!.trim(); + try { + return decodeURIComponent(raw); + } catch { + return raw; + } + }; + + const workflowId = normalizeNotionWorkflowId( + workflowFromPs || + readCookie("workflow_id") || + readCookie("notion_workflow_id") || + readCookie("agent_id") + ); + const contextPageId = + pageFromPs || + readCookie("context_page_id") || + readCookie("notion_context_page_id") || + ""; + + return { + workflowId: workflowId || undefined, + contextPageId: contextPageId ? contextPageId.trim() : undefined, + }; +} + /** * Sends the createThread request to Notion and returns either the raw * inference text or an error result — callers just check `.errorResult`. @@ -715,6 +516,9 @@ export class NotionWebExecutor extends BaseExecutor { ); } + // Optional custom agent (workflowId). Empty → default Notion AI (not agentic-specific). + const agent = resolveNotionAgentOptions(credentials, cookie); + const messages = requestBody.messages || []; if (!messages.some((m) => m.role === "user")) { return makeErrorResult(400, "No user message found", body, NOTION_URL); @@ -738,32 +542,122 @@ export class NotionWebExecutor extends BaseExecutor { const clientFacing = clientFacingModelId(model); const modelId = clientFacing || notionCodename || "notion-ai"; - const threadId = randomUUID(); - const transcript = buildNotionTranscript(messages, { - notionModel: notionCodename || undefined, - spaceId, - userId: userId || undefined, - }); + // Thread continuity (sticky): + // - Prefer X-Notion-Thread-Id / body pin from the client + // - Else sticky root key from first user message (UREW-normalized, durable on disk) + // - Bind threadId *before* the upstream call so error retries never mint a new chat + // - createThread:true only for brand-new roots; never again for that root + const inboundHeaders = + (input.clientHeaders as Record | null | undefined) ?? + ((input as { headers?: Record }).headers as + | Record + | undefined); + const clientThreadId = readClientThreadId(requestBody, inboundHeaders ?? undefined); + // Namespace thread cache by custom agent so default AI and agents never share threads. + const threadSpaceKey = agent.workflowId ? `${spaceId}|wf:${agent.workflowId}` : spaceId; + const binding = resolveNotionThreadBinding(threadSpaceKey, messages, clientThreadId); + let { threadId, createThread, rootKey } = binding; - const reqBody = buildNotionCreateThreadRequestBody({ spaceId, userId, threadId, transcript }); - const reqHeaders = buildNotionExecuteHeaders({ cookie, spaceId, userId }); + const reqHeaders = buildNotionExecuteHeaders({ cookie, spaceId, userId, agent }); - const { rawText, errorResult } = await sendNotionInferenceRequest({ - reqBody, - reqHeaders, - signal, - }); - if (errorResult) return errorResult; + const runOnce = async (opts: { + createThread: boolean; + threadId: string; + }): Promise< + | { ok: true; finalText: string; reqBody: Record } + | { ok: false; errorResult: ReturnType; retryable: boolean; reqBody: Record } + > => { + const transcript = buildNotionTranscript(messages, { + notionModel: notionCodename || undefined, + spaceId, + userId: userId || undefined, + agent, + isFollowUp: !opts.createThread, + }); + const reqBody = buildNotionInferenceRequestBody({ + spaceId, + userId, + threadId: opts.threadId, + transcript, + createThread: opts.createThread, + agent, + }); - const finalText = parseNotionInferenceStream(rawText || ""); - if (!finalText) { - return makeErrorResult(502, "No response from Notion AI", reqBody, NOTION_URL); + if (opts.createThread) { + notionThreadMarkCreateAttempted(rootKey, opts.threadId); + } + + const { rawText, errorResult } = await sendNotionInferenceRequest({ + reqBody, + reqHeaders, + signal, + }); + + if (errorResult) { + // HTTP-level failure — keep sticky binding so the next turn reuses threadId + const status = errorResult.response?.status ?? 502; + const retryable = status === 429 || status === 503 || status >= 500; + return { ok: false, errorResult, retryable, reqBody }; + } + + const raw = rawText || ""; + const upstreamErr = extractNotionUpstreamError(raw); + if (upstreamErr) { + // In-band Notion error (often HTTP 200 NDJSON). Sticky thread stays bound. + const status = upstreamErr.isRetryable ? 503 : 502; + return { + ok: false, + retryable: upstreamErr.isRetryable, + reqBody, + errorResult: makeErrorResult( + status, + `Notion ${upstreamErr.subType || "error"}: ${upstreamErr.message}`, + reqBody, + NOTION_URL + ), + }; + } + + const finalText = parseNotionInferenceStream(raw); + if (!finalText) { + return { + ok: false, + retryable: true, + reqBody, + errorResult: makeErrorResult(502, "No response from Notion AI", reqBody, NOTION_URL), + }; + } + + return { ok: true, finalText, reqBody }; + }; + + // First attempt + let attempt = await runOnce({ createThread, threadId }); + + // One automatic retry for transient Notion faults — same threadId, never create again + if (!attempt.ok && attempt.retryable) { + const delayMs = process.env.NODE_ENV === "test" || process.env.VITEST ? 20 : 700 + Math.floor(Math.random() * 400); + await new Promise((r) => setTimeout(r, delayMs)); + attempt = await runOnce({ createThread: false, threadId }); } - const response = wantStream - ? pseudoStreamResponse(finalText, modelId) - : chatCompletionResponse(finalText, modelId, messages); + if (!attempt.ok) { + return attempt.errorResult; + } - return { response, url: NOTION_URL, headers: reqHeaders, transformedBody: reqBody }; + // Confirm sticky binding + prefix keys for multi-turn continuity + notionThreadMarkConfirmed(rootKey, threadId); + notionThreadSessionStore(threadSpaceKey, messages, attempt.finalText, threadId); + + const response = wantStream + ? pseudoStreamResponse(attempt.finalText, modelId, threadId) + : chatCompletionResponse(attempt.finalText, modelId, messages, threadId); + + return { + response, + url: NOTION_URL, + headers: reqHeaders, + transformedBody: attempt.reqBody, + }; } } diff --git a/open-sse/services/notionStreamParser.ts b/open-sse/services/notionStreamParser.ts new file mode 100644 index 0000000000..6a7790962c --- /dev/null +++ b/open-sse/services/notionStreamParser.ts @@ -0,0 +1,275 @@ +/** + * Notion AI Web — NDJSON `runInferenceTranscript` response parsing. + * + * Extracted from `executors/notion-web.ts` (file-size gate) — parses Notion's + * undocumented streaming response format (legacy rich-text tuples, patch-start / + * patch ops, and terminal record-map agent-inference steps) into plain text, and + * detects in-band Notion error objects (often shipped with HTTP 200). + */ + +/** Strips lang tags / BOM noise Notion sometimes wraps assistant text in. */ +export function sanitizeNotionAssistantText(text: string): string { + if (!text) return ""; + let clean = text.replace(/^\uFEFF/, "").trim(); + // Self-closing or paired lang tags at the start (and anywhere). + clean = clean.replace(/<\/?lang\b[^>]*\/?>/gi, ""); + clean = clean.replace(/<\/lang>/gi, ""); + // Incomplete leading ")) return ""; + return clean.trim(); +} + +/** Extract plain text from Notion's rich-text tuple value: `[[text, marks?]]`. */ +function extractRichText(value: unknown): string { + if (!Array.isArray(value)) return ""; + return value + .map((segment) => (Array.isArray(segment) && typeof segment[0] === "string" ? segment[0] : "")) + .join(""); +} + +function extractAgentInferenceText(value: unknown): string { + if (!Array.isArray(value)) return ""; + const parts: string[] = []; + for (const item of value) { + if (!item || typeof item !== "object" || Array.isArray(item)) continue; + const part = item as Record; + const t = typeof part.type === "string" ? part.type.toLowerCase() : ""; + if (t === "text" && typeof part.content === "string" && part.content) { + parts.push(part.content); + } + } + return parts.join(""); +} + +/** Unwraps `thread_message[key].value.value.step` from a Notion record-map entry. */ +function extractThreadMessageStep(msg: unknown): Record | null { + if (!msg || typeof msg !== "object") return null; + const valueWrapper = (msg as Record).value; + if (!valueWrapper || typeof valueWrapper !== "object") return null; + const inner = (valueWrapper as Record).value; + if (!inner || typeof inner !== "object") return null; + const step = (inner as Record).step; + if (!step || typeof step !== "object") return null; + return step as Record; +} + +/** Extracts the text carried by a single thread-message step, or "" if none. */ +function extractStepText(stepObj: Record): string { + const stepType = typeof stepObj.type === "string" ? stepObj.type : ""; + if (stepType === "agent-inference") { + return extractAgentInferenceText(stepObj.value); + } + if (stepType === "markdown-chat" && typeof stepObj.value === "string") { + return stepObj.value; + } + return ""; +} + +function extractFromRecordMap(recordMap: unknown): string { + if (!recordMap || typeof recordMap !== "object" || Array.isArray(recordMap)) return ""; + const tm = (recordMap as Record).thread_message; + if (!tm || typeof tm !== "object" || Array.isArray(tm)) return ""; + let best = ""; + for (const msg of Object.values(tm as Record)) { + const stepObj = extractThreadMessageStep(msg); + if (!stepObj) continue; + const text = extractStepText(stepObj); + if (text && text.length >= best.length) best = text; + } + return best; +} + +/** Accumulator threaded through {@link parseNotionInferenceStream}'s line parsing. */ +type NotionStreamState = { + lastLegacy: string; + lastPatchFinal: string; + lastIncremental: string; + lastRecordMap: string; +}; + +/** Full agent-inference text-part append: `o:"a", p:".../value/-"`. */ +function applyNotionValuePartAppend(v: unknown, state: NotionStreamState): void { + if (!v || typeof v !== "object" || Array.isArray(v)) return; + const part = v as Record; + if (part.type === "text" && typeof part.content === "string" && part.content) { + state.lastPatchFinal = part.content; + } + if (part.type === "markdown-chat" && typeof part.value === "string" && part.value) { + state.lastPatchFinal = part.value; + } +} + +/** Step append with markdown-chat / agent-inference: `o:"a", p:".../s/-"`. */ +function applyNotionStepAppend(v: unknown, state: NotionStreamState): void { + if (!v || typeof v !== "object" || Array.isArray(v)) return; + const step = v as Record; + if (step.type === "markdown-chat" && typeof step.value === "string" && step.value) { + state.lastPatchFinal = step.value; + } + if (step.type === "agent-inference") { + const text = extractAgentInferenceText(step.value); + if (text) state.lastPatchFinal = text; + } +} + +function applyNotionPatchOp(rawOp: unknown, state: NotionStreamState): void { + if (!rawOp || typeof rawOp !== "object") return; + const op = rawOp as Record; + const o = typeof op.o === "string" ? op.o : ""; + const p = typeof op.p === "string" ? op.p : ""; + const v = op.v; + + if (o === "a" && p.endsWith("/value/-")) { + applyNotionValuePartAppend(v, state); + } else if (o === "a" && p.endsWith("/s/-")) { + applyNotionStepAppend(v, state); + } else if ((o === "x" || o === "p") && p.includes("/value") && typeof v === "string" && v) { + // Incremental string patches + state.lastIncremental += v; + } +} + +/** Applies one parsed NDJSON record (markdown-chat / agent-inference / patch / record-map / legacy). */ +function applyNotionStreamRecord(rec: Record, state: NotionStreamState): void { + const type = typeof rec.type === "string" ? rec.type : ""; + + // 1) Direct markdown-chat event + if (type === "markdown-chat" && typeof rec.value === "string" && rec.value) { + state.lastPatchFinal = rec.value; + return; + } + + // 2) Direct agent-inference event + if (type === "agent-inference") { + const text = extractAgentInferenceText(rec.value); + if (text) state.lastPatchFinal = text; + return; + } + + // 3) Patch stream + if (type === "patch" && Array.isArray(rec.v)) { + for (const rawOp of rec.v) applyNotionPatchOp(rawOp, state); + return; + } + + // 4) record-map terminal + if (type === "record-map" || rec.recordMap) { + const text = extractFromRecordMap(rec.recordMap || rec); + if (text) state.lastRecordMap = text; + return; + } + + // 5) Legacy rich-text value (cumulative) + const rich = extractRichText(rec.value); + if (rich) state.lastLegacy = rich; +} + +/** Parses one raw NDJSON line (trims / strips SSE `data:` prefix / JSON-parses) into state. */ +function applyNotionStreamLine(rawLine: string, state: NotionStreamState): void { + const line = rawLine.trim(); + if (!line || line === "[DONE]") return; + // Strip optional SSE "data:" prefix if a proxy rewrote it. + const payloadLine = line.startsWith("data:") ? line.slice(5).trim() : line; + if (!payloadLine) return; + + let record: unknown; + try { + record = JSON.parse(payloadLine); + } catch { + return; + } + if (!record || typeof record !== "object" || Array.isArray(record)) return; + applyNotionStreamRecord(record as Record, state); +} + +/** + * Parse Notion's NDJSON `runInferenceTranscript` response body. + * Supports: + * 1. Legacy rich-text tuples on `value` (cumulative snapshots) + * 2. Modern patch-start / patch streams (text / markdown-chat ops) + * 3. Terminal record-map with agent-inference steps (authoritative final) + */ +export function parseNotionInferenceStream(raw: string): string { + if (!raw) return ""; + const state: NotionStreamState = { + lastLegacy: "", + lastPatchFinal: "", + lastIncremental: "", + lastRecordMap: "", + }; + + for (const rawLine of raw.split("\n")) { + applyNotionStreamLine(rawLine, state); + } + + const candidates = [ + state.lastRecordMap, + state.lastPatchFinal, + state.lastIncremental, + state.lastLegacy, + ] + .map(sanitizeNotionAssistantText) + .filter(Boolean); + // Prefer the longest non-empty candidate; record-map usually wins. + return candidates.sort((a, b) => b.length - a.length)[0] || ""; +} + +/** + * Detect Notion in-band errors (often HTTP 200 with NDJSON/JSON error objects), + * e.g. `{ type:"error", subType:"temporarily-unavailable", message:"…" }`. + */ +export function extractNotionUpstreamError(raw: string): { + message: string; + subType?: string; + isRetryable: boolean; +} | null { + if (!raw || !raw.trim()) return null; + const tryParse = (s: string): Record | null => { + try { + const o = JSON.parse(s) as Record; + return o && typeof o === "object" ? o : null; + } catch { + return null; + } + }; + + const candidates: Record[] = []; + const whole = tryParse(raw.trim()); + if (whole) candidates.push(whole); + for (const line of raw.split("\n")) { + const t = line.trim(); + if (!t) continue; + const o = tryParse(t); + if (o) candidates.push(o); + } + + for (const o of candidates) { + const type = typeof o.type === "string" ? o.type.toLowerCase() : ""; + const subType = typeof o.subType === "string" ? o.subType : undefined; + const message = + (typeof o.message === "string" && o.message) || + (typeof o.error === "string" && o.error) || + ""; + const isError = + type === "error" || + Boolean(subType) || + (typeof o.isRetryable === "boolean" && message.toLowerCase().includes("went wrong")); + if (!isError && !subType) continue; + + const sub = (subType || "").toLowerCase(); + const retryable = + o.isRetryable === true || + sub.includes("temporarily") || + sub.includes("unavailable") || + sub.includes("rate") || + sub.includes("timeout") || + sub.includes("overloaded"); + + return { + message: message || subType || "Notion upstream error", + subType, + isRetryable: retryable, + }; + } + return null; +} diff --git a/open-sse/services/notionThreadSessions.ts b/open-sse/services/notionThreadSessions.ts new file mode 100644 index 0000000000..de6378e745 --- /dev/null +++ b/open-sse/services/notionThreadSessions.ts @@ -0,0 +1,419 @@ +/** + * Notion AI Web — thread session continuity (OpenAI multi-turn → one Notion chat). + * + * Extracted from `executors/notion-web.ts` (file-size gate) — everything needed to + * bind an OpenAI-style multi-turn conversation to a single Notion `threadId` + * instead of minting a fresh Notion chat on every request. See + * `executors/notion-web.ts` for the upstream transcript/response translation + * that consumes this module. + * + * - History-keyed in-memory session cache (spaceId + conversation prefix hash), + * backed by an on-disk snapshot under DATA_DIR so continuity survives restarts. + * - Sticky root binding written *before* the upstream call so error retries never + * mint a second Notion chat for the same conversation. + * - Optional client-supplied continuity via body (`notion_thread_id`/`thread_id`) + * or the `X-Notion-Thread-Id` header (via `ExecuteInput.clientHeaders`). + */ +import { randomUUID } from "node:crypto"; +import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { dirname, join } from "node:path"; + +export interface NotionMessage { + role: string; + /** OpenAI string content OR content-parts array — normalized by extractNotionMessageText. */ + content: unknown; +} + +/** Minimal shape readClientThreadId needs from the OpenAI-style request body. */ +export interface NotionThreadRequestBody { + notion_thread_id?: string; + thread_id?: string; +} + +/** + * Normalize OpenAI-style message content to a plain string. + * Accepts a string or content-parts array (`{ type:"text", text }` / `{ text }`). + * Previously only string content was accepted — array-shaped system/user messages + * (common from agent clients) were silently dropped, so system/jailbreak/agentic + * injects never reached Notion when any message used parts. + */ +export function extractNotionMessageText(content: unknown): string { + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + const parts: string[] = []; + for (const p of content) { + if (typeof p === "string") { + if (p) parts.push(p); + continue; + } + if (!p || typeof p !== "object") continue; + const o = p as Record; + if (typeof o.text === "string" && o.text) parts.push(o.text); + else if (typeof o.content === "string" && o.content) parts.push(o.content); + } + return parts.join("\n"); +} + +const THREAD_SESSION_MAX_AGE_MS = 6 * 3600_000; // 6h — agent tool loops can be long +const THREAD_SESSION_MAX_ENTRIES = 500; + +interface ThreadSessionEntry { + threadId: string; + ts: number; + /** True once we successfully completed at least one turn on this thread. */ + confirmed?: boolean; + /** True once we issued createThread:true for this threadId (even if the reply failed). */ + createAttempted?: boolean; +} + +/** In-memory map: conversation key → Notion threadId. Backed by DATA_DIR when available. */ +const threadSessionCache = new Map(); +let threadStoreLoaded = false; +let threadStoreDirty = false; +let threadStoreTimer: ReturnType | null = null; + +function getThreadStorePath(): string | null { + try { + const dataDir = + process.env.DATA_DIR || + process.env.OMNIROUTE_DATA_DIR || + process.env.VIBEPROXY_DATA_DIR || + ""; + if (!dataDir) return null; + return join(dataDir, "notion-web-thread-sessions.json"); + } catch { + return null; + } +} + +function loadThreadStoreFromDisk(): void { + if (threadStoreLoaded) return; + threadStoreLoaded = true; + const path = getThreadStorePath(); + if (!path || !existsSync(path)) return; + try { + const raw = readFileSync(path, "utf8"); + const parsed = JSON.parse(raw) as Record; + const now = Date.now(); + for (const [k, v] of Object.entries(parsed || {})) { + if (!v?.threadId || typeof v.ts !== "number") continue; + if (now - v.ts > THREAD_SESSION_MAX_AGE_MS) continue; + threadSessionCache.set(k, v); + } + } catch { + // corrupt store — start fresh + } +} + +function scheduleThreadStoreFlush(): void { + threadStoreDirty = true; + if (threadStoreTimer) return; + threadStoreTimer = setTimeout(() => { + threadStoreTimer = null; + flushThreadStoreToDisk(); + }, 250); + // Don't keep the process alive solely for the flush. + if (typeof threadStoreTimer === "object" && threadStoreTimer && "unref" in threadStoreTimer) { + try { + (threadStoreTimer as NodeJS.Timeout).unref(); + } catch { + /* ignore */ + } + } +} + +function flushThreadStoreToDisk(): void { + if (!threadStoreDirty) return; + const path = getThreadStorePath(); + if (!path) return; + try { + const dir = dirname(path); + if (!existsSync(dir)) mkdirSync(dir, { recursive: true }); + const obj: Record = {}; + for (const [k, v] of threadSessionCache) obj[k] = v; + writeFileSync(path, JSON.stringify(obj), "utf8"); + threadStoreDirty = false; + } catch { + // best-effort persistence + } +} + +/** Exported for unit tests. */ +export function __resetNotionThreadSessionsForTests(): void { + threadSessionCache.clear(); + threadStoreLoaded = true; // skip disk reload in tests + threadStoreDirty = false; + if (threadStoreTimer) { + clearTimeout(threadStoreTimer); + threadStoreTimer = null; + } +} + +/** + * Normalize user/assistant text for thread-cache hashing. + * + * SkillsManager / OpenAI clients keep the *original* user text in history, while + * VibeProxy agentic conversion may rewrite the last user turn (UREW pin with + * "My current task: …"). Without normalization, turn-2 lookup never matches + * turn-1 store → createThread:true every request (new Notion chat each time). + */ +export function normalizeNotionContentForHash(content: unknown): string { + let text = extractNotionMessageText(content).replace(/\r\n/g, "\n").trim(); + if (!text) return ""; + + // Agentic / UREW pin: keep only the stable task suffix when present. + const taskMarkers = ["My current task:", "my current task:"]; + for (const marker of taskMarkers) { + const idx = text.lastIndexOf(marker); + if (idx >= 0) { + text = text.slice(idx + marker.length).trim(); + break; + } + } + + // Drop other common agentic preamble fingerprints if the whole pin leaked in. + if (text.includes("local workflow automation tool") || text.includes("clipboard parser")) { + const intentIdx = text.lastIndexOf("Intent:"); + // Prefer last non-empty line after stripping long preambles + const lines = text + .split("\n") + .map((l) => l.trim()) + .filter(Boolean); + if (lines.length > 0) text = lines[lines.length - 1]!; + void intentIdx; + } + + return text.replace(/\s+/g, " ").trim(); +} + +/** FNV-1a style hash of spaceId + normalized message list (conversation prefix). */ +export function hashNotionConversation(spaceId: string, msgs: NotionMessage[]): string { + const parts = [ + `space:${spaceId}`, + ...msgs.map((h) => `${(h.role || "").toLowerCase()}:${normalizeNotionContentForHash(h.content)}`), + ]; + const raw = parts.join("\n"); + let hash = 0x811c9dc5; + for (let i = 0; i < raw.length; i++) { + hash ^= raw.charCodeAt(i); + hash = Math.imul(hash, 0x01000193) >>> 0; + } + return hash.toString(16).padStart(8, "0"); +} + +/** Everything before the last user message (empty ⇒ first user turn / new thread). */ +export function conversationPrefixBeforeLastUser(messages: NotionMessage[]): NotionMessage[] { + if (!messages.length) return []; + let lastUser = -1; + for (let i = messages.length - 1; i >= 0; i--) { + const role = (messages[i]?.role || "").toLowerCase(); + if (role === "user" || role === "human") { + lastUser = i; + break; + } + } + if (lastUser <= 0) return []; + return messages.slice(0, lastUser); +} + +function readThreadSessionEntry(key: string): ThreadSessionEntry | null { + loadThreadStoreFromDisk(); + const entry = threadSessionCache.get(key); + if (!entry) return null; + if (Date.now() - entry.ts > THREAD_SESSION_MAX_AGE_MS) { + threadSessionCache.delete(key); + scheduleThreadStoreFlush(); + return null; + } + return entry; +} + +function readThreadSession(key: string): string | null { + return readThreadSessionEntry(key)?.threadId ?? null; +} + +function putThreadSession( + key: string, + threadId: string, + flags: { confirmed?: boolean; createAttempted?: boolean } = {} +): void { + loadThreadStoreFromDisk(); + const prev = threadSessionCache.get(key); + threadSessionCache.set(key, { + threadId, + ts: Date.now(), + confirmed: flags.confirmed ?? prev?.confirmed ?? false, + createAttempted: flags.createAttempted ?? prev?.createAttempted ?? false, + }); + // Evict oldest if over cap + if (threadSessionCache.size > THREAD_SESSION_MAX_ENTRIES) { + let oldestKey: string | null = null; + let oldestTs = Infinity; + for (const [k, v] of threadSessionCache) { + if (v.ts < oldestTs) { + oldestTs = v.ts; + oldestKey = k; + } + } + if (oldestKey) threadSessionCache.delete(oldestKey); + } + scheduleThreadStoreFlush(); +} + +/** Root sticky key for a conversation (space/agent + first user turn). */ +export function notionThreadRootKey(spaceKey: string, messages: NotionMessage[]): string | null { + const first = firstUserMessage(messages); + if (!first) return null; + return `root:${hashNotionConversation(spaceKey, [first])}`; +} + +/** + * Resolve which Notion thread to use and whether to mint a new one. + * - Sticky root binding is written *before* the upstream call so errors/retries + * never open a second Notion chat for the same conversation. + * - Any prior assistant history forces createThread:false when a sticky id exists. + */ +export function resolveNotionThreadBinding( + spaceKey: string, + messages: NotionMessage[], + clientThreadId?: string +): { threadId: string; createThread: boolean; rootKey: string | null } { + loadThreadStoreFromDisk(); + const rootKey = notionThreadRootKey(spaceKey, messages); + const hasHistory = conversationHasAssistant(messages); + + if (clientThreadId && clientThreadId.trim()) { + const id = clientThreadId.trim(); + if (rootKey) putThreadSession(rootKey, id, { createAttempted: true }); + return { threadId: id, createThread: false, rootKey }; + } + + // Prefer sticky root (survives UREW rewrites + error retries) + if (rootKey) { + const sticky = readThreadSessionEntry(rootKey); + if (sticky?.threadId) { + // Touch TTL + putThreadSession(rootKey, sticky.threadId, { + confirmed: sticky.confirmed, + createAttempted: sticky.createAttempted, + }); + // If we already attempted create for this root, never create again + // (even when the first reply failed — Notion may already have the thread). + const createThread = !sticky.createAttempted && !sticky.confirmed && !hasHistory; + return { + threadId: sticky.threadId, + createThread, + rootKey, + }; + } + } + + // Exact prefix match (full history before last user) + const prefix = conversationPrefixBeforeLastUser(messages); + if (prefix.length > 0) { + const exactId = readThreadSession(hashNotionConversation(spaceKey, prefix)); + if (exactId) { + if (rootKey) putThreadSession(rootKey, exactId, { createAttempted: true, confirmed: true }); + return { threadId: exactId, createThread: false, rootKey }; + } + } + + // Mint a new thread id and bind it immediately (optimistic) so concurrent / + // failed retries reuse the same id instead of spam-creating Notion chats. + const threadId = randomUUID(); + if (rootKey) { + putThreadSession(rootKey, threadId, { + createAttempted: false, + confirmed: false, + }); + } + // Multi-turn history without sticky (e.g. process restart): still create once + // with the full transcript so the agent can continue in a fresh Notion chat. + return { threadId, createThread: true, rootKey }; +} + +/** Mark that we sent createThread:true for this root (even if the body errored). */ +export function notionThreadMarkCreateAttempted(rootKey: string | null, threadId: string): void { + if (!rootKey || !threadId) return; + putThreadSession(rootKey, threadId, { createAttempted: true }); +} + +/** Mark successful inference on this thread. */ +export function notionThreadMarkConfirmed(rootKey: string | null, threadId: string): void { + if (!rootKey || !threadId) return; + putThreadSession(rootKey, threadId, { createAttempted: true, confirmed: true }); +} + +function firstUserMessage(messages: NotionMessage[]): NotionMessage | null { + for (const m of messages) { + const role = (m?.role || "").toLowerCase(); + if (role === "user" || role === "human") return m; + } + return null; +} + +function conversationHasAssistant(messages: NotionMessage[]): boolean { + return messages.some((m) => { + const role = (m?.role || "").toLowerCase(); + return role === "assistant" || role === "ai" || role === "model"; + }); +} + +/** Lookup-only (does not mint). Used by tests and diagnostics. */ +export function notionThreadSessionLookup(spaceId: string, messages: NotionMessage[]): string | null { + loadThreadStoreFromDisk(); + const rootKey = notionThreadRootKey(spaceId, messages); + if (rootKey) { + const sticky = readThreadSession(rootKey); + if (sticky) return sticky; + } + const prefix = conversationPrefixBeforeLastUser(messages); + if (prefix.length === 0) return null; + return readThreadSession(hashNotionConversation(spaceId, prefix)); +} + +/** + * After a successful turn, remember threadId under the completed conversation + * (request messages + this assistant reply) so the next OpenAI multi-turn request + * whose prefix matches that history reuses the same Notion chat. + */ +export function notionThreadSessionStore( + spaceId: string, + messages: NotionMessage[], + assistantText: string, + threadId: string +): void { + if (!threadId || !spaceId) return; + const full: NotionMessage[] = [...messages, { role: "assistant", content: assistantText }]; + putThreadSession(hashNotionConversation(spaceId, full), threadId, { + confirmed: true, + createAttempted: true, + }); + + // Root key for agent multi-turn clients that keep original user wording. + const rootKey = notionThreadRootKey(spaceId, messages); + if (rootKey) { + putThreadSession(rootKey, threadId, { confirmed: true, createAttempted: true }); + } + void assistantText; +} + +/** Client-supplied thread continuity pin: body (`notion_thread_id`/`thread_id`) or + * the `X-Notion-Thread-Id` header (case-insensitive). */ +export function readClientThreadId( + body: NotionThreadRequestBody, + headers?: Record +): string { + const fromBody = + (typeof body.notion_thread_id === "string" && body.notion_thread_id.trim()) || + (typeof body.thread_id === "string" && body.thread_id.trim()) || + ""; + if (fromBody) return fromBody; + if (!headers) return ""; + for (const [k, v] of Object.entries(headers)) { + if (k.toLowerCase() === "x-notion-thread-id" && typeof v === "string" && v.trim()) { + return v.trim(); + } + } + return ""; +} diff --git a/open-sse/services/notionTranscriptBuilder.ts b/open-sse/services/notionTranscriptBuilder.ts new file mode 100644 index 0000000000..3fa7d7b981 --- /dev/null +++ b/open-sse/services/notionTranscriptBuilder.ts @@ -0,0 +1,194 @@ +/** + * Notion AI Web — `runInferenceTranscript` transcript construction. + * + * Extracted from `executors/notion-web.ts` (file-size gate) — builds a Notion + * transcript array (`config` + `context` + per-message steps) from OpenAI-style + * chat messages. + * + * Live contract (verified 2026-07-19): + * - Leading `config` (workflow + optional model food-codename) + * - Leading `context` (spaceId / userId / surface / timezone) + * - User turns as `type: "user"` (legacy `human` also works with createThread, + * but `user` matches the current web client) + * - Assistant turns as `agent-inference` text parts + */ +import { randomUUID } from "node:crypto"; +import { extractNotionMessageText, type NotionMessage } from "./notionThreadSessions.ts"; + +/** Custom Notion AI agent (workflow) options from account credential / providerSpecificData. */ +export interface NotionAgentOptions { + /** UUID of a custom agent workflow. Empty = default Notion AI (ai_module). */ + workflowId?: string; + /** Optional context page id for custom agents. */ + contextPageId?: string; +} + +function isoNow(): string { + // Millisecond precision matches the browser client. + return new Date().toISOString().replace(/\.\d{3}Z$/, (m) => m); // keep ms + Z +} + +function buildNotionConfigStep(model: string, agent?: NotionAgentOptions): Record { + const isCustom = Boolean(agent?.workflowId); + const configValue: Record = { + type: "workflow", + // Match live browser defaults (2026-07-20 capture) for fewer plan/feature mismatches. + enableAgentAutomations: true, + enableAgentIntegrations: true, + enableCustomAgents: true, + enableScriptAgent: true, + enableAgentDiffs: true, + enableCsvAttachmentSupport: true, + enableComputer: true, + enableCreateAndRunThread: true, + enableAgentGenerateImage: !isCustom, + useWebSearch: true, + searchScopes: [{ type: "everything" }], + availableConnectors: [], + enableUserSessionContext: false, + isCustomAgent: isCustom, + isCustomAgentBuilder: false, + isCustomAgentCreate: false, + isAgentResearchRequest: false, + useCustomAgentDraft: isCustom, + modelFromUser: !isCustom && Boolean(model), + databaseAgentConfigMode: false, + isOnboardingAgent: false, + isMobile: false, + }; + if (isCustom && agent?.workflowId) { + configValue.workflowId = agent.workflowId; + } + // Default Notion AI: pin the food codename when the client selected a model. + // Custom agents usually use the agent-configured model (modelFromUser:false). + if (!isCustom && model) configValue.model = model; + return { id: randomUUID(), type: "config", value: configValue }; +} + +function buildNotionContextValue(opts: { + spaceId?: string; + userId?: string; + now: string; + agent?: NotionAgentOptions; +}): Record { + const isCustom = Boolean(opts.agent?.workflowId); + const contextValue: Record = { + timezone: "UTC", + surface: isCustom ? "custom_agent" : "ai_module", + currentDatetime: opts.now, + }; + if (opts.spaceId) contextValue.spaceId = opts.spaceId; + if (opts.userId) contextValue.userId = opts.userId; + if (isCustom && opts.agent?.workflowId) { + contextValue.workflowId = opts.agent.workflowId; + if (opts.agent.contextPageId) { + contextValue.context_page_id = opts.agent.contextPageId; + } + } + return contextValue; +} + +/** Converts one OpenAI-style message into a transcript step, or `null` when it + * was folded into the context (system prompts). */ +function buildNotionMessageStep( + m: NotionMessage, + contextValue: Record, + opts: { userId?: string; now: string } +): Record | null { + // Accept string OR content-parts array (agent clients often send parts). + const text = extractNotionMessageText((m as { content?: unknown })?.content); + if (!text || text.length === 0) return null; + const role = (m.role || "").toLowerCase(); + + if (role === "system") { + // Fold system prompts into context instructions rather than a separate step. + const existing = typeof contextValue.instructions === "string" ? contextValue.instructions : ""; + contextValue.instructions = existing ? `${existing}\n${text}` : text; + return null; + } + + if (role === "assistant") { + return { + id: randomUUID(), + type: "agent-inference", + value: [{ type: "text", content: text }], + }; + } + + // user (and anything else treated as user) + const userStep: Record = { + id: randomUUID(), + type: "user", + value: [[text]], + createdAt: opts.now, + }; + if (opts.userId) userStep.userId = opts.userId; + return userStep; +} + +/** + * For follow-ups, only send steps after the last assistant turn (partial transcript). + * Notion already has prior steps when createThread:false + sticky threadId. + * Re-sending the entire agent tool loop every turn triggers temporarily-unavailable. + */ +export function messagesForNotionTranscript( + messages: NotionMessage[], + isFollowUp: boolean +): NotionMessage[] { + if (!isFollowUp || !messages.length) return messages; + let lastAsst = -1; + for (let i = messages.length - 1; i >= 0; i--) { + const role = (messages[i]?.role || "").toLowerCase(); + if (role === "assistant" || role === "ai" || role === "model") { + lastAsst = i; + break; + } + } + if (lastAsst < 0) return messages; + const slice = messages.slice(lastAsst + 1); + // Always include at least the last user message + if (slice.length === 0) { + const lastUser = [...messages].reverse().find((m) => { + const r = (m.role || "").toLowerCase(); + return r === "user" || r === "human"; + }); + return lastUser ? [lastUser] : messages; + } + return slice; +} + +export function buildNotionTranscript( + messages: NotionMessage[], + opts: { + notionModel?: string; + spaceId?: string; + userId?: string; + agent?: NotionAgentOptions; + /** When true, only append steps after the last assistant (partial follow-up). */ + isFollowUp?: boolean; + } = {} +): Array> { + const trimmedModel = typeof opts.notionModel === "string" ? opts.notionModel.trim() : ""; + const model = trimmedModel && trimmedModel !== "notion-ai" ? trimmedModel : ""; + const now = isoNow(); + const agent = opts.agent?.workflowId ? opts.agent : undefined; + const isFollowUp = Boolean(opts.isFollowUp); + + const contextValue = buildNotionContextValue({ + spaceId: opts.spaceId, + userId: opts.userId, + now, + agent, + }); + const entries: Array> = [ + buildNotionConfigStep(model, agent), + { id: randomUUID(), type: "context", value: contextValue }, + ]; + + const msgs = messagesForNotionTranscript(messages, isFollowUp); + for (const m of msgs) { + const step = buildNotionMessageStep(m, contextValue, { userId: opts.userId, now }); + if (step) entries.push(step); + } + return entries; +} diff --git a/tests/unit/executor-notion-web-thread-sessions.test.ts b/tests/unit/executor-notion-web-thread-sessions.test.ts new file mode 100644 index 0000000000..9af09c1a02 --- /dev/null +++ b/tests/unit/executor-notion-web-thread-sessions.test.ts @@ -0,0 +1,330 @@ +// Split out of executor-notion-web.test.ts (file-size gate) — Notion AI Web +// thread session continuity: sticky root binding, prefix-hash lookup/store, +// error-retry stickiness, and the OpenAI multi-turn createThread flip +// (createThread:true on turn 1, createThread:false + same threadId on turn 2+). +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; + +const mod = await import("../../open-sse/executors/notion-web.ts"); + +const COOKIE_WITH_SPACE = "token_v2=xyz; space_id=space-1; notion_user_id=user-1"; + +describe("Notion thread session continuity", () => { + const { + __resetNotionThreadSessionsForTests, + conversationPrefixBeforeLastUser, + hashNotionConversation, + notionThreadSessionLookup, + notionThreadSessionStore, + } = mod; + + it("first user turn has no prior assistant history (lookup misses)", () => { + assert.deepEqual( + conversationPrefixBeforeLastUser([{ role: "user", content: "hi" }]), + [] + ); + // System-only prefix is fine — still no stored thread for a first user turn + const withSys = [ + { role: "system", content: "sys" }, + { role: "user", content: "hi" }, + ]; + assert.deepEqual(conversationPrefixBeforeLastUser(withSys), [ + { role: "system", content: "sys" }, + ]); + __resetNotionThreadSessionsForTests(); + assert.equal(notionThreadSessionLookup("space-1", withSys), null); + }); + + it("prefix includes prior turns for multi-turn OpenAI history", () => { + const msgs = [ + { role: "user", content: "hi" }, + { role: "assistant", content: "hello" }, + { role: "user", content: "next" }, + ]; + const prefix = conversationPrefixBeforeLastUser(msgs); + assert.equal(prefix.length, 2); + assert.equal(prefix[0].content, "hi"); + assert.equal(prefix[1].role, "assistant"); + }); + + it("stores threadId after turn 1 and reuses it on turn 2 (same space)", async () => { + __resetNotionThreadSessionsForTests(); + const spaceId = "space-1"; + const turn1 = [{ role: "user", content: "first question" }]; + assert.equal(notionThreadSessionLookup(spaceId, turn1), null); + + const threadId = "11111111-2222-3333-4444-555555555555"; + notionThreadSessionStore(spaceId, turn1, "assistant reply one", threadId); + + const turn2 = [ + { role: "user", content: "first question" }, + { role: "assistant", content: "assistant reply one" }, + { role: "user", content: "follow up" }, + ]; + assert.equal(notionThreadSessionLookup(spaceId, turn2), threadId); + // Different space must not share the thread + assert.equal(notionThreadSessionLookup("other-space", turn2), null); + }); + + it("reuses thread when turn-1 user was UREW-rewritten but client replays original text", () => { + __resetNotionThreadSessionsForTests(); + const spaceId = "space-urew"; + const threadId = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"; + // What OmniRoute saw after VibeProxy agentic/UREW rewrite on turn 1 + const turn1Rewritten = [ + { + role: "user", + content: + "Hi! I'm using my local workflow automation tool…\nMy current task: first question", + }, + ]; + notionThreadSessionStore(spaceId, turn1Rewritten, "assistant reply one", threadId); + + // SkillsManager / OpenAI client history keeps the original user wording + const turn2Client = [ + { role: "user", content: "first question" }, + { role: "assistant", content: "assistant reply one" }, + { role: "user", content: "follow up" }, + ]; + assert.equal(notionThreadSessionLookup(spaceId, turn2Client), threadId); + }); + + it("sticky root survives a failed first request (no second createThread)", async () => { + __resetNotionThreadSessionsForTests(); + const { + resolveNotionThreadBinding, + notionThreadMarkCreateAttempted, + NotionWebExecutor, + } = mod as typeof mod & { + resolveNotionThreadBinding: ( + spaceKey: string, + messages: { role: string; content: string }[], + clientThreadId?: string + ) => { threadId: string; createThread: boolean; rootKey: string | null }; + notionThreadMarkCreateAttempted: (rootKey: string | null, threadId: string) => void; + }; + + const spaceId = "space-fail-sticky"; + const turn1 = [{ role: "user", content: "will fail once" }]; + const b1 = resolveNotionThreadBinding(spaceId, turn1); + assert.equal(b1.createThread, true); + notionThreadMarkCreateAttempted(b1.rootKey, b1.threadId); + + // Simulated error: binding for the same conversation must NOT mint a new thread + const b2 = resolveNotionThreadBinding(spaceId, turn1); + assert.equal(b2.threadId, b1.threadId); + assert.equal(b2.createThread, false); + + // Live execute: first upstream error (in-band temporarily-unavailable), second ok + const executor = new NotionWebExecutor(); + const captured: Array<{ createThread?: boolean; threadId?: string }> = []; + let n = 0; + const originalFetch = globalThis.fetch; + try { + globalThis.fetch = (async (_url: string | URL, opts: RequestInit) => { + const body = JSON.parse(String(opts.body)) as { + createThread?: boolean; + threadId?: string; + }; + captured.push(body); + n++; + if (n === 1) { + return new Response( + JSON.stringify({ + id: "e1", + type: "error", + message: "Something went wrong. Please try again later.", + subType: "temporarily-unavailable", + isRetryable: false, + }), + { status: 200 } + ); + } + const ndjson = [ + JSON.stringify({ type: "patch-start", data: { s: [] } }), + JSON.stringify({ + type: "record-map", + recordMap: { + thread_message: { + m1: { + value: { + value: { + step: { + type: "agent-inference", + value: [{ type: "text", content: "recovered" }], + }, + }, + }, + }, + }, + }, + }), + ].join("\n"); + return new Response(ndjson, { status: 200 }); + }) as typeof fetch; + + const result = await executor.execute({ + model: "fable-5", + body: { messages: turn1 }, + stream: false, + credentials: { apiKey: "token_v2=test; space_id=space-fail-sticky" }, + signal: null, + } as never); + assert.equal(result.response.status, 200); + // Retry must keep the same threadId and flip createThread off + assert.ok(captured.length >= 2); + assert.equal(captured[0]!.threadId, captured[1]!.threadId); + assert.equal(captured[1]!.createThread, false); + const json = (await result.response.json()) as { choices?: { message?: { content?: string } }[] }; + assert.match(String(json.choices?.[0]?.message?.content || ""), /recovered/); + } finally { + globalThis.fetch = originalFetch; + __resetNotionThreadSessionsForTests(); + } + }); + + it("hash is stable for the same conversation prefix", () => { + const a = hashNotionConversation("s", [ + { role: "user", content: "x" }, + { role: "assistant", content: "y" }, + ]); + const b = hashNotionConversation("s", [ + { role: "user", content: "x" }, + { role: "assistant", content: "y" }, + ]); + assert.equal(a, b); + assert.notEqual( + a, + hashNotionConversation("s", [ + { role: "user", content: "x" }, + { role: "assistant", content: "z" }, + ]) + ); + }); + + it("execute: first request createThread=true; second multi-turn reuses threadId + createThread=false", async () => { + __resetNotionThreadSessionsForTests(); + const executor = new mod.NotionWebExecutor(); + const captured: Array<{ createThread?: boolean; threadId?: string }> = []; + const originalFetch = globalThis.fetch; + try { + globalThis.fetch = (async (_url: string | URL, opts: RequestInit) => { + captured.push(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: { + step: { + type: "agent-inference", + value: [{ type: "text", content: "ok" }], + }, + }, + }, + }, + }, + }, + }), + ].join("\n"); + return new Response(ndjson, { status: 200 }); + }) as typeof fetch; + + const r1 = await executor.execute({ + model: "fable-5", + body: { messages: [{ role: "user", content: "hello continuity" }] }, + stream: false, + credentials: { apiKey: COOKIE_WITH_SPACE }, + signal: null, + } as never); + assert.equal(r1.response.status, 200); + assert.equal(captured[0].createThread, true); + const t1 = captured[0].threadId; + assert.ok(t1 && t1.length > 10); + + const json1 = (await r1.response.json()) as { notion_thread_id?: string; id?: string }; + assert.equal(json1.notion_thread_id, t1); + + const r2 = await executor.execute({ + model: "fable-5", + body: { + messages: [ + { role: "user", content: "hello continuity" }, + { role: "assistant", content: "ok" }, + { role: "user", content: "second turn" }, + ], + }, + stream: false, + credentials: { apiKey: COOKIE_WITH_SPACE }, + signal: null, + } as never); + assert.equal(r2.response.status, 200); + assert.equal(captured[1].createThread, false); + assert.equal(captured[1].threadId, t1); + } finally { + globalThis.fetch = originalFetch; + __resetNotionThreadSessionsForTests(); + } + }); + + it("execute: honors X-Notion-Thread-Id via ExecuteInput.clientHeaders (not input.headers)", async () => { + __resetNotionThreadSessionsForTests(); + const executor = new mod.NotionWebExecutor(); + const pinned = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"; + let capturedCreateThread: boolean | undefined; + let capturedThreadId: string | undefined; + const originalFetch = globalThis.fetch; + try { + globalThis.fetch = (async (_url: string | URL, opts: RequestInit) => { + const body = JSON.parse(String(opts.body)) as { + createThread?: boolean; + threadId?: string; + }; + capturedCreateThread = body.createThread; + capturedThreadId = body.threadId; + const ndjson = [ + JSON.stringify({ type: "patch-start", data: { s: [] } }), + JSON.stringify({ + type: "record-map", + recordMap: { + thread_message: { + m1: { + value: { + value: { + step: { + type: "agent-inference", + value: [{ type: "text", content: "ok" }], + }, + }, + }, + }, + }, + }, + }), + ].join("\n"); + return new Response(ndjson, { status: 200 }); + }) as typeof fetch; + + // Real ExecuteInput shape: clientHeaders only (headers is undefined). + const result = await executor.execute({ + model: "fable-5", + body: { messages: [{ role: "user", content: "resume thread" }] }, + stream: false, + credentials: { apiKey: COOKIE_WITH_SPACE }, + signal: null, + clientHeaders: { "X-Notion-Thread-Id": pinned }, + } as never); + + assert.equal(result.response.status, 200); + assert.equal(capturedThreadId, pinned); + // Client-supplied thread id must force follow-up mode (createThread=false). + assert.equal(capturedCreateThread, false); + } finally { + globalThis.fetch = originalFetch; + __resetNotionThreadSessionsForTests(); + } + }); +}); diff --git a/tests/unit/executor-notion-web.test.ts b/tests/unit/executor-notion-web.test.ts index 4c2b26e35a..ced4f186fa 100644 --- a/tests/unit/executor-notion-web.test.ts +++ b/tests/unit/executor-notion-web.test.ts @@ -510,6 +510,29 @@ describe("buildNotionTranscript", () => { }); assert.equal((transcript[0].value as { model?: string }).model, "acai-budino-high"); }); + + it("accepts OpenAI content-parts arrays for system + user", () => { + const transcript = buildNotionTranscript( + [ + { + role: "system", + content: [{ type: "text", text: "be helpful" }] as unknown as string, + }, + { + role: "user", + content: [{ type: "text", text: "hi parts" }] as unknown as string, + }, + ], + { spaceId: "s1" } + ); + assert.deepEqual( + transcript.map((t) => t.type), + ["config", "context", "user"] + ); + const ctx = transcript[1].value as { instructions?: string }; + assert.match(String(ctx.instructions), /be helpful/); + assert.deepEqual(transcript[2].value, [["hi parts"]]); + }); }); describe("estimateNotionUsage", () => { @@ -529,6 +552,97 @@ describe("estimateNotionUsage", () => { }); }); +describe("Notion upstream error extraction", () => { + const { extractNotionUpstreamError } = mod as typeof mod & { + extractNotionUpstreamError: (raw: string) => { + message: string; + subType?: string; + isRetryable: boolean; + } | null; + }; + + it("parses temporarily-unavailable NDJSON/JSON errors", () => { + const err = extractNotionUpstreamError( + JSON.stringify({ + id: "e141a6fd-79fa-4bec-9a19-ac41e9728ee6", + type: "error", + message: "Something went wrong. Please try again later.", + subType: "temporarily-unavailable", + isRetryable: false, + }) + ); + assert.ok(err); + assert.match(err!.message, /went wrong/i); + assert.equal(err!.subType, "temporarily-unavailable"); + assert.equal(err!.isRetryable, true); // subtype forces retryable + }); +}); + +describe("Notion custom agent + workflow id", () => { + const { + normalizeNotionWorkflowId, + resolveNotionAgentOptions, + buildNotionTranscript, + __resetNotionThreadSessionsForTests, + } = mod; + + it("normalizes agent URL and dashless hex to UUID", () => { + assert.equal( + normalizeNotionWorkflowId( + "https://app.notion.com/agent/3a3fa5616e71804098510092923e14f9?wfv=chat" + ), + "3a3fa561-6e71-8040-9851-0092923e14f9" + ); + assert.equal( + normalizeNotionWorkflowId("3a3fa561-6e71-8040-9851-0092923e14f9"), + "3a3fa561-6e71-8040-9851-0092923e14f9" + ); + }); + + it("reads workflow_id from cookie string", () => { + const cookie = + "token_v2=abc; space_id=space-1; workflow_id=3a3fa561-6e71-8040-9851-0092923e14f9"; + const agent = resolveNotionAgentOptions({ apiKey: cookie }, cookie); + assert.equal(agent.workflowId, "3a3fa561-6e71-8040-9851-0092923e14f9"); + }); + + it("buildNotionTranscript sets custom agent flags when workflowId present", () => { + const transcript = buildNotionTranscript([{ role: "user", content: "hi" }], { + spaceId: "space-1", + userId: "user-1", + agent: { workflowId: "3a3fa561-6e71-8040-9851-0092923e14f9" }, + }); + const config = transcript.find((t) => t.type === "config") as { + value: Record; + }; + const context = transcript.find((t) => t.type === "context") as { + value: Record; + }; + assert.equal(config.value.isCustomAgent, true); + assert.equal(config.value.useCustomAgentDraft, true); + assert.equal(config.value.workflowId, "3a3fa561-6e71-8040-9851-0092923e14f9"); + assert.equal(context.value.surface, "custom_agent"); + assert.equal(context.value.workflowId, "3a3fa561-6e71-8040-9851-0092923e14f9"); + }); + + it("default AI transcript is not a custom agent", () => { + __resetNotionThreadSessionsForTests(); + const transcript = buildNotionTranscript([{ role: "user", content: "hi" }], { + spaceId: "space-1", + notionModel: "acai-budino-high", + }); + const config = transcript.find((t) => t.type === "config") as { + value: Record; + }; + const context = transcript.find((t) => t.type === "context") as { + value: Record; + }; + assert.equal(config.value.isCustomAgent, false); + assert.equal(context.value.surface, "ai_module"); + assert.equal(config.value.model, "acai-budino-high"); + }); +}); + describe("resolveNotionWebCookie", () => { const { resolveNotionWebCookie, normalizeNotionCookieInput } = mod;