mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-31 02:02:22 +01:00
fix(compare): reject incomplete and errored SSE responses
This commit is contained in:
red person
@@ -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();
|
||||
}
|
||||
}
|
||||
+12
-22
@@ -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 = '<span style="color:var(--color-error);">Error: ' + escapeHtml(error.message) + '</span>';
|
||||
awaitingChoice = false;
|
||||
const message = state._blindMode ? 'Model request failed. Try again.' : error.message;
|
||||
aiBody.innerHTML = '<span style="color:var(--color-error);">Error: ' + escapeHtml(message) + '</span>';
|
||||
}
|
||||
} finally {
|
||||
if (aiMsgEl._spinner) {
|
||||
aiMsgEl._spinner.destroy();
|
||||
aiMsgEl._spinner = null;
|
||||
}
|
||||
clearTimeout(timeoutId);
|
||||
_timerDone = true;
|
||||
cancelAnimationFrame(_rafId);
|
||||
|
||||
+2
-1
@@ -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-v659-notes-reference';
|
||||
const CACHE_NAME = 'odysseus-v660-compare-stream-events';
|
||||
|
||||
// KaTeX resolves these from its own stylesheet, so caching the CSS without them
|
||||
// gives offline math fallback glyphs instead of proper typesetting.
|
||||
@@ -90,6 +90,7 @@ const PRECACHE = [
|
||||
'/static/js/search-chat.js',
|
||||
'/static/js/compare/index.js?v=20260909mobilepaneaddscroll1',
|
||||
'/static/js/compare/vote.js?v=20260828resendcaldrag1',
|
||||
'/static/js/compare/sse.js',
|
||||
'/static/js/colorPicker.js?v=20260910eyedropper1',
|
||||
'/static/js/panels.js?v=20260909movepicklayer1',
|
||||
'/static/js/theme.js?v=20260911organsrain1',
|
||||
|
||||
@@ -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 <img src=x onerror=alert(1)>"}}\n\ndata: [DONE]\n\n', False, False, "Provider Z <img"),
|
||||
("blind-error", 'event:error\ndata:{"message":"Provider Z private-model failed"}\n\n', True, False, "request failed"),
|
||||
("empty-error", 'event:error\ndata:\n\n', False, False, "request failed"),
|
||||
("done-open", 'data: {"delta":"Hello"}\n\ndata: [DONE]\n\n', False, True, "Hello"),
|
||||
("heartbeat", 'data: {"delta":"Hello"}\n\ndata: [DONE]\n\n', False, True, "Hello"),
|
||||
("cr", 'data: {"delta":"Hello"}\r\rdata:[DONE]\r\r', False, True, "Hello"),
|
||||
("invalid", 'data: not-json\n\ndata: [DONE]\n\n', False, False, "Invalid model stream data"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kind,wire,blind,success,text", CASES)
|
||||
def test_compare_stream_contract(kind, wire, blind, success, text):
|
||||
if not shutil.which("node"):
|
||||
pytest.skip("node not installed")
|
||||
script = r"""
|
||||
const assert=require('node:assert/strict');
|
||||
let chromium;try{({chromium}=require('playwright'));}catch(e){if(e.code==='MODULE_NOT_FOUND')process.exit(77);throw e;}
|
||||
(async()=>{
|
||||
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('<div class="compare-pane" data-pane="0"><div id="cmp-badge-0"></div><div id="cmp-history-0"><div class="msg msg-ai"><div class="body"></div></div></div></div>');
|
||||
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(index<heartbeats.length){await new Promise(r=>setTimeout(r,200));controller.enqueue(new TextEncoder().encode(heartbeats[index++]));}else controller.close();}else if(index<bytes.length)controller.enqueue(bytes.slice(index,index+=1));else if(data.kind!=='done-open')controller.close();},cancel(){cancelled=true;}}),{headers:{'Content-Type':'text/event-stream'}});
|
||||
};
|
||||
const message=document.querySelector('.msg-ai'),spinner=document.createElement('span');let spinnerDestroyed=false;spinner.textContent='Loading';message.querySelector('.body').appendChild(spinner);message._spinner={element:spinner,destroy(){spinnerDestroyed=true;spinner.remove();}};
|
||||
await Promise.race([streamToPane(0,'session','Prompt',message,{skipBadge:true}),new Promise((_,reject)=>setTimeout(()=>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
|
||||
Reference in New Issue
Block a user