diff --git a/CHANGELOG.md b/CHANGELOG.md index b904d91c..5bd9a0c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ ### Fixed +- **Steering a live *gateway-owned* run no longer errors and drops your message.** When you steered a stream backed by a gateway run (which has no process-local agent in `SESSION_AGENT_CACHE`), the old path showed a misleading "No agent cached for this session" toast and lost the text you'd typed. WebUI now detects the live gateway run and queues your steer as the next turn on that session (snapshotting model/provider/profile first), updates the queue badge, clears the composer draft, and shows a "Steer queued for next turn" notice — without cancelling the active stream. This is the gateway sibling of the existing local-agent queued-steer fallback. Thanks @rodboev. (#5823, #5585) + - **CORS preflight (`OPTIONS`) responses are now framed with `Content-Length: 0`.** Under HTTP/1.1 keep-alive the preflight `200` was sent with no `Content-Length`, so a browser/proxy/curl client read the body until connection close (RFC 7230 §3.3.3) — a keep-alive preflight would hang to the 30s timeout and couldn't reuse the connection. The handler now emits `Content-Length: 0` before `end_headers()`; status stays `200` and CORS headers are unchanged. Thanks @ai-ag2026. (#5828) - **The session index is now parsed from bytes instead of a decoded string.** The `_index.json` read sites decoded to `str` before `json.loads`; they now pass `read_bytes()` directly (`json.loads` accepts bytes with RFC-4627 auto-detect), which is behavior-equivalent for the BOM-less UTF-8 the app writes and slightly more tolerant of a BOM. Shaves a redundant decode off the session-list hot path. Thanks @ai-ag2026. (#5834) diff --git a/api/streaming.py b/api/streaming.py index 3b8d687f..0455eca9 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -10057,7 +10057,29 @@ def _handle_chat_steer(handler, body: dict) -> bool: except Exception: logger.debug("Failed to close steer identity-mismatched cached agent for session %s", sid, exc_info=True) if not cached: - # No active agent for this session — caller surfaces a steer failure + try: + s = get_session(sid) + active_stream_id = getattr(s, "active_stream_id", None) or None + except KeyError: + active_stream_id = None + if active_stream_id: + with _cfg.STREAMS_LOCK: + stream_alive = active_stream_id in _cfg.STREAMS + if stream_alive: + try: + with _cfg.ACTIVE_RUNS_LOCK: + active_run = dict((_cfg.ACTIVE_RUNS or {}).get(str(active_stream_id)) or {}) + if active_run.get("backend") == "gateway": + return j(handler, {"accepted": False, "fallback": "gateway_steer_queued", + "stream_id": active_stream_id}) + except Exception: + logger.warning( + "Gateway ownership lookup failed before steer fallback for session=%s stream_id=%s", + sid, + active_stream_id, + exc_info=True, + ) + # No active local agent for this session — caller surfaces a steer failure # without cancelling the active run. return j(handler, {"accepted": False, "fallback": "no_cached_agent", "stream_id": None}) diff --git a/static/commands.js b/static/commands.js index ddf2d2bf..2c4d89ff 100644 --- a/static/commands.js +++ b/static/commands.js @@ -1350,6 +1350,7 @@ async function cmdSteer(args){ } function _steerFailureMessageKey(fallback) { + if(fallback==='gateway_steer_queued')return 'steer_fail_no_cached_agent'; const key = 'steer_fail_' + (fallback || 'unknown'); return (typeof LOCALES !== 'undefined' && LOCALES.en && LOCALES.en[key]) ? key : 'steer_fail_unknown'; @@ -1546,6 +1547,10 @@ async function _trySteer(msg, explicitSteer){ const ownerSid=(typeof S!=='undefined'&&S.session&&S.session.session_id)||null; const ownerStreamId=(typeof S!=='undefined'&&(S.activeStreamId||(S.session&&S.session.active_stream_id)))||null; const pendingFilesSnapshot=typeof S!=='undefined'&&Array.isArray(S.pendingFiles)?[...S.pendingFiles]:[]; + const ownerProfile=typeof S!=='undefined'&&(S.activeProfile||'default'); + const ownerModelState=typeof _chatPayloadModelState==='function' + ? _chatPayloadModelState() + : {model:(typeof S!=='undefined'&&S.session&&S.session.model)||'',model_provider:(typeof S!=='undefined'&&S.session&&S.session.model_provider)||''}; if(!ownerSid){showToast(t('no_active_session'));return false;} let steerText=originalMsg; try{ @@ -1612,6 +1617,25 @@ async function _trySteer(msg, explicitSteer){ showToast(t('cmd_steer_delivered'),2500); return true; } + if(result&&result.fallback==='gateway_steer_queued'&&typeof queueSessionMessage==='function'){ + _steerUploadCache=null; + queueSessionMessage(ownerSid,{ + text:originalMsg, + files:pendingFilesSnapshot, + model:ownerModelState.model, + model_provider:ownerModelState.model_provider, + profile:ownerProfile, + }); + if(typeof updateQueueBadge==='function')updateQueueBadge(ownerSid); + if(ownerSid&&typeof _clearComposerDraft==='function') _clearComposerDraft(ownerSid,_steerRestoreText(originalMsg,explicitSteer),pendingFilesSnapshot); + if(_steerOwnerIsCurrent(ownerSid)&&typeof S!=='undefined'&&Array.isArray(S.pendingFiles)&&S.pendingFiles.length&&pendingFilesSnapshot.length){ + const _queued=new Set(pendingFilesSnapshot); + const _remaining=S.pendingFiles.filter(f=>!_queued.has(f)); + if(_remaining.length!==S.pendingFiles.length){S.pendingFiles=_remaining;if(typeof renderTray==='function')renderTray();} + } + showToast(t('steer_leftover_queued'),3000); + return true; + } // Do not fall back to interrupt: Steer failure is not permission to cancel // the active run. Restore the draft so the user can explicitly Queue or // Interrupt if that is what they want next. Pending files remain staged. diff --git a/tests/test_1062_busy_input_modes.py b/tests/test_1062_busy_input_modes.py index 80204a1c..347e3495 100644 --- a/tests/test_1062_busy_input_modes.py +++ b/tests/test_1062_busy_input_modes.py @@ -96,20 +96,23 @@ class TestSlashCommandHandlers: assert "cancelStream" in body, "/interrupt must call cancelStream() so the drain re-sends" def test_cmd_steer_delegates_to_try_steer(self): - """/steer delegates to _trySteer which calls /api/chat/steer with - a non-destructive fallback. The fallback path is exercised by tests - in test_real_steer.py — this test just pins the delegation.""" + """/steer delegates to _trySteer which calls /api/chat/steer. + Fallback behavior is covered in test_real_steer.py; this test pins + delegation and blocks the old cancel-on-failure path.""" idx = COMMANDS_JS.find("async function cmdSteer(") assert idx >= 0 body = COMMANDS_JS[idx:idx + 800] - # cmdSteer delegates to _trySteer; fallback must not queue+cancel. + # cmdSteer delegates to _trySteer; fallback must not cancel. assert "_trySteer" in body, "cmdSteer must call _trySteer to use the real /api/chat/steer endpoint" - # The shared helper must contain the non-destructive fallback path. + # The shared helper must contain the non-cancelling fallback path. helper_idx = COMMANDS_JS.find("async function _trySteer(") assert helper_idx >= 0, "_trySteer helper must exist" helper_body = _source_between(COMMANDS_JS, "async function _trySteer(", "\nasync function cmdTitle") - assert "queueSessionMessage" not in helper_body assert "cancelStream" not in helper_body + gateway_fallback_idx = helper_body.find("result&&result.fallback==='gateway_steer_queued'") + gateway_queue_idx = helper_body.find("queueSessionMessage", gateway_fallback_idx) + assert gateway_fallback_idx >= 0 + assert gateway_queue_idx > gateway_fallback_idx assert "inp.value" in helper_body assert "if(result&&result.accepted)" in helper_body assert "S.pendingFiles=_remaining" in helper_body diff --git a/tests/test_issue4749_steer_reason_and_recovery.py b/tests/test_issue4749_steer_reason_and_recovery.py index 7a946b58..4726da56 100644 --- a/tests/test_issue4749_steer_reason_and_recovery.py +++ b/tests/test_issue4749_steer_reason_and_recovery.py @@ -39,6 +39,10 @@ BACKEND_CODES = { "steer_error", } +HANDLED_NON_RECOVERY_CODES = { + "gateway_steer_queued", +} + FRONTEND_NETWORK_CODE = "network_error" @@ -73,6 +77,7 @@ def test_reason_map_contract(): }}; function _steerFailureMessageKey(fallback) { + if (fallback === 'gateway_steer_queued') return 'steer_fail_no_cached_agent'; const key = 'steer_fail_' + (fallback || 'unknown'); return (typeof LOCALES !== 'undefined' && LOCALES.en && LOCALES.en[key]) ? key : 'steer_fail_unknown'; @@ -108,6 +113,11 @@ def test_reason_map_contract(): console.error('FAIL undefined: got=' + u); ok = false; } + const gatewayQueued = _steerFailureMessageKey('gateway_steer_queued'); + if (gatewayQueued !== 'steer_fail_no_cached_agent') { + console.error('FAIL gateway_steer_queued: got=' + gatewayQueued); + ok = false; + } process.exit(ok ? 0 : 1); """) result = subprocess.run([node, "-e", script], capture_output=True, text=True) @@ -129,10 +139,10 @@ def test_backend_parity(): if c # non-empty after filtering ) assert found_codes, "No fallback codes found in _handle_chat_steer" - assert found_codes == BACKEND_CODES, ( + assert found_codes == BACKEND_CODES | HANDLED_NON_RECOVERY_CODES, ( f"Backend fallback codes mismatch.\n" f" Found: {sorted(found_codes)}\n" - f" Expected: {sorted(BACKEND_CODES)}" + f" Expected: {sorted(BACKEND_CODES | HANDLED_NON_RECOVERY_CODES)}" ) # Also confirm frontend adds network_error commands_text = COMMANDS_JS.read_text(encoding="utf-8") diff --git a/tests/test_real_steer.py b/tests/test_real_steer.py index 67835a43..ea9ca820 100644 --- a/tests/test_real_steer.py +++ b/tests/test_real_steer.py @@ -41,13 +41,23 @@ def _restore_auth_sessions(): @pytest.fixture def _clear_caches(): """Snapshot SESSION_AGENT_CACHE and STREAMS so tests don't bleed.""" - from api.config import SESSION_AGENT_CACHE, SESSION_AGENT_CACHE_LOCK, STREAMS, STREAMS_LOCK + from api.config import ( + ACTIVE_RUNS, + ACTIVE_RUNS_LOCK, + SESSION_AGENT_CACHE, + SESSION_AGENT_CACHE_LOCK, + STREAMS, + STREAMS_LOCK, + ) with SESSION_AGENT_CACHE_LOCK: cache_snap = dict(SESSION_AGENT_CACHE) SESSION_AGENT_CACHE.clear() with STREAMS_LOCK: streams_snap = dict(STREAMS) STREAMS.clear() + with ACTIVE_RUNS_LOCK: + active_runs_snap = dict(ACTIVE_RUNS) + ACTIVE_RUNS.clear() yield with SESSION_AGENT_CACHE_LOCK: SESSION_AGENT_CACHE.clear() @@ -55,6 +65,9 @@ def _clear_caches(): with STREAMS_LOCK: STREAMS.clear() STREAMS.update(streams_snap) + with ACTIVE_RUNS_LOCK: + ACTIVE_RUNS.clear() + ACTIVE_RUNS.update(active_runs_snap) def _make_handler(): @@ -122,6 +135,30 @@ class TestHandleChatSteerFallbacks: assert body["accepted"] is False assert body["fallback"] == "no_cached_agent" + def test_gateway_owned_stream_without_cached_agent_queues_fallback(self, _clear_caches): + from api.streaming import _handle_chat_steer + from api.config import ACTIVE_RUNS, ACTIVE_RUNS_LOCK, STREAMS, STREAMS_LOCK + import queue as _q + + sid, stream_id = "sid_gateway", "stream_gateway" + with STREAMS_LOCK: + STREAMS[stream_id] = _q.Queue() + with ACTIVE_RUNS_LOCK: + ACTIVE_RUNS[stream_id] = {"session_id": sid, "backend": "gateway"} + + sess = MagicMock() + sess.active_stream_id = stream_id + with patch("api.streaming.get_session", return_value=sess): + handler = _make_handler() + _handle_chat_steer(handler, {"session_id": sid, "text": "preserve this"}) + + body = _captured_response(handler) + assert body == { + "accepted": False, + "fallback": "gateway_steer_queued", + "stream_id": stream_id, + } + def test_agent_lacks_steer_method(self, _clear_caches): from api.streaming import _handle_chat_steer from api.config import SESSION_AGENT_CACHE, SESSION_AGENT_CACHE_LOCK @@ -272,9 +309,10 @@ class TestFrontendWiring: def test_try_steer_handles_fallback_without_cancelling(self): idx = self.cmds.find("async function _trySteer(") body = _source_between(self.cmds, "async function _trySteer(", "\nasync function cmdTitle") - # Must check result.accepted and surface fallback without queueing or cancelling. + # Must check result.accepted and keep generic failures from cancelling. assert "result&&result.accepted" in body or "result.accepted" in body - assert "queueSessionMessage" not in body + assert "result&&result.fallback==='gateway_steer_queued'" in body + assert "queueSessionMessage(ownerSid" in body assert "cancelStream" not in body, "fallback path must not cancel the stream" assert "inp.value" in body, "fallback path must restore the composer draft" @@ -541,11 +579,11 @@ class TestFrontendWiring: function _steerFailureMessageKey(fallback){{return 'steer_fail_'+fallback;}} function scrollToBottom(){{}} function setComposerStatus(){{}} - function showToast(){{}} - function renderTray(){{}} + function showToast(key){{if(globalThis.__toasts)globalThis.__toasts.push(key);}} + function renderTray(){{if(globalThis.__trayRenders)globalThis.__trayRenders.count += 1;}} function autoResize(){{}} function _showSteerIndicator(){{}} - function _clearComposerDraft(){{}} + function _clearComposerDraft(sid,text,files){{if(globalThis.__draftClears)globalThis.__draftClears.push({{sid,text,files}});}} async function uploadPendingFiles(){{return [];}} eval(steerSrc); @@ -643,6 +681,83 @@ class TestFrontendWiring: assert.strictEqual(apiCalls, 2); }} + async function runGatewayQueuedFallback(switchDuringAwait=false){{ + let input = {{value:''}}; + let clearInflightCalls = []; + let updateSendBtnCalls = 0; + let queued = []; + let queueBadges = []; + let draftClears = []; + let trayRenders = 0; + let toasts = []; + let submittedFile = {{name:'a.pdf'}}; + let replacementFile = {{name:'replacement.pdf'}}; + let apiPayload = null; + inner = makeElement('div'); + globalThis.S = {{ + session:{{session_id:'A', active_stream_id:'stream-1', model:'fallback-model', model_provider:'fallback-provider'}}, + activeStreamId:'stream-1', + activeProfile:'work', + busy:true, + pendingFiles:[submittedFile], + }}; + globalThis.INFLIGHT = {{A:{{messages:[]}}}}; + globalThis.$ = id => input; + globalThis.clearInflightState = sid => clearInflightCalls.push(sid); + globalThis.updateSendBtn = () => {{updateSendBtnCalls += 1;}}; + globalThis.queueSessionMessage = (sid, payload) => queued.push({{sid, payload}}); + globalThis.updateQueueBadge = sid => queueBadges.push(sid); + globalThis.__draftClears = draftClears; + globalThis.__trayRenders = {{count:0}}; + globalThis.__toasts = toasts; + globalThis._chatPayloadModelState = () => ({{model:'captured-model', model_provider:'captured-provider'}}); + globalThis.api = async (url, options) => {{ + assert.strictEqual(url, '/api/chat/steer'); + apiPayload = JSON.parse(options.body); + if(switchDuringAwait){{ + S.session={{session_id:'B', active_stream_id:'stream-B'}}; + S.activeStreamId='stream-B'; + S.pendingFiles=[replacementFile]; + }}else{{ + S.pendingFiles=[submittedFile, replacementFile]; + }} + return {{accepted:false, fallback:'gateway_steer_queued'}}; + }}; + + const delivered = await _trySteer('queue me', false); + assert.strictEqual(delivered, true); + assert.deepStrictEqual(apiPayload, {{session_id:'A', text:'queue me'}}); + assert.strictEqual(S.busy, true); + assert.ok(Object.prototype.hasOwnProperty.call(INFLIGHT, 'A')); + assert.deepStrictEqual(clearInflightCalls, []); + assert.strictEqual(updateSendBtnCalls, 0); + assert.strictEqual(inner.children.length, 0); + assert.deepStrictEqual(queueBadges, ['A']); + assert.strictEqual(queued.length, 1); + assert.strictEqual(queued[0].sid, 'A'); + assert.strictEqual(queued[0].payload.text, 'queue me'); + assert.deepStrictEqual(queued[0].payload.files, [submittedFile]); + assert.strictEqual(queued[0].payload.model, 'captured-model'); + assert.strictEqual(queued[0].payload.model_provider, 'captured-provider'); + assert.strictEqual(queued[0].payload.profile, 'work'); + assert.strictEqual(draftClears.length, 1); + assert.strictEqual(draftClears[0].sid, 'A'); + assert.strictEqual(draftClears[0].text, 'queue me'); + assert.deepStrictEqual(draftClears[0].files, [submittedFile]); + assert.deepStrictEqual(toasts, ['steer_leftover_queued']); + if(switchDuringAwait){{ + assert.strictEqual(S.session.session_id, 'B'); + assert.deepStrictEqual(S.pendingFiles, [replacementFile]); + assert.strictEqual(globalThis.__trayRenders.count, 0); + }}else{{ + assert.deepStrictEqual(S.pendingFiles, [replacementFile]); + assert.strictEqual(globalThis.__trayRenders.count, 1); + }} + delete globalThis.__draftClears; + delete globalThis.__trayRenders; + delete globalThis.__toasts; + }} + async function runLateDeadFallbackDoesNotClearNewStream(){{ let input = {{value:''}}; let clearInflightCalls = []; @@ -713,6 +828,8 @@ class TestFrontendWiring: (async()=>{{ await runNoCachedAgentFallback(); await runNoCachedAgentFallback(true); + await runGatewayQueuedFallback(false); + await runGatewayQueuedFallback(true); await runStreamDeadFallback(); await runStreamDeadFallback(true); await runStreamDeadFallback(true, '/help');