diff --git a/AGENTS.md b/AGENTS.md index b103761e..713ebedb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -87,7 +87,12 @@ Follow that checklist's safety rules: Do not re-register a cancelled worker; finalize outside the stream lock. Local Steer accepts only explicit starting/running phases. Close admission by publishing finalizing under STREAMS_LOCK before the last pending-steer - drain; earlier accepted guidance is drained, later guidance is rejected. + drain; earlier accepted guidance is drained, later guidance is rejected with + `not_running` while the owned stream is still live (not `stream_dead`). Use + one idempotent terminal settlement before done/error/end and final cleanup, + covering returned errors, exceptions and self-heal, not only the success path. + Merge Agent-returned `pending_steer` with the registered worker's final slot + drain and emit leftovers before terminal events, outside registry locks. Test both drain/Steer orderings, including compression-rotated identities. - Inactive-session recovery is separate from live Steer. Resolve durable compression lineage in the session's profile database, read-only, even when diff --git a/api/compression_continuation.py b/api/compression_continuation.py index d4233116..410f2e66 100644 --- a/api/compression_continuation.py +++ b/api/compression_continuation.py @@ -45,7 +45,10 @@ def durable_compression_continuation(session): return True, None # Only the observed automatic idle closure is resumable here. Explicit # resets/closures and unknown future reasons must not become redirects. - if child.get("ended_at") is not None and child.get("end_reason") != "idle_timeout": + reason = child.get("end_reason") + if reason not in (None, "", "idle_timeout"): + return True, None + if child.get("ended_at") is not None and reason != "idle_timeout": return True, None if db.get_compression_tip(sid) != tip: return True, None diff --git a/api/streaming.py b/api/streaming.py index 4a6cdf3f..29cda8be 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -9541,8 +9541,42 @@ def _run_agent_streaming( _metering_thread = threading.Thread(target=_metering_ticker, daemon=True) _success_writeback_committed = False + _steer_settled = False + _returned_pending_steer = [] + + def _remember_pending_steer_result(value): + text = value.get('pending_steer') if isinstance(value, dict) else None + if isinstance(text, str) and text: + _returned_pending_steer.append(text) + + def _settle_pending_steer(): + # Single terminal boundary for success, returned errors, exceptions and + # successful self-heal. Admission closes before the last slot consumer; + # registry locks never cover SSE writes or Agent drain callbacks. + nonlocal _steer_settled + with STREAMS_LOCK: + if _steer_settled: + return + _steer_settled = True + target = AGENT_INSTANCES.get(stream_id) or agent + if stream_id in STREAMS and not cancel_event.is_set(): + update_active_run(stream_id, phase="finalizing") + leftovers = list(_returned_pending_steer) + try: + drain = getattr(target, '_drain_pending_steer', None) + text = drain() if callable(drain) else None + if text: + leftovers.append(str(text)) + except Exception: + logger.debug("Failed to drain pending steer for session %s", session_id) + if leftovers: + put('pending_steer_leftover', { + 'session_id': session_id, 'text': '\n'.join(leftovers), + }) def put(event, data): + if event in ('done', 'apperror', 'stream_end'): + _settle_pending_steer() # If cancelled, drop all further events except the cancel event itself if cancel_event.is_set() and not _success_writeback_committed and event not in ('cancel', 'apperror'): return @@ -11272,6 +11306,7 @@ def _run_agent_streaming( _run_conversation_kwargs["user_message"] = user_message _result_partial_pre_call_context = list(_previous_context_messages) result = agent.run_conversation(**_run_conversation_kwargs) + _remember_pending_steer_result(result) _active_turn_identity = _resolve_active_turn_authority( _active_turn_identity, result=result, @@ -11807,6 +11842,7 @@ def _run_agent_streaming( _heal_context_messages ) _heal_result = agent.run_conversation(**_heal_kwargs) + _remember_pending_steer_result(_heal_result) _active_turn_identity = _resolve_active_turn_authority( _active_turn_identity, result=_heal_result, @@ -12753,30 +12789,7 @@ def _run_agent_streaming( else None ) # (reasoning trace already attached + saved above, before s.save()) - # Leftover-steer delivery: if a /steer was queued (via - # api/chat/steer) but the agent finished its turn before - # reaching a tool-result boundary that would consume it, - # the text is still stashed in agent._pending_steer. Drain - # it now and emit a pending_steer_leftover SSE event so the - # frontend can queue it for the next turn — same fallback - # path as the CLI in cli.py:8788-8794. - # Close Steer admission before its last consumer runs. Enqueues - # share this lock: earlier guidance is drained below; later requests - # fail closed instead of reporting success on an undeliverable input. - # Stop may have claimed first; never overwrite its cancelling phase. - with STREAMS_LOCK: - if stream_id in STREAMS: - update_active_run(stream_id, phase="finalizing") - try: - _drain_pending_steer = getattr(agent, '_drain_pending_steer', None) - _leftover = _drain_pending_steer() if _drain_pending_steer else None - if _leftover: - put('pending_steer_leftover', { - 'session_id': session_id, - 'text': str(_leftover), - }) - except Exception: - logger.debug("Failed to drain pending steer for session %s", session_id) + _settle_pending_steer() # /goal parity: after a successful assistant turn, run the Hermes # GoalManager judge before terminal done/stream_end events. The # frontend surfaces the status line and queues continuation_prompt as @@ -13180,6 +13193,7 @@ def _run_agent_streaming( _heal_context_messages ) _heal_result = _heal_agent.run_conversation(**_heal_kwargs2) + _remember_pending_steer_result(_heal_result) _active_turn_identity = _resolve_active_turn_authority( _active_turn_identity, result=_heal_result, @@ -13433,6 +13447,7 @@ def _run_agent_streaming( _error_payload['old_session_id'] = session_id put('apperror', _error_payload) finally: + _settle_pending_steer() # #4633/#2476: symmetric metering teardown. begin_session() (top of the # outer try) had no paired end_session(), so zero-token turns leaked a # _sessions[stream_id] entry that get_stats() pruning never reclaims (its @@ -13460,7 +13475,6 @@ def _run_agent_streaming( if (s is not None and getattr(s, 'active_stream_id', None) == stream_id and getattr(s, 'pending_user_message', None)): - update_active_run(stream_id, phase="finalizing") _last_resort_sync_from_core(s, stream_id, _agent_lock) _clear_thread_env() # TD1: always clear thread-local context if _streaming_cron_profile_home_token is not None: @@ -13580,6 +13594,11 @@ def _steer_bound_stream(sid: str, stream_id: str, text: str) -> dict | None: or run.get("phase") == "cancelling"): return {"accepted": False, "fallback": "stream_dead", "stream_id": None} return {"accepted": False, "fallback": "gateway_steer_queued", "stream_id": stream_id} + if (stream_id in cfg.STREAMS and owner == sid + and run.get("session_id") == sid + and run.get("backend") == WEBUI_LOCAL_CHAT_BACKEND + and run.get("phase") == "finalizing"): + return {"accepted": False, "fallback": "not_running", "stream_id": stream_id} if agent is None: if ((owner and owner != sid) or (run.get("session_id") and run["session_id"] != sid) @@ -13734,6 +13753,9 @@ def _handle_chat_steer(handler, body: dict) -> bool: # Gateway owns transport; a local cache object is never steered. result = {"accepted": False, "fallback": "gateway_steer_queued", "stream_id": active_stream_id} + elif backend == WEBUI_LOCAL_CHAT_BACKEND and run.get("phase") == "finalizing": + result = {"accepted": False, "fallback": "not_running", + "stream_id": active_stream_id} elif (backend == WEBUI_LOCAL_CHAT_BACKEND and run.get("phase") in _LOCAL_STEERABLE_PHASES): try: diff --git a/static/messages.js b/static/messages.js index 86aa3b29..e7bcb73c 100644 --- a/static/messages.js +++ b/static/messages.js @@ -1368,7 +1368,7 @@ function _restoreComposerDraftAfterFailedSend(draftText, filesSnapshot, sid, cle } else if(!restoredVisible){ // Background failure (sid was never the visible session): no live // composer to read, so persist the captured snapshot — it's the only copy. - _saveComposerDraftNow(sid, restore, []); + _saveComposerDraftNow(sid, restore, files); } // else: restored the visible composer, then the user switched away — the // session-switch save path already saved sid's composer; skip stale write. diff --git a/tests/test_compressed_idle_resume.py b/tests/test_compressed_idle_resume.py index 7644460a..a6da417e 100644 --- a/tests/test_compressed_idle_resume.py +++ b/tests/test_compressed_idle_resume.py @@ -12,7 +12,11 @@ from api import routes, profiles @pytest.fixture def lineage(tmp_path, monkeypatch): - SessionDB = pytest.importorskip("hermes_state").SessionDB + state = pytest.importorskip("hermes_state") + # Worker tests restore sys.path after loading Agent lazily. Scope its + # sibling-module lookup to this fixture, never the global test importer. + monkeypatch.syspath_prepend(str(Path(state.__file__).parent)) + SessionDB = state.SessionDB db = SessionDB(tmp_path / "state.db") db.create_session("sealedparent", source="webui") db.create_session("idlechild", source="tui", parent_session_id="sealedparent") @@ -69,9 +73,10 @@ def test_idle_tip_accepts_normal_persistence_without_reopening_parent(lineage): @pytest.mark.parametrize("reason", ["reset", "user_closed", "compression"]) -def test_unknown_or_sealed_terminal_tip_is_not_redirected(lineage, reason): +@pytest.mark.parametrize("ended_at", [None, 1234.0]) +def test_unknown_or_sealed_terminal_tip_is_not_redirected(lineage, reason, ended_at): db, session = lineage - db._conn.execute("UPDATE sessions SET end_reason=? WHERE id='idlechild'", (reason,)) + db._conn.execute("UPDATE sessions SET end_reason=?, ended_at=? WHERE id='idlechild'", (reason, ended_at)) db._conn.commit() assert routes._pre_compression_continuation_session_id(session) is None @@ -128,3 +133,22 @@ assert.equal(calls.length,2); })().catch(e=>{console.error(e);process.exit(1)}); ''' subprocess.run(['node', '-e', helper + script], check=True, timeout=15) + + +def test_background_failed_draft_preserves_attachments_after_navigation_race(): + source = (Path(__file__).parents[1] / 'static/messages.js').read_text() + helper = source[source.index('function _restoreComposerDraftAfterFailedSend('):source.index('async function send(){')] + script = r''' +const assert=require('node:assert/strict'); +const S={session:{session_id:'unrelated'},pendingFiles:[]}; +const $=()=>{throw Error('unrelated visible composer must not be touched')}; +const calls=[]; +const _saveComposerDraftNow=(...args)=>calls.push(args); +(async()=>{ +const files=[{name:'drawing.png',path:'/uploads/drawing.png'}]; +_restoreComposerDraftAfterFailedSend('retained',files,'continuation',Promise.resolve()); +await Promise.resolve(); +assert.deepEqual(calls,[['continuation','retained',files]]); +})().catch(e=>{console.error(e);process.exit(1)}); +''' + subprocess.run(['node','-e',helper+script], check=True, timeout=15) diff --git a/tests/test_real_steer.py b/tests/test_real_steer.py index e131f166..de85ff32 100644 --- a/tests/test_real_steer.py +++ b/tests/test_real_steer.py @@ -1026,7 +1026,7 @@ class TestLeftoverDelivery: on the same turn.""" src = (Path(__file__).parent.parent / "api" / "streaming.py").read_text(encoding="utf-8") # Find the drain invocation and the next put('done', ...) AFTER it - drain_idx = src.find("_drain_pending_steer()") + drain_idx = src.find("_settle_pending_steer()", src.index(" def put(event, data):", src.index("def _run_agent_streaming("))) assert drain_idx >= 0 done_idx = src.find("put('done'", drain_idx) assert done_idx >= 0 diff --git a/tests/test_steer_active_run_identity.py b/tests/test_steer_active_run_identity.py index a8ca1b2b..84979ad2 100644 --- a/tests/test_steer_active_run_identity.py +++ b/tests/test_steer_active_run_identity.py @@ -247,4 +247,4 @@ def test_local_steer_requires_an_explicit_consuming_phase(scene, registered, pha assert result["accepted"] is (phase in {"starting", "running"}) assert agent.steer.call_count == (1 if result["accepted"] else 0) if not result["accepted"]: - assert result["fallback"] == "stream_dead" + assert result["fallback"] == ("not_running" if phase == "finalizing" else "stream_dead") diff --git a/tests/test_steer_cancel_ordering.py b/tests/test_steer_cancel_ordering.py index e7628164..5927aa72 100644 --- a/tests/test_steer_cancel_ordering.py +++ b/tests/test_steer_cancel_ordering.py @@ -264,6 +264,6 @@ def test_cache_only_selection_revalidates_phase_before_enqueue(scene, monkeypatc return matched monkeypatch.setattr(streaming, "_cached_agent_matches_session", select_then_finalize) - assert steer() == {"accepted": False, "fallback": "stream_dead", "stream_id": None} + assert steer() == {"accepted": False, "fallback": "not_running", "stream_id": "run"} assert selected == [agent] agent.steer.assert_not_called() diff --git a/tests/test_steer_terminal_settlement.py b/tests/test_steer_terminal_settlement.py new file mode 100644 index 00000000..1824d9b8 --- /dev/null +++ b/tests/test_steer_terminal_settlement.py @@ -0,0 +1,68 @@ +"""All terminal worker exits must close Steer and preserve accepted guidance.""" +import pytest +from api import config +from tests import test_steer_worker_boundaries as boundaries + +worker_scene = boundaries.worker_scene + + +@pytest.mark.parametrize('outcome', ['success', 'returned-error', 'exception']) +def test_terminal_exit_fences_before_event_and_keeps_guidance(worker_scene, monkeypatch, outcome): + scene = worker_scene + observed = [] + put = scene.events.put_nowait + + def observe(item): + if item[0] in ('done', 'apperror'): + observed.append(config.ACTIVE_RUNS.get('run', {}).get('phase')) + return put(item) + + monkeypatch.setattr(scene.events, 'put_nowait', observe) + + def running(): + scene.agent.session_id = 'compressed-child' + scene.agent.steer('retained guidance') + if outcome == 'exception': + raise RuntimeError('terminal worker failure') + if outcome == 'returned-error': + scene.result = {'messages': [], 'error': 'terminal worker failure'} + else: + # The real Agent consumes this slot into its returned result. + text = scene.agent._drain_pending_steer() + scene.result = {'messages': [ + {'role': 'user', 'content': 'Do the task.'}, + {'role': 'assistant', 'content': 'Finished.'}, + ], 'pending_steer': text} + + scene.on_run = running + scene.run() + assert observed and all(phase == 'finalizing' for phase in observed) + events = list(scene.events.queue) + leftovers = [data['text'] for event, data in events if event == 'pending_steer_leftover'] + assert leftovers == ['retained guidance'] + assert next(i for i, item in enumerate(events) if item[0] == 'pending_steer_leftover') < next(i for i, item in enumerate(events) if item[0] in ('done', 'apperror')) + assert scene.agent.pending == [] + + +def test_finalizing_response_preserves_browser_live_stream(): + import subprocess + from pathlib import Path + source = (Path(__file__).parents[1] / 'static/commands.js').read_text() + fn = source[source.index('async function _trySteer('):source.index('async function cmdTitle(')] + predicate = source[source.index('function _steerFallbackIsDeadRun('):source.index('function _steerOwnerStreamIsCurrent(')] + script = r''' +const assert=require('node:assert/strict'); +const S={session:{session_id:'s',active_stream_id:'run'},activeStreamId:'run',busy:true,pendingFiles:[]}; +const INFLIGHT={s:{stream_id:'run'}},inp={value:''}; +const $=()=>inp,t=x=>x,showToast=()=>{}; +const _steerTextWithPendingFiles=async x=>x,_steerOwnerIsCurrent=()=>true,_steerRestoreText=x=>x; +const _steerFailureMessageKey=x=>x,_showSteerRecovery=()=>{}; +const _steerClearCurrentOwnerDeadRun=()=>{S.busy=false;S.activeStreamId=null;delete INFLIGHT.s}; +const api=async()=>({accepted:false,fallback:'not_running',stream_id:'run'}); +(async()=>{ +assert.equal(await _trySteer('guidance',true),false); +assert.equal(S.busy,true);assert.equal(S.activeStreamId,'run'); +assert.equal(INFLIGHT.s.stream_id,'run');assert.equal(inp.value,'guidance'); +})().catch(e=>{console.error(e);process.exit(1)}); +''' + subprocess.run(['node','-e',fn+predicate+script],check=True,timeout=15) diff --git a/tests/test_steer_worker_boundaries.py b/tests/test_steer_worker_boundaries.py index 5c06796d..d55d3691 100644 --- a/tests/test_steer_worker_boundaries.py +++ b/tests/test_steer_worker_boundaries.py @@ -53,7 +53,7 @@ def worker_scene(tmp_path, monkeypatch): maps["STREAM_SESSION_OWNERS"]["run"] = "original" scene = SimpleNamespace(session=session, events=events, lock=lock, agent=None, on_init=lambda: None, on_run=lambda: None, - on_drain=lambda: None, calls=[], drained=[]) + on_drain=lambda: None, calls=[], drained=[], result=None) class FakeAgent: def __init__(self, **kwargs): @@ -71,6 +71,8 @@ def worker_scene(tmp_path, monkeypatch): def run_conversation(self, **kwargs): scene.calls.append("run") scene.on_run() + if scene.result is not None: + return scene.result return {"messages": [ {"role": "user", "content": "Do the task."}, {"role": "assistant", "content": "Finished."}, @@ -261,7 +263,7 @@ def test_final_drain_fences_steer(worker_scene, monkeypatch, registered, rotated if first == "steer": assert result == {"accepted": True, "fallback": None, "stream_id": "run"} else: - assert result == {"accepted": False, "fallback": "stream_dead", "stream_id": None} + assert result == {"accepted": False, "fallback": "not_running", "stream_id": "run"} assert scene.agent.pending == [] finally: release.set()