diff --git a/docs/EXTENSIONS.md b/docs/EXTENSIONS.md index 61d6b848..7d7c8f0b 100644 --- a/docs/EXTENSIONS.md +++ b/docs/EXTENSIONS.md @@ -335,6 +335,11 @@ Core suppresses reconnect duplicates so a recently observed `sessionId`/`streamId` pair produces at most one start event and one terminal event. `sessionId` is the owner of the original stream; server-side compression or session rotation at completion does not change that lifecycle identity. +Terminal callbacks run after Core has cleared the original stream ownership, +applied the current session's immediate terminal transcript projection, and +returned the owned pane to its idle state. They do not wait for unrelated +follow-up work such as notifications, queued turns, or optional recovery +refreshes. This is a page-local, live-stream notification surface. It does not replay history, report turns that finish without this page observing their stream, or diff --git a/static/messages.js b/static/messages.js index 1073169f..14b5f287 100644 --- a/static/messages.js +++ b/static/messages.js @@ -6099,10 +6099,6 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ } if(isSessionViewed) _markSessionViewed(completedSid, completedMessageCount); _clearOwnerInflightState(); - _dispatchExtensionTurnLifecycle('turn:complete',activeSid,streamId,{ - status:d.status||'completed', - endedAt:Date.now()/1000, - }); if(typeof _markSessionCompletedInList==='function'){ _markSessionCompletedInList(completedSession, activeSid); } @@ -6329,6 +6325,10 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if(isActiveSession) _queueDrainSid=activeSid; renderSessionList(); _setActivePaneIdleIfOwner(); + _dispatchExtensionTurnLifecycle('turn:complete',activeSid,streamId,{ + status:d.status||'completed', + endedAt:Date.now()/1000, + }); playNotificationSound(); // #4416: notify if the tab was hidden at ANY point during this stream // (not just at done-receive time, which a throttled background-tab SSE @@ -6514,10 +6514,6 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ let d={}; try{ d=JSON.parse(e.data||'{}')||{}; }catch(_){ d={}; } const _extensionErrorType=(d.type==='cancelled'||d.type==='interrupted')?'turn:cancel':'turn:error'; - _dispatchExtensionTurnLifecycle(_extensionErrorType,activeSid,streamId,{ - status:d.status||d.type||(_extensionErrorType==='turn:cancel'?'cancelled':'error'), - endedAt:Date.now()/1000, - }); const currentSid=S.session&&S.session.session_id; const eventSid=d.old_session_id||d.session_id||''; const continuationSid=(d.session&&d.session.session_id)||d.new_session_id||d.continuation_session_id||''; @@ -6615,6 +6611,10 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ } _setActivePaneIdleIfOwner(); renderSessionList(); // clear streaming indicator immediately on apperror + _dispatchExtensionTurnLifecycle(_extensionErrorType,activeSid,streamId,{ + status:d.status||d.type||(_extensionErrorType==='turn:cancel'?'cancelled':'error'), + endedAt:Date.now()/1000, + }); }); source.addEventListener('warning',e=>{ @@ -6774,10 +6774,6 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ _clearClarifyForOwner('cancelled'); let _cancelData={}; try{ _cancelData=JSON.parse(e.data||'{}')||{}; }catch(_){ _cancelData={}; } - _dispatchExtensionTurnLifecycle('turn:cancel',activeSid,streamId,{ - status:_cancelData.status||_cancelData.type||'cancelled', - endedAt:Date.now()/1000, - }); _flushReasoningToAnchor(); _applyToAnchor('cancel',{ status:_cancelData.status||_cancelData.type||'cancelled', @@ -6821,6 +6817,8 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ // a second GET race where the visible cancelled work briefly collapses to only // the fallback "Task cancelled" marker (#4076). const _cancelSessionPayload=_cancelData&&typeof _cancelData.session==='object'?_cancelData.session:null; + renderSessionList(); + _setActivePaneIdleIfOwner(); (async()=>{ try{ if(_applyCancelSessionPayload(_cancelSessionPayload)) return; @@ -6846,10 +6844,13 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ if(_wasFollowingAtCancelFb && typeof scrollToBottom==='function') scrollToBottom(); _markSessionViewed(activeSid, S.messages.length); } + }finally{ + _dispatchExtensionTurnLifecycle('turn:cancel',activeSid,streamId,{ + status:_cancelData.status||_cancelData.type||'cancelled', + endedAt:Date.now()/1000, + }); } })(); - renderSessionList(); - _setActivePaneIdleIfOwner(); }); for(const _runJournalEventName of ['token','interim_assistant','reasoning','tool','tool_complete','todo_state','approval','clarify','state_saved','title','title_status','context_status','goal','goal_continue','done','stream_end','pending_steer_leftover','compressing','compressed','metering','apperror','warning','error','cancel']){ @@ -7076,11 +7077,11 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ trackBackgroundError(activeSid,_errTitle,'Connection interrupted'); } } + _setActivePaneIdleIfOwner(); _dispatchExtensionTurnLifecycle('turn:error',activeSid,streamId,{ status:'connection_lost', endedAt:Date.now()/1000, }); - _setActivePaneIdleIfOwner(); } (async()=>{ diff --git a/tests/test_extension_turn_lifecycle.py b/tests/test_extension_turn_lifecycle.py index 856b6fe1..4a1f5b71 100644 --- a/tests/test_extension_turn_lifecycle.py +++ b/tests/test_extension_turn_lifecycle.py @@ -2,6 +2,7 @@ import json from pathlib import Path +import re import shutil import subprocess import textwrap @@ -10,8 +11,173 @@ import pytest ROOT = Path(__file__).parent.parent +STATIC = ROOT / "static" EXTENSION_SETTINGS_JS = ROOT / "static" / "extension_settings.js" MESSAGES_JS = (ROOT / "static" / "messages.js").read_text(encoding="utf-8") +INDEX_HTML = (STATIC / "index.html").read_text(encoding="utf-8") +UI_JS = (STATIC / "ui.js").read_text(encoding="utf-8") +SUPPORT_SCRIPTS = [ + (STATIC / name).read_text(encoding="utf-8") + for name in ("i18n.js", "icons.js", "assistant_turn_anchors.js") +] +HARNESS_HTML = re.sub(r"]*>.*?", "", INDEX_HTML, flags=re.I | re.S) +HARNESS_HTML = re.sub(r"]*>", "", HARNESS_HTML, flags=re.I) + + +_MOCK_EVENT_SOURCE = r""" +class MockEventSource { + static CONNECTING = 0; + static OPEN = 1; + static CLOSED = 2; + static instances = []; + constructor(url) { + this.url = String(url); + this.readyState = MockEventSource.OPEN; + this.listeners = new Map(); + MockEventSource.instances.push(this); + } + addEventListener(name, listener) { + const listeners = this.listeners.get(name) || []; + listeners.push(listener); + this.listeners.set(name, listeners); + } + removeEventListener(name, listener) { + this.listeners.set(name, (this.listeners.get(name) || []).filter(fn => fn !== listener)); + } + async emit(name, payload) { + const event = {type: name, data: JSON.stringify(payload), lastEventId: ''}; + await Promise.all((this.listeners.get(name) || []).map(listener => listener(event))); + } + close() { this.readyState = MockEventSource.CLOSED; } +} +window.EventSource = MockEventSource; +window.MockEventSource = MockEventSource; +""" + + +_STABILIZE_UNRELATED_UI = r""" +appendLiveCompressionCard = () => false; +resetTurnWorkspaceMutations = () => {}; +_resetStreamScrollFollow = () => {}; +ensureLiveWorklogShell = () => {}; +_shouldUseLiveProseFade = () => false; +_shouldFollowMessagesOnDomReplace = () => false; +_isMessagePaneNearBottom = () => false; +_markSessionViewed = () => {}; +setBusy = value => { S.busy = Boolean(value); }; +setComposerStatus = () => {}; +setStatus = () => {}; +syncTopbar = () => {}; +renderMessages = () => {}; +renderSessionList = () => {}; +renderSessionArtifacts = () => {}; +loadDir = () => {}; +playNotificationSound = () => {}; +sendBrowserNotification = () => {}; +api = async () => ({}); +""" + + +_LIFECYCLE_SCENARIO = r""" +async ({kind}) => { + const activeSid = 'session-a'; + const streamId = `stream-${kind}`; + const lifecycle = []; + const extension = window.hermesExt.register('lifecycle-probe'); + if (!extension) throw new Error('lifecycle probe did not register'); + for (const type of ['turn:start', 'turn:complete', 'turn:error', 'turn:cancel']) { + extension.events.on(type, event => { + const lastMessage = Array.isArray(S.messages) ? S.messages.at(-1) : null; + lifecycle.push({ + event: {...event}, + activeStreamId: S.activeStreamId || null, + busy: Boolean(S.busy), + currentSid: S.session && S.session.session_id || null, + lastContent: lastMessage && lastMessage.content || null, + }); + }); + } + + S.session = {session_id: activeSid, messages: [{role: 'user', content: 'question'}]}; + S.messages = [{role: 'user', content: 'question'}]; + S.toolCalls = []; + S.activeStreamId = streamId; + S.busy = true; + attachLiveStream(activeSid, streamId, []); + const source = MockEventSource.instances.at(-1); + if (!source) throw new Error('attachLiveStream did not construct EventSource'); + + if (kind === 'done') { + await source.emit('done', { + status: 'completed', + stream_id: streamId, + session: { + session_id: activeSid, + messages: [{role: 'assistant', content: 'settled-done'}], + tool_calls: [], + }, + }); + } else if (kind === 'apperror') { + await source.emit('apperror', { + type: 'provider_error', + status: 'provider_error', + message: 'provider failed', + session_id: activeSid, + session: { + session_id: activeSid, + messages: [{role: 'assistant', content: 'settled-error'}], + }, + }); + } else if (kind === 'apperror-cancel') { + await source.emit('apperror', { + type: 'interrupted', + status: 'interrupted', + message: 'interrupted', + session_id: activeSid, + session: { + session_id: activeSid, + messages: [{role: 'assistant', content: 'settled-interrupted'}], + }, + }); + } else if (kind === 'cancel') { + await source.emit('cancel', { + type: 'cancelled', + status: 'cancelled', + session_id: activeSid, + session: { + session_id: activeSid, + messages: [{role: 'assistant', content: 'settled-cancel'}], + }, + }); + } else if (kind === 'connection-error') { + const nativeSetTimeout = window.setTimeout; + api = async url => String(url).includes('/api/chat/stream/status') + ? {active: false, replay_available: false} + : {}; + window.setTimeout = callback => { + queueMicrotask(callback); + return 1; + }; + try { + await source.emit('error', {}); + await new Promise((resolve, reject) => { + const deadline = Date.now() + 2000; + const poll = () => { + if (lifecycle.some(entry => entry.event.type === 'turn:error')) return resolve(); + if (Date.now() >= deadline) return reject(new Error('connection terminal event not observed')); + nativeSetTimeout(poll, 0); + }; + poll(); + }); + } finally { + window.setTimeout = nativeSetTimeout; + } + } else { + throw new Error(`unknown lifecycle scenario: ${kind}`); + } + return lifecycle; +} +""" def _function_body(source: str, name: str) -> str: @@ -206,6 +372,91 @@ def test_live_stream_bridge_forwards_normalized_lifecycle_details(): _run_node(script) +@pytest.fixture(scope="module") +def lifecycle_browser(): + playwright_api = pytest.importorskip("playwright.sync_api") + with playwright_api.sync_playwright() as playwright: + browser = playwright.chromium.launch( + headless=True, + args=["--no-sandbox", "--disable-dev-shm-usage"], + ) + yield browser + browser.close() + + +def _run_lifecycle_scenario(browser, kind: str) -> list[dict]: + page = browser.new_page() + page.route( + "**/*", + lambda route: route.fulfill( + status=200, + content_type="text/html" + if route.request.url == "http://harness.test/" + else "text/plain", + body=HARNESS_HTML if route.request.url == "http://harness.test/" else "", + ), + ) + page.add_init_script(_MOCK_EVENT_SOURCE) + try: + page.goto("http://harness.test/", wait_until="domcontentloaded") + page.evaluate( + """ + window.__HERMES_EXTENSION_CONFIG__ = { + extensions: [{id: 'lifecycle-probe', storage_owned: false}], + }; + """ + ) + for script in SUPPORT_SCRIPTS: + page.add_script_tag(content=script) + page.add_script_tag(content=EXTENSION_SETTINGS_JS.read_text(encoding="utf-8")) + page.add_script_tag(content=UI_JS) + page.add_script_tag(content=MESSAGES_JS) + page.evaluate(_STABILIZE_UNRELATED_UI) + return page.evaluate(_LIFECYCLE_SCENARIO, {"kind": kind}) + finally: + page.close() + + +@pytest.mark.parametrize( + ("kind", "terminal_type", "terminal_status", "last_content"), + [ + ("done", "turn:complete", "completed", "settled-done"), + ("apperror", "turn:error", "provider_error", "settled-error"), + ("apperror-cancel", "turn:cancel", "interrupted", "settled-interrupted"), + ("cancel", "turn:cancel", "cancelled", "settled-cancel"), + ( + "connection-error", + "turn:error", + "connection_lost", + "**Connection interrupted:** The browser lost the live SSE connection before the response finished. If the worker completed, reopening this session should restore the settled transcript.", + ), + ], +) +def test_real_sse_terminal_callback_observes_settled_core_state( + lifecycle_browser, + kind, + terminal_type, + terminal_status, + last_content, +): + lifecycle = _run_lifecycle_scenario(lifecycle_browser, kind) + + assert [entry["event"]["type"] for entry in lifecycle] == [ + "turn:start", + terminal_type, + ] + start, terminal = lifecycle + assert start["activeStreamId"] == f"stream-{kind}" + assert start["busy"] is True + assert terminal["event"]["sessionId"] == "session-a" + assert terminal["event"]["streamId"] == f"stream-{kind}" + assert terminal["event"]["status"] == terminal_status + assert terminal["activeStreamId"] is None + assert terminal["busy"] is False + assert terminal["currentSid"] == "session-a" + assert terminal["lastContent"] == last_content + + def test_live_stream_terminal_paths_use_original_stream_owner_identity(): attach_start = MESSAGES_JS.index("function attachLiveStream(") attach_body = MESSAGES_JS[attach_start : MESSAGES_JS.index("\nfunction transcript(", attach_start)] @@ -223,6 +474,22 @@ def test_live_stream_terminal_paths_use_original_stream_owner_identity(): assert "_dispatchExtensionTurnLifecycle('turn:cancel',activeSid,streamId" in cancel assert "_dispatchExtensionTurnLifecycle('turn:error',activeSid,streamId" in connection_error + assert done.index("_setActivePaneIdleIfOwner()") < done.index( + "_dispatchExtensionTurnLifecycle('turn:complete'" + ) + assert application_error.index("renderSessionList()") < application_error.index( + "_dispatchExtensionTurnLifecycle(_extensionErrorType" + ) + assert cancel.index("await api(") < cancel.index( + "_dispatchExtensionTurnLifecycle('turn:cancel'" + ) + assert cancel.index("_setActivePaneIdleIfOwner()") < cancel.index( + "_dispatchExtensionTurnLifecycle('turn:cancel'" + ) + assert connection_error.index("_setActivePaneIdleIfOwner()") < connection_error.index( + "_dispatchExtensionTurnLifecycle('turn:error'" + ) + assert "_dispatchExtensionTurnLifecycle('turn:complete',completedSid" not in done assert "_dispatchExtensionTurnLifecycle(_extensionErrorType,d.session_id" not in application_error assert "_dispatchExtensionTurnLifecycle('turn:cancel',_cancelData.session_id" not in cancel diff --git a/tests/test_issue5720_reasoning_owner.py b/tests/test_issue5720_reasoning_owner.py index 6cc31e99..975a7e39 100644 --- a/tests/test_issue5720_reasoning_owner.py +++ b/tests/test_issue5720_reasoning_owner.py @@ -531,6 +531,7 @@ global.EventSource=FakeEventSource; const attachStart=messagesSrc.indexOf('function attachLiveStream('); const attachEnd=messagesSrc.indexOf('\nfunction transcript(){',attachStart); if(attachStart<0||attachEnd<0) throw new Error('attachLiveStream source boundary not found'); +eval(extractFunc(messagesSrc,'_dispatchExtensionTurnLifecycle')); eval(messagesSrc.slice(attachStart,attachEnd)); attachLiveStream('sid-1','stream-1'); const source=FakeEventSource.instances[0];