diff --git a/static/js/compare/sse.js b/static/js/compare/sse.js new file mode 100644 index 000000000..7e60b0e12 --- /dev/null +++ b/static/js/compare/sse.js @@ -0,0 +1,65 @@ +function streamError(payload) { + const detail = payload?.data ?? payload; + const candidates = [detail?.message, detail?.text, detail?.error?.message, + detail?.error, payload?.message, payload?.text, payload?.error?.message, + payload?.error, typeof payload === 'string' ? payload : null]; + const message = candidates.find(value => typeof value === 'string' && value.trim()); + return new Error(message || 'Model request failed.'); +} + +/** Decode complete SSE frames, requiring the backend's terminal DONE event. */ +export async function consumeCompareEvents(response, onEvent, onActivity) { + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let line = '', event = 'message', data = [], skipLF = false; + try { + while (true) { + const { done, value } = await reader.read(); + if (!done) onActivity?.(); + let chunk = decoder.decode(value, { stream: !done }); + if (skipLF && chunk.length) { + if (chunk.startsWith('\n')) chunk = chunk.slice(1); + skipLF = false; + } + const separators = /\r\n|\r|\n/g; + let start = 0, separator; + while ((separator = separators.exec(chunk))) { + line += chunk.slice(start, separator.index); + start = separator.index + separator[0].length; + skipLF = separator[0] === '\r' && start === chunk.length; + if (!line) { + if (data.length || event === 'error') { + const raw = data.join('\n'); + if (event !== 'error' && raw === '[DONE]') return; + let payload; + try { payload = JSON.parse(raw); } + catch (_) { + if (event === 'error') throw streamError(raw); + throw new Error('Invalid model stream data.'); + } + if (event === 'error' || payload?.type === 'error') throw streamError(payload); + if (!payload || typeof payload !== 'object' || Array.isArray(payload)) { + throw new Error('Invalid model stream data.'); + } + await onEvent(payload); + } + event = 'message'; + data = []; + } else if (!line.startsWith(':')) { + const colon = line.indexOf(':'); + const field = colon < 0 ? line : line.slice(0, colon); + let value = colon < 0 ? '' : line.slice(colon + 1); + if (value.startsWith(' ')) value = value.slice(1); + if (field === 'data') data.push(value); + else if (field === 'event') event = value || 'message'; + } + line = ''; + } + line += chunk.slice(start); + if (done) throw new Error('Stream ended before completion.'); + } + } finally { + try { await reader.cancel(); } catch (_) {} + reader.releaseLock(); + } +} diff --git a/static/js/compare/stream.js b/static/js/compare/stream.js index 4ea5c4e74..c2577cbed 100644 --- a/static/js/compare/stream.js +++ b/static/js/compare/stream.js @@ -1,5 +1,6 @@ // compare/stream.js — SSE streaming to panes import state from './state.js'; +import { consumeCompareEvents } from './sse.js'; import { addFinishBadge } from './vote.js?v=20260828resendcaldrag1'; import { getModelCost, renderAskUserCard, safeDisplayImageSrc } from '../chatRenderer.js?v=20260914metricssummary1'; import markdownModule from '../markdown.js'; @@ -472,25 +473,9 @@ async function streamToPane(paneIdx, sessionId, message, aiMsgEl, opts) { }); if (!response.ok) throw new Error(`HTTP ${response.status}`); - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ''; - - while (true) { - const { done, value } = await reader.read(); - if (done) break; - _resetIdleTimeout(); // any chunk = stream is alive - - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split('\n'); - buffer = lines.pop() || ''; - - for (const line of lines) { - if (!line.startsWith('data: ')) continue; - const data = line.slice(6); - if (data === '[DONE]') break; + await consumeCompareEvents(response, + (json) => { try { - const json = JSON.parse(data); if (json.type === 'metrics') { metrics = json.data; @@ -672,7 +657,7 @@ async function streamToPane(paneIdx, sessionId, message, aiMsgEl, opts) { if (hist) hist.scrollTop = hist.scrollHeight; } else if (json.delta) { // Skip text deltas if we already rendered an image - if (aiMsgEl._imageData) continue; + if (aiMsgEl._imageData) return; // Capture TTFT on very first text delta if (!accumulated && !_ttft) _ttft = performance.now() - _timerStart; // On first delta, destroy spinner and prepare text area @@ -701,8 +686,7 @@ async function streamToPane(paneIdx, sessionId, message, aiMsgEl, opts) { _scheduleLiveRender(target); } } catch (e) { console.warn('Compare stream render error:', e); } - } - } + }, _resetIdleTimeout); streamOk = true; // Destroy any remaining spinner @@ -900,9 +884,15 @@ async function streamToPane(paneIdx, sessionId, message, aiMsgEl, opts) { } } else { console.error('Compare stream error:', error); - aiBody.innerHTML = 'Error: ' + escapeHtml(error.message) + ''; + awaitingChoice = false; + const message = state._blindMode ? 'Model request failed. Try again.' : error.message; + aiBody.innerHTML = 'Error: ' + escapeHtml(message) + ''; } } finally { + if (aiMsgEl._spinner) { + aiMsgEl._spinner.destroy(); + aiMsgEl._spinner = null; + } clearTimeout(timeoutId); _timerDone = true; cancelAnimationFrame(_rafId); diff --git a/static/sw.js b/static/sw.js index de964dd14..a3feba04a 100644 --- a/static/sw.js +++ b/static/sw.js @@ -7,7 +7,7 @@ // - Other static assets (images/fonts/libs): cache-first with bg refresh. // - API / non-GET: never cached. // Bump CACHE_NAME whenever the precache list or SW logic changes. -const CACHE_NAME = 'odysseus-v1114-pr40-editor-precache'; +const CACHE_NAME = 'odysseus-v1115-pr40-editor-precache'; // KaTeX resolves these from its own stylesheet, so caching the CSS without them // gives offline math fallback glyphs instead of proper typesetting. @@ -38,6 +38,7 @@ const KATEX_FONTS = [ // Both are fetched at install time, in the background. Entries must match the // exact URL the browser requests, query string included. const PRECACHE = [ + '/static/js/compare/sse.js', '/', '/static/css/00-tokens.css?v=20260929csssplit', '/static/css/01-agent-chat.css?v=20260929csssplit', diff --git a/tests/test_compare_stream_contract_browser.py b/tests/test_compare_stream_contract_browser.py new file mode 100644 index 000000000..620a1768d --- /dev/null +++ b/tests/test_compare_stream_contract_browser.py @@ -0,0 +1,70 @@ +"""Compare's actual stream consumer must preserve evidence and terminal state.""" +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] + +CASES = [ + ("lf", 'data: {"delta":"Hello"}\n\ndata: [DONE]\n\n', False, True, "Hello"), + ("crlf", 'data:{"delta":"Héllö"}\r\n\r\ndata:[DONE]\r\n\r\n', False, True, "Héllö"), + ("multiline", 'data: {"delta":\ndata: "Hello"}\n\ndata: [DONE]\n\n', False, True, "Hello"), + ("eof", 'data: {"delta":"Partial"}\n\n', False, False, "ended before completion"), + ("named-error", 'event: error\ndata: {"text":"Provider Z private-model failed"}\n\n', False, False, "Provider Z private-model failed"), + ("typed-error", 'data: {"type":"error","data":{"message":"Provider Z "}}\n\ndata: [DONE]\n\n', False, False, "Provider Z { + const browser=await chromium.launch({headless:true}); + try{ + const page=await browser.newPage(); + await page.goto(process.env.ODYSSEUS_TEST_STATIC_ORIGIN+'/static/js/storage.js'); + await page.setContent('
'); + const result=await page.evaluate(async(data)=>{ + const {default:state}=await import('/static/js/compare/state.js'); + const {streamToPane}=await import('/static/js/compare/stream.js'); + Object.assign(state,{isActive:true,_selectedModels:[{model:'fixture',name:'Fixture'}],_paneSessionIds:['session'],_compareMode:'chat',_parallel:true,_blindMode:data.blind,_timeout:data.kind==='heartbeat'?1:5}); + const nativeFetch=window.fetch,bytes=new TextEncoder().encode(data.wire);let index=0,cancelled=false,context=null; + window.fetch=async(url,options)=>{ + if(!String(url).endsWith('/api/chat_stream'))return nativeFetch(url,options); + context=options.body.get('search_context'); + const heartbeats=[...Array(6).fill(': ping\n\n'),data.wire]; + return new Response(new ReadableStream({start(controller){options.signal.addEventListener('abort',()=>controller.error(new DOMException('Stopped','AbortError')),{once:true});},async pull(controller){if(data.kind==='heartbeat'){if(indexsetTimeout(r,200));controller.enqueue(new TextEncoder().encode(heartbeats[index++]));}else controller.close();}else if(indexsetTimeout(()=>reject(Error('Consumer did not finish at DONE')),data.kind==='heartbeat'?6000:1500))]); + const pane=document.querySelector('.compare-pane'); + return {done:pane.classList.contains('is-done'),failed:pane.classList.contains('is-failed'),text:pane.innerText,images:pane.querySelectorAll('img').length,cancelled,spinnerDestroyed,spinnerOwned:!!message._spinner}; + },JSON.parse(process.argv[1])); + const expected=JSON.parse(process.argv[1]); + assert.equal(result.done,expected.success);assert.equal(result.failed,!expected.success);assert(result.text.includes(expected.text),JSON.stringify(result));assert.equal(result.images,0); + assert(result.spinnerDestroyed);assert(!result.spinnerOwned); + if(expected.blind)assert(!/Provider Z|private-model/.test(result.text),JSON.stringify(result)); + if(expected.kind==='done-open')assert(result.cancelled); + console.log(JSON.stringify(result)); + }finally{await browser.close();} +})().catch(e=>{console.error(e);process.exitCode=1;}); +""" + data = dict(kind=kind, wire=wire, blind=blind, success=success, text=text) + result = subprocess.run(["node", "-e", script, json.dumps(data)], cwd=ROOT, + capture_output=True, text=True, timeout=30) + if result.returncode == 77: + pytest.skip("playwright not installed") + assert result.returncode == 0, result.stdout + result.stderr