// Chat API — SSE streaming client for /skill/ask/progress + REST client for // /skill/conversations*, with a mock-mode fallback for offline dev. // Exposes: // askStream({query, conversationId, signal}, handlers) — POST SSE // listConversations() — GET /skill/conversations // getConversation(id) — GET /skill/conversations/{id} // deleteConversation(id) — DEL /skill/conversations/{id} // fetchFile(filenameOrUrl) — GET /files/{filename} // REST errors surface as Error with `.status` set to the HTTP code. // Mock SSE step envelopes — mirror the real /skill/ask/progress protocol: // compatibility `id`/`label` plus a full `step` object matching final // AskResponse.steps[] schema (api_contract_ask.md "Step"). const CHAT_MOCK_STEPS = [ { id: 'routing', label: 'Routing', step: { id: 'routing', title: 'Routing', status: 'done', message: 'Searching concept "GMV"', data: {}, thought: 'I need to confirm which metric "GMV" maps to before locating tables.', }, }, { id: 'routing', label: 'Routing', step: { id: 'routing', title: 'Routing', status: 'done', message: 'Loading "fact_gmv_region"\'s columns schema', data: {}, thought: 'I need to confirm which metric "GMV" maps to before locating tables.', }, }, { id: 'executing', label: 'Executing', step: { id: 'executing', title: 'Executing', status: 'done', message: 'Executing SQL', data: { sql: "select region, sum(gmv) from fact_gmv_region where quarter='2026Q1' group by region" }, thought: 'Aggregate GMV by region for the requested quarter.', }, }, { id: 'rendering', label: 'Rendering', step: { id: 'rendering', title: 'Rendering', status: 'done', message: 'Rendering metric', data: {}, thought: 'Surface the headline number as a metric card.', }, }, { id: 'composing', label: 'Composing', step: { id: 'composing', title: 'Composing', status: 'done', message: null, data: {}, thought: null, }, }, ]; const CHAT_MOCK_FINALS = [ { conversation_id: 'conv_mock_001', status: 'success', message: 'East region GMV last quarter was ¥28.5M, up 12.4% YoY. March contributed the most at 41%.', steps: [ { id: 'intent.route', title: 'Route intent', status: 'done', message: 'Routed to structured query', data: {} }, { id: 'schema.locate', title: 'Locate schema', status: 'done', message: 'Matched fact_gmv_region + dim_region', data: { tables: ['fact_gmv_region', 'dim_region'] } }, { id: 'sql.compose', title: 'Compose SQL', status: 'done', message: "select region, sum(gmv) from fact_gmv_region where quarter='2026Q1' group by region", data: {} }, { id: 'sql.execute', title: 'Execute SQL', status: 'done', message: 'Returned 1 row in 82ms', data: { row_count: 1, latency_ms: 82 } }, { id: 'answer.compose', title: 'Compose answer', status: 'done', message: 'Generated natural-language summary', data: {} }, ], result: { payloads: [ { type: 'metric', format: 'json', filename: 'metric_gmv_east.json', url: '/files/metric_gmv_east.json' }, { type: 'table', format: 'csv', filename: 'table_gmv_by_region.csv', url: '/files/table_gmv_by_region.csv' }, { type: 'chart', format: 'png', filename: 'chart_gmv_trend.png', url: '/files/chart_gmv_trend.png' }, ], sources: [ { type: 'concept', id: 'concept_gmv', name: 'GMV (e-commerce definition)' }, { type: 'concept', id: 'concept_region_e', name: 'East Region' }, { type: 'database', id: 'tbl_fact_gmv', name: 'fact_gmv_region' }, { type: 'database', id: 'tbl_dim_region', name: 'dim_region' }, { type: 'file', id: 'doc_glossary_v3', name: 'Business Glossary v3.md' }, ], confidence: 0.91, }, graph_delta: { version: 1, changed: false, generated_at: null, nodes_upserted: [], edges_upserted: [] }, }, { conversation_id: 'conv_mock_001', status: 'success', message: 'South region GMV last quarter was ¥19.2M, up 8.1% YoY. Growth is slower than East.', steps: [ { id: 'intent.route', title: 'Route intent', status: 'done', message: 'Follow-up recognized', data: {} }, { id: 'schema.locate', title: 'Locate schema', status: 'done', message: 'Reused fact_gmv_region', data: { tables: ['fact_gmv_region'] } }, { id: 'sql.execute', title: 'Execute SQL', status: 'done', message: 'Returned 1 row in 71ms', data: { row_count: 1, latency_ms: 71 } }, { id: 'answer.compose', title: 'Compose answer', status: 'done', message: 'Composed comparative answer', data: {} }, ], result: { payloads: [ { type: 'metric', format: 'json', filename: 'metric_gmv_south.json', url: '/files/metric_gmv_south.json' }, { type: 'table', format: 'csv', filename: 'table_gmv_by_region.csv', url: '/files/table_gmv_by_region.csv' }, ], sources: [ { type: 'concept', id: 'concept_gmv', name: 'GMV (e-commerce definition)' }, { type: 'concept', id: 'concept_region_s', name: 'South Region' }, { type: 'database', id: 'tbl_fact_gmv', name: 'fact_gmv_region' }, ], confidence: 0.89, }, graph_delta: { version: 1, changed: false, generated_at: null, nodes_upserted: [], edges_upserted: [] }, }, { conversation_id: 'conv_mock_001', status: 'clarify', message: '"Active users" is defined differently for the product metric and the revenue metric — which definition do you mean?', steps: [ { id: 'intent.route', title: 'Route intent', status: 'done', message: 'Detected ambiguity on metric definition', data: {} }, ], result: { payloads: [], sources: [], confidence: null }, graph_delta: { version: 1, changed: false, generated_at: null, nodes_upserted: [], edges_upserted: [] }, }, { conversation_id: 'conv_mock_001', status: 'no_data', message: 'Order detail data for East region in April 2026 is not yet available in the system.', steps: [ { id: 'intent.route', title: 'Route intent', status: 'done', message: 'Routed to structured query', data: {} }, { id: 'sql.execute', title: 'Execute SQL', status: 'done', message: 'Query returned 0 rows', data: { row_count: 0 } }, ], result: { payloads: [], sources: [], confidence: null }, graph_delta: { version: 1, changed: false, generated_at: null, nodes_upserted: [], edges_upserted: [] }, }, { conversation_id: 'conv_mock_001', status: 'failure', message: 'Query failed: database connection timed out. Please retry shortly.', steps: [ { id: 'intent.route', title: 'Route intent', status: 'done', message: 'Routed to structured query', data: {} }, { id: 'sql.execute', title: 'Execute SQL', status: 'failed', message: 'Database connection timeout after 30s', data: { latency_ms: 30000 } }, ], result: { payloads: [], sources: [], confidence: null }, graph_delta: { version: 1, changed: false, generated_at: null, nodes_upserted: [], edges_upserted: [] }, }, ]; // Mock file contents keyed by filename. Viewer reads by format. const CHAT_MOCK_FILES = { 'metric_gmv_east.json': { kind: 'json', body: { label: 'East Region GMV', value: '¥28.5M', delta: '+12.4% YoY', period: '2026 Q1', source: 'fact_gmv_region' } }, 'metric_gmv_south.json': { kind: 'json', body: { label: 'South Region GMV', value: '¥19.2M', delta: '+8.1% YoY', period: '2026 Q1', source: 'fact_gmv_region' } }, 'table_gmv_by_region.csv': { kind: 'text', body: 'region,gmv_cny,yoy,top_month\nEast,28500000,+12.4%,2026-03\nSouth,19200000,+8.1%,2026-03\nNorth,15800000,-2.3%,2026-01\nSouthwest,9200000,+4.7%,2026-02\n' }, 'chart_gmv_trend.png': { kind: 'svgDataUrl', body: chatMockTrendSvg() }, }; function chatMockTrendSvg() { const pts = [28, 34, 41, 38, 47, 52, 55, 61, 58, 66, 71, 84]; const max = Math.max(...pts); const svg = ` ${pts.map((v, i) => { const x = 24 + i * 26; const h = (v / max) * 130; return ``; }).join('')} GMV by month · 2025–2026 Jan Jul Dec `; return 'data:image/svg+xml;utf8,' + encodeURIComponent(svg); } // Recover a just-completed turn's answer by polling the conversation. The // backend finishes the agent run server-side and persists the turn even if the // SSE stream drops (e.g. a Cloudflare QUIC error on the long-lived connection), // so on a mid-stream network failure we can still surface the real answer // instead of a "failed" turn. We match the turn by its query text (scanning // from the newest) so this works for both new and follow-up conversations // without needing the prior turn count. Returns the recovered final-shaped // response (with conversation_id), or null if it never appears in time. async function _recoverFinalFromConversation(convId, query, { attempts = 12, delayMs = 2500 } = {}) { for (let i = 0; i < attempts; i++) { await new Promise(r => setTimeout(r, delayMs)); let conv; try { conv = await getConversation(convId); } catch (e) { continue; } // 404 / transient — the turn may not be saved yet const turns = (conv && conv.turns) || []; for (let j = turns.length - 1; j >= 0; j--) { const t = turns[j]; if (t && t.query === query && t.response) { return { ...t.response, conversation_id: convId }; } } } return null; } // Real SSE parser via fetch + ReadableStream. EventSource is not used because it can't POST. async function askStreamReal({ query, conversationId, signal }, handlers) { const base = window.PARCLE_API_BASE || '/api'; const authHeaders = typeof parcleAuthHeaders === 'function' ? parcleAuthHeaders() : {}; const res = await fetch(base + '/skill/ask/progress', { method: 'POST', headers: { 'Content-Type': 'application/json', 'Accept': 'text/event-stream', ...authHeaders }, body: JSON.stringify({ query, conversation_id: conversationId || null }), signal, }); if (res.status === 401 && typeof parcleNotifyUnauthorized === 'function') parcleNotifyUnauthorized(); if (!res.ok) throw new Error(`HTTP ${res.status}`); const reader = res.body.getReader(); const dec = new TextDecoder(); let buf = ''; let finalReceived = false; // The backend emits a terminal `error` frame (then closes the stream) when the // run fails server-side. That is an authoritative failure — distinct from a // network/QUIC drop — so when we've seen one we must NOT poll for recovery; // surface it immediately instead of stalling on _recoverFinalFromConversation. let lastError = null; // Best-known conversation id: the request arg, upgraded by the early `meta` // event (covers brand-new conversations) and finally by the `final` event. let streamConvId = conversationId || null; // On a mid-stream drop (no `final` yet, not a user abort), try to recover the // answer the backend persisted out-of-band. Returns true if recovery // delivered a final via onFinal. const tryRecover = async () => { if (finalReceived || !streamConvId) return false; const recovered = await _recoverFinalFromConversation(streamConvId, query); if (recovered) { finalReceived = true; handlers.onFinal?.(recovered); return true; } return false; }; try { for (;;) { const { done, value } = await reader.read(); if (done) break; buf += dec.decode(value, { stream: true }); let idx; while ((idx = buf.indexOf('\n\n')) >= 0) { const frame = buf.slice(0, idx); buf = buf.slice(idx + 2); let event = 'message'; const dataLines = []; for (const raw of frame.split('\n')) { if (raw.startsWith('event:')) event = raw.slice(6).trim(); else if (raw.startsWith('data:')) dataLines.push(raw.slice(5).replace(/^\s/, '')); } if (dataLines.length === 0) continue; // `: keepalive` comment frames land here let parsed; try { parsed = JSON.parse(dataLines.join('\n')); } catch (e) { console.warn('SSE parse error', e, dataLines); continue; } if (event === 'meta') { if (parsed && parsed.conversation_id) streamConvId = parsed.conversation_id; handlers.onMeta?.(parsed); } else if (event === 'step') handlers.onStep?.(parsed); else if (event === 'knowledge') handlers.onKnowledge?.(parsed); else if (event === 'final') { finalReceived = true; streamConvId = (parsed && parsed.conversation_id) || streamConvId; handlers.onFinal?.(parsed); } else if (event === 'error') { lastError = parsed; handlers.onError?.(parsed); } } } } catch (err) { // User-initiated abort — propagate so the UI shows "Stopped". Otherwise the // connection dropped (QUIC/network): the server-side run continues, so try // to recover the persisted answer before surfacing a failure. if ((signal && signal.aborted) || (err && err.name === 'AbortError')) throw err; if (finalReceived) return; if (lastError) throw new Error(_errorDetail(lastError)); if (await tryRecover()) return; throw err; } if (!finalReceived) { // Backend already reported a terminal failure — surface it now rather than // polling ~30s for a recovery that can only return that same failure. if (lastError) throw new Error(_errorDetail(lastError)); // Stream ended cleanly but no `final` (e.g. proxy truncation) — attempt the // same out-of-band recovery before giving up. if (await tryRecover()) return; throw new Error('stream ended before final event'); } } function _errorDetail(payload) { const d = payload && payload.detail; if (typeof d === 'string') return d; if (d != null) return JSON.stringify(d); return 'request failed'; } // Mock stream — cycles through canned finals based on a module-local counter. // Finals are also appended to an in-memory mock conversation store so the // new conversation-listing / delete endpoints work without a real backend. let chatMockTurnIdx = 0; let _mockConvStore = []; // [{conversation_id, turns: [{query, response}]}] let _mockConvCounter = 0; function resetChatMockIdx() { chatMockTurnIdx = 0; } function _mockConvAppend({ conversationId, query, response }) { let conv = _mockConvStore.find(c => c.conversation_id === conversationId); if (!conv) { conv = { conversation_id: conversationId, turns: [] }; _mockConvStore.unshift(conv); } conv.turns.push({ query, response }); } function askStreamMock({ query, conversationId, signal }, handlers) { if (!conversationId) chatMockTurnIdx = 0; return new Promise((resolve, reject) => { const timers = []; const schedule = (ms, fn) => { const t = setTimeout(fn, ms); timers.push(t); }; if (signal) { signal.addEventListener('abort', () => { timers.forEach(clearTimeout); reject(new DOMException('aborted', 'AbortError')); }); } CHAT_MOCK_STEPS.forEach((s, i) => { schedule(300 + i * 420, () => handlers.onStep?.(s)); }); // Knowledge events (api_contract_ask.md "knowledge"): emit the to-be-picked // final's sources progressively so live mode demonstrates real-time // accumulation. chatMockTurnIdx isn't mutated until the finalDelay callback, // so the final picked there is the same one we peek here — live sources end // up matching final.result.sources exactly. Non-success finals carry no // sources, so nothing is emitted for them (matches the empty-result rule). const pickedFinal = CHAT_MOCK_FINALS[chatMockTurnIdx % CHAT_MOCK_FINALS.length]; const mockSources = (pickedFinal && pickedFinal.result && pickedFinal.result.sources) || []; mockSources.forEach((src, i) => { schedule(520 + i * 360, () => handlers.onKnowledge?.(src)); }); const finalDelay = 300 + CHAT_MOCK_STEPS.length * 420 + 300; schedule(finalDelay, () => { const picked = CHAT_MOCK_FINALS[chatMockTurnIdx % CHAT_MOCK_FINALS.length]; chatMockTurnIdx += 1; const final = JSON.parse(JSON.stringify(picked)); if (conversationId) { final.conversation_id = conversationId; } else { _mockConvCounter += 1; final.conversation_id = `conv_mock_${String(_mockConvCounter).padStart(3, '0')}`; } // Synthesize a per-turn graph_delta on success so mock mode visibly // demonstrates the per-turn learning live update. Each turn registers // a uniquely-keyed concept timestamped at "now" — the learning stream // surfaces it in the "Last 5 minutes" bucket on the next render. if (final.status === 'success') { const ts = new Date().toISOString().slice(0, 16); const turnTag = `${final.conversation_id}_t${chatMockTurnIdx}`; const conceptId = `concept_mock_${turnTag}`; const conceptName = (query || 'mock learning') .replace(/[\n\r]+/g, ' ').trim().slice(0, 48) || 'mock learning'; final.graph_delta = { version: 1, changed: true, generated_at: ts, nodes_upserted: [{ id: conceptId, kind: 'concept', name: conceptName, synonyms: [], properties: { description: `Concept inferred from "${conceptName}".` }, origin: 'conversation_learning', created_at: ts, status: 'active', merged_into: null, merged_at: null, merged_from: [], dedup_events: [], }], edges_upserted: [], }; } _mockConvAppend({ conversationId: final.conversation_id, query, response: final }); handlers.onFinal?.(final); resolve(); }); }); } function chatMockEnabled() { if (typeof window.PARCLE_CHAT_MOCK === 'boolean') return window.PARCLE_CHAT_MOCK; return !window.PARCLE_API_BASE; // default: mock on when no API base configured } async function askStream(req, handlers) { if (chatMockEnabled()) return askStreamMock(req, handlers); return askStreamReal(req, handlers); } // ────────────────────────────────────────────────────────────────────────── // Conversation persistence — GET/DELETE against /skill/conversations*. // The backend is the source of truth for every turn's query + response; // the client only caches the currently-selected conversation id. // ────────────────────────────────────────────────────────────────────────── async function _parseDetailError(res) { if (res.status === 401 && typeof parcleNotifyUnauthorized === 'function') parcleNotifyUnauthorized(); let msg = `HTTP ${res.status}`; try { const j = await res.json(); if (j && j.detail) msg = typeof j.detail === 'string' ? j.detail : JSON.stringify(j.detail); } catch (e) { /* non-JSON body — keep generic message */ } const err = new Error(msg); err.status = res.status; return err; } async function listConversations() { if (chatMockEnabled()) { return JSON.parse(JSON.stringify(_mockConvStore)); } const base = window.PARCLE_API_BASE || '/api'; const res = await fetch(base + '/skill/conversations', { headers: parcleAuthHeaders() }); if (!res.ok) throw await _parseDetailError(res); const data = await res.json(); return Array.isArray(data && data.conversations) ? data.conversations : []; } // Application-lifetime cache. Primed once at App mount so ChatPage opens // on already-resolved data instead of waiting for a fresh fetch. Local // mutations (send / delete) keep their own state in ChatPage; this cache // is only the snapshot of what the server had at first load. let _conversationsCache = null; // { promise } | null function listConversationsCached() { if (_conversationsCache) return _conversationsCache.promise; const promise = listConversations(); _conversationsCache = { promise }; promise.catch(() => { if (_conversationsCache && _conversationsCache.promise === promise) _conversationsCache = null; }); return promise; } function primeConversations() { return listConversationsCached().catch(() => { /* silent — ChatPage surfaces its own error */ }); } function invalidateConversations() { _conversationsCache = null; } async function getConversation(conversationId) { if (chatMockEnabled()) { const c = _mockConvStore.find(c => c.conversation_id === conversationId); if (!c) { const e = new Error('Conversation not found'); e.status = 404; throw e; } return JSON.parse(JSON.stringify(c)); } const base = window.PARCLE_API_BASE || '/api'; const res = await fetch(base + '/skill/conversations/' + encodeURIComponent(conversationId), { headers: parcleAuthHeaders() }); if (!res.ok) throw await _parseDetailError(res); return await res.json(); } async function deleteConversation(conversationId) { if (chatMockEnabled()) { const idx = _mockConvStore.findIndex(c => c.conversation_id === conversationId); if (idx < 0) { const e = new Error('Conversation not found'); e.status = 404; throw e; } const deleted = _mockConvStore[idx].turns.length; _mockConvStore.splice(idx, 1); return { conversation_id: conversationId, deleted_turns: deleted }; } const base = window.PARCLE_API_BASE || '/api'; const res = await fetch(base + '/skill/conversations/' + encodeURIComponent(conversationId), { method: 'DELETE', headers: parcleAuthHeaders() }); if (!res.ok) throw await _parseDetailError(res); return await res.json(); } // fetchFile — resolves a payload URL to renderable content async function fetchFile(filenameOrUrl) { const filename = filenameOrUrl.replace(/^\/files\//, '').replace(/^.*\/files\//, ''); if (chatMockEnabled()) { const m = CHAT_MOCK_FILES[filename]; if (!m) throw new Error(`mock file not found: ${filename}`); return m; } const base = window.PARCLE_API_BASE || '/api'; const url = filenameOrUrl.startsWith('http') ? filenameOrUrl : base + '/files/' + filename; const res = await fetch(url); if (!res.ok) throw new Error(`HTTP ${res.status}`); const ct = res.headers.get('content-type') || ''; if (ct.includes('application/json')) { return { kind: 'json', body: await res.json() }; } if (ct.startsWith('image/')) { const blob = await res.blob(); return { kind: 'blobUrl', body: URL.createObjectURL(blob), contentType: ct }; } return { kind: 'text', body: await res.text(), contentType: ct }; } Object.assign(window, { askStream, fetchFile, listConversations, listConversationsCached, primeConversations, invalidateConversations, getConversation, deleteConversation, chatMockEnabled, resetChatMockIdx, CHAT_MOCK_FINALS, CHAT_MOCK_STEPS, });