diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 66a4024809b9..e943ad7329b9 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1514,6 +1514,19 @@ def _ensure_hermes_home_managed(home: Path): # session_search and recoverable, not deleted. # Default False during rollout; will flip on # after live validation. + "idle": { # Gateway idle pre-compression. Opt-in: when + # enabled, the Gateway schedules a quiet background + # compact pass after a session has been idle so the + # next prompt is less likely to block on preflight + # compression. + "enabled": False, + # threshold is intentionally omitted here: if the user + # does not set compression.idle.threshold explicitly, + # Gateway derives it as compression.threshold * 0.9. + "idle_after_seconds": 120, + "min_interval_seconds": 1800, + "emit_status": False, # keep quiet by default; logs record outcomes. + }, }, # Kanban subsystem (orchestrator workers + dispatcher-driven child tasks). diff --git a/tests/test_tui_gateway_idle_compression.py b/tests/test_tui_gateway_idle_compression.py new file mode 100644 index 000000000000..a01b075175a0 --- /dev/null +++ b/tests/test_tui_gateway_idle_compression.py @@ -0,0 +1,962 @@ +import os +import threading +import types + +import pytest + +from hermes_constants import ( + get_hermes_home, + get_hermes_home_override, + reset_hermes_home_override, + set_hermes_home_override, +) +from tui_gateway import server + + +def _history(): + return [ + {"role": "user", "content": "u1"}, + {"role": "assistant", "content": "a1"}, + {"role": "user", "content": "u2"}, + {"role": "assistant", "content": "a2"}, + ] + + +def _session(agent=None, **extra): + return { + "agent": agent if agent is not None else types.SimpleNamespace(), + "session_key": "session-key", + "history": [], + "history_lock": threading.Lock(), + "history_version": 0, + "running": False, + "last_active": 0.0, + "transport": None, + "attached_images": [], + "cols": 80, + "slash_worker": None, + **extra, + } + + +def _idle_cfg(): + return { + "enabled": True, + "threshold": 0.5, + "idle_after_seconds": 10.0, + "min_interval_seconds": 60.0, + "emit_status": False, + } + + +def _configure_ready_idle_pass(monkeypatch): + monkeypatch.setattr(server, "_load_idle_compression_config", _idle_cfg) + monkeypatch.setattr(server.time, "time", lambda: 100.0) + monkeypatch.setattr( + "agent.model_metadata.estimate_request_tokens_rough", + lambda messages, system_prompt="", tools=None: 600, + ) + monkeypatch.setattr( + server, "_session_compression_threshold_tokens", lambda _agent, _cfg: 500 + ) + + +def _run_in_real_thread(target): + result = [] + errors = [] + + def run(): + try: + result.append(target()) + except BaseException as exc: # surface worker failures in the test thread + errors.append(exc) + + thread = threading.Thread(target=run) + thread.start() + thread.join(timeout=5) + assert not thread.is_alive() + assert errors == [] + return result[0] + + +class _SyncLease: + def __init__(self): + self.release_calls = 0 + + def release(self): + self.release_calls += 1 + + +class _ContendedLock: + """A Lock-compatible Event probe for deterministic finalize handoff tests.""" + + def __init__(self): + self._lock = threading.Lock() + self.contended = threading.Event() + + def acquire(self, *args, **kwargs): + if self._lock.locked(): + self.contended.set() + return self._lock.acquire(*args, **kwargs) + + def release(self): + self._lock.release() + + def __enter__(self): + self.acquire() + return self + + def __exit__(self, *_args): + self.release() + + +def _spy_session_key_approval_effects(monkeypatch): + from tools import approval + + calls = [] + monkeypatch.setattr( + approval, + "is_session_yolo_enabled", + lambda key: calls.append(("is_yolo", key)) or key == "old-key", + ) + monkeypatch.setattr( + approval, + "enable_session_yolo", + lambda key: calls.append(("enable_yolo", key)), + ) + monkeypatch.setattr( + approval, + "disable_session_yolo", + lambda key: calls.append(("disable_yolo", key)), + ) + monkeypatch.setattr( + approval, + "register_gateway_notify", + lambda key, _callback: calls.append(("register_notify", key)), + ) + monkeypatch.setattr( + approval, + "unregister_gateway_notify", + lambda key: calls.append(("unregister_notify", key)), + ) + return calls + + +def _write_idle_config(home, *, enabled): + home.mkdir(parents=True) + (home / "config.yaml").write_text( + "compression:\n" + " enabled: true\n" + " threshold: 0.5\n" + " idle:\n" + f" enabled: {'true' if enabled else 'false'}\n" + " idle_after_seconds: 10\n" + " min_interval_seconds: 60\n", + encoding="utf-8", + ) + + +@pytest.mark.parametrize( + ("default_enabled", "named_enabled", "expected_calls"), + [(False, True, 1), (True, False, 0)], +) +def test_idle_worker_real_thread_uses_named_profile_both_directions( + monkeypatch, tmp_path, default_enabled, named_enabled, expected_calls +): + default_home = tmp_path / "default" + named_home = tmp_path / "profiles" / "research" + _write_idle_config(default_home, enabled=default_enabled) + _write_idle_config(named_home, enabled=named_enabled) + monkeypatch.setenv("HERMES_HOME", str(default_home)) + monkeypatch.setattr(server, "_hermes_home", default_home) + monkeypatch.setattr( + "agent.model_metadata.estimate_request_tokens_rough", + lambda messages, system_prompt="", tools=None: 600, + ) + calls = [] + + def compress(*_args, **_kwargs): + calls.append((get_hermes_home(), get_hermes_home_override())) + return 0, {} + + monkeypatch.setattr(server, "_compress_session_history", compress) + session = _session( + sid="profile-sid", + agent=types.SimpleNamespace( + session_id="session-key", + context_compressor=types.SimpleNamespace(context_length=1000), + ), + history=_history(), + profile_home=str(named_home), + ) + server._sessions["profile-sid"] = session + server._cfg_cache = None + server._cfg_mtime = None + server._cfg_path = None + parent_token = set_hermes_home_override(default_home) + env_before = os.environ.get("HERMES_HOME") + thread_scope = {} + + def worker_call(): + thread_scope["before"] = get_hermes_home() + result = server._run_idle_compression_once("profile-sid", session) + thread_scope["after"] = get_hermes_home() + return result + + try: + assert _run_in_real_thread(worker_call) is False + assert len(calls) == expected_calls + if calls: + assert calls == [(named_home, str(named_home))] + assert thread_scope == {"before": default_home, "after": default_home} + assert get_hermes_home_override() == str(default_home) + assert os.environ.get("HERMES_HOME") == env_before + finally: + reset_hermes_home_override(parent_token) + server._sessions.pop("profile-sid", None) + server._cfg_cache = None + server._cfg_mtime = None + server._cfg_path = None + + +def test_schedule_clears_ambient_profile_for_default_session(monkeypatch, tmp_path): + default_home = tmp_path / "default" + named_home = tmp_path / "profiles" / "disabled" + _write_idle_config(default_home, enabled=True) + _write_idle_config(named_home, enabled=False) + monkeypatch.setenv("HERMES_HOME", str(default_home)) + monkeypatch.setattr(server, "_hermes_home", default_home) + server._cfg_cache = None + server._cfg_mtime = None + server._cfg_path = None + + class FakeThread: + created = [] + + def __init__(self, target, **_kwargs): + self.target = target + self.__class__.created.append(self) + + def start(self): + return None + + def is_alive(self): + return True + + monkeypatch.setattr(server.threading, "Thread", FakeThread) + session = _session(sid="default-sid", profile_home=None) + server._sessions["default-sid"] = session + parent_token = set_hermes_home_override(named_home) + env_before = os.environ.get("HERMES_HOME") + try: + server._schedule_idle_compression("default-sid", session) + assert len(FakeThread.created) == 1 + assert get_hermes_home_override() == str(named_home) + assert os.environ.get("HERMES_HOME") == env_before + finally: + reset_hermes_home_override(parent_token) + server._sessions.pop("default-sid", None) + server._IDLE_COMPRESSION_THREADS.pop("default-sid", None) + server._cfg_cache = None + server._cfg_mtime = None + server._cfg_path = None + + +def test_schedule_registry_backed_deferred_record_without_sid(monkeypatch): + class FakeThread: + created = [] + + def __init__(self, target, **_kwargs): + self.target = target + self.__class__.created.append(self) + + def start(self): + return None + + def is_alive(self): + return True + + monkeypatch.setattr(server.threading, "Thread", FakeThread) + monkeypatch.setattr(server, "_load_idle_compression_config", _idle_cfg) + monkeypatch.setattr(server, "_load_show_reasoning", lambda: False) + monkeypatch.setattr(server, "_load_tool_progress_mode", lambda: "all") + session = server._deferred_session_record( + "session-key", + cols=80, + cwd="/tmp", + history=[], + lease=None, + ) + assert "sid" not in session + assert "_sid" not in session + server._sessions["deferred-sid"] = session + + try: + server._schedule_idle_compression_scoped("deferred-sid", session) + assert len(FakeThread.created) == 1 + assert "deferred-sid" in server._IDLE_COMPRESSION_THREADS + finally: + server._sessions.pop("deferred-sid", None) + server._IDLE_COMPRESSION_THREADS.pop("deferred-sid", None) + + +@pytest.mark.parametrize("mode", ["queue", "steer", "interrupt"]) +def test_prompt_submit_queues_during_idle_compression_without_control_calls( + monkeypatch, mode +): + calls = [] + agent = types.SimpleNamespace( + steer=lambda text: calls.append(("steer", text)) or True, + interrupt=lambda: calls.append(("interrupt", None)), + ) + transport = object() + session = _session( + agent=agent, + idle_compression_running=True, + transport=transport, + ) + server._sessions["busy-sid"] = session + monkeypatch.setattr(server, "_load_busy_input_mode", lambda: mode) + try: + response = server.handle_request( + { + "id": "1", + "method": "prompt.submit", + "params": {"session_id": "busy-sid", "text": "do not lose me"}, + } + ) + assert response["result"]["status"] == "queued" + assert session["queued_prompt"] == { + "text": "do not lose me", + "transport": transport, + } + assert session["running"] is False + assert "idle_compression_cancel_requested" not in session + assert calls == [] + finally: + server._sessions.pop("busy-sid", None) + + +def test_two_prompts_merge_then_compression_commits_syncs_and_drains_once(monkeypatch): + started = threading.Event() + release = threading.Event() + original = _history() + sequence = [] + + class BlockingAgent: + session_id = "old-key" + context_compressor = types.SimpleNamespace(context_length=1000) + + def _compress_context(self, history, system_message, **_kwargs): + assert session["history_lock"].acquire(blocking=False) + session["history_lock"].release() + started.set() + assert release.wait(timeout=5) + self.session_id = "continuation-key" + return history[:2], {} + + session = _session( + sid="commit-sid", + agent=BlockingAgent(), + session_key="old-key", + history=list(original), + ) + server._sessions["commit-sid"] = session + _configure_ready_idle_pass(monkeypatch) + + def sync(sid, current, **_kwargs): + assert sid == "commit-sid" + assert current["history"] == original[:2] + assert current["history_version"] == 1 + assert current["history_lock"].acquire(blocking=False) + current["history_lock"].release() + current["session_key"] = current["agent"].session_id + sequence.append("sync") + + def drain(_rid, sid, current): + assert sid == "commit-sid" + assert current["idle_compression_running"] is False + assert current["session_key"] == "continuation-key" + assert current["history"] == original[:2] + assert current["history_lock"].acquire(blocking=False) + current["history_lock"].release() + queued = current.pop("queued_prompt", None) + sequence.append(("drain", queued)) + return queued is not None + + monkeypatch.setattr(server, "_sync_session_key_after_compress", sync) + monkeypatch.setattr(server, "_drain_queued_prompt", drain) + result = [] + worker = threading.Thread( + target=lambda: result.append( + server._run_idle_compression_once("commit-sid", session) + ) + ) + worker.start() + assert started.wait(timeout=5) + assert server._handle_busy_submit("1", "commit-sid", session, "first", "ws-1")[ + "result" + ]["status"] == "queued" + assert server._handle_busy_submit("2", "commit-sid", session, "second", "ws-2")[ + "result" + ]["status"] == "queued" + assert session["queued_prompt"] == { + "text": "first\n\nsecond", + "transport": "ws-2", + } + assert "idle_compression_cancel_requested" not in session + release.set() + worker.join(timeout=5) + try: + assert not worker.is_alive() + assert result == [True] + assert sequence == [ + "sync", + ( + "drain", + {"text": "first\n\nsecond", "transport": "ws-2"}, + ), + ] + assert session["history"] == original[:2] + assert session["history_version"] == 1 + assert session["session_key"] == session["agent"].session_id + finally: + release.set() + server._sessions.pop("commit-sid", None) + + +@pytest.mark.parametrize("outcome", ["success", "noop", "exception"]) +def test_idle_cleanup_releases_semaphore_and_drains_live_queue(monkeypatch, outcome): + semaphore = threading.BoundedSemaphore(1) + monkeypatch.setattr(server, "_IDLE_COMPRESSION_GLOBAL_SEMAPHORE", semaphore) + _configure_ready_idle_pass(monkeypatch) + session = _session( + sid="cleanup-sid", + agent=types.SimpleNamespace(session_id="session-key"), + history=_history(), + queued_prompt={"text": "next", "transport": None}, + ) + server._sessions["cleanup-sid"] = session + drained = [] + + def compress(*_args, **_kwargs): + if outcome == "exception": + raise RuntimeError("compress failed") + return (2 if outcome == "success" else 0), {} + + def drain(_rid, _sid, current): + assert current["idle_compression_running"] is False + drained.append(current["queued_prompt"]["text"]) + return True + + monkeypatch.setattr(server, "_compress_session_history", compress) + monkeypatch.setattr(server, "_drain_queued_prompt", drain) + try: + assert server._run_idle_compression_once("cleanup-sid", session) is ( + outcome == "success" + ) + assert session["idle_compression_running"] is False + assert drained == ["next"] + assert semaphore.acquire(blocking=False) + semaphore.release() + finally: + server._sessions.pop("cleanup-sid", None) + + +@pytest.mark.parametrize( + ("fence", "expect_drain"), + [ + ("history_version", True), + ("normal_turn", False), + ("finalized", False), + ("replacement", False), + ], +) +def test_idle_result_fences_stale_or_non_live_session( + monkeypatch, fence, expect_drain +): + started = threading.Event() + release = threading.Event() + original = _history() + + class BlockingAgent: + session_id = "session-key" + context_compressor = types.SimpleNamespace(context_length=1000) + + def _compress_context(self, history, system_message, **_kwargs): + started.set() + assert release.wait(timeout=5) + return history[:2], {} + + session = _session( + sid="fence-sid", + agent=BlockingAgent(), + history=list(original), + history_version=7, + queued_prompt={"text": "queued", "transport": None}, + ) + server._sessions["fence-sid"] = session + _configure_ready_idle_pass(monkeypatch) + synced = [] + drained = [] + monkeypatch.setattr( + server, "_sync_session_key_after_compress", lambda *_a, **_k: synced.append(True) + ) + monkeypatch.setattr( + server, "_drain_queued_prompt", lambda *_a, **_k: drained.append(True) or True + ) + worker = threading.Thread( + target=server._run_idle_compression_once, + args=("fence-sid", session), + ) + worker.start() + assert started.wait(timeout=5) + with session["history_lock"]: + if fence == "history_version": + session["history_version"] += 1 + elif fence == "normal_turn": + session["running"] = True + elif fence == "finalized": + session["_finalized"] = True + if fence == "replacement": + server._sessions["fence-sid"] = _session(sid="fence-sid") + release.set() + worker.join(timeout=5) + try: + assert not worker.is_alive() + assert session["history"] == original + assert session["history_version"] == (8 if fence == "history_version" else 7) + assert synced == [] + assert drained == ([True] if expect_drain else []) + assert session["idle_compression_running"] is False + finally: + release.set() + server._sessions.pop("fence-sid", None) + + +def test_replaced_registry_record_without_sid_cannot_publish_or_drain(monkeypatch): + started = threading.Event() + release = threading.Event() + original = _history() + + class BlockingAgent: + session_id = "session-key" + context_compressor = types.SimpleNamespace(context_length=1000) + + def _compress_context(self, history, system_message, **_kwargs): + started.set() + assert release.wait(timeout=5) + return history[:2], {} + + session = _session( + agent=BlockingAgent(), + history=list(original), + queued_prompt={"text": "never drain", "transport": None}, + ) + assert "sid" not in session + assert "_sid" not in session + server._sessions["replacement-sid"] = session + _configure_ready_idle_pass(monkeypatch) + synced = [] + drained = [] + monkeypatch.setattr( + server, "_sync_session_key_after_compress", lambda *_a, **_k: synced.append(True) + ) + monkeypatch.setattr( + server, "_drain_queued_prompt", lambda *_a, **_k: drained.append(True) or True + ) + worker = threading.Thread( + target=server._run_idle_compression_once, + args=("replacement-sid", session), + ) + worker.start() + assert started.wait(timeout=5) + server._sessions["replacement-sid"] = _session() + live_after_replacement = server._session_has_live_identity( + "replacement-sid", session + ) + release.set() + worker.join(timeout=5) + + try: + assert not worker.is_alive() + assert live_after_replacement is False + assert session["history"] == original + assert synced == [] + assert drained == [] + finally: + release.set() + server._sessions.pop("replacement-sid", None) + + +def test_finalize_during_compression_prevents_commit_sync_and_drain(monkeypatch): + started = threading.Event() + release = threading.Event() + original = _history() + + class BlockingAgent: + session_id = "session-key" + context_compressor = types.SimpleNamespace(context_length=1000) + + def _compress_context(self, history, system_message, **_kwargs): + started.set() + assert release.wait(timeout=5) + return history[:2], {} + + session = _session( + sid="finalize-sid", + agent=BlockingAgent(), + history=list(original), + queued_prompt={"text": "never run", "transport": None}, + ) + server._sessions["finalize-sid"] = session + _configure_ready_idle_pass(monkeypatch) + synced = [] + drained = [] + monkeypatch.setattr(server, "_release_active_session_slot", lambda *_a: None) + monkeypatch.setattr(server, "_notify_session_boundary", lambda *_a, **_k: None) + monkeypatch.setattr(server, "_get_db", lambda: None) + monkeypatch.setattr( + server, "_sync_session_key_after_compress", lambda *_a, **_k: synced.append(True) + ) + monkeypatch.setattr( + server, "_drain_queued_prompt", lambda *_a, **_k: drained.append(True) or True + ) + worker = threading.Thread( + target=server._run_idle_compression_once, + args=("finalize-sid", session), + ) + worker.start() + assert started.wait(timeout=5) + server._finalize_session(session) + release.set() + worker.join(timeout=5) + try: + assert not worker.is_alive() + assert session["_finalized"] is True + assert session["history"] == original + assert session["history_version"] == 0 + assert synced == [] + assert drained == [] + finally: + release.set() + server._sessions.pop("finalize-sid", None) + + +def test_session_key_sync_finalize_during_transfer_aborts_without_side_effects( + monkeypatch, +): + sid = "sync-finalize-sid" + entered = threading.Event() + release = threading.Event() + finalized_marked = threading.Event() + sync_lock = _ContendedLock() + lease = _SyncLease() + restart_calls = [] + approval_calls = _spy_session_key_approval_effects(monkeypatch) + + class FinalizeObservedSession(dict): + def __setitem__(self, key, value): + super().__setitem__(key, value) + if key == "_finalized" and value is True: + finalized_marked.set() + + session = FinalizeObservedSession( + _session( + sid=sid, + agent=types.SimpleNamespace( + session_id="new-key", model="test-model", platform="tui" + ), + session_key="old-key", + active_session_lease=lease, + _session_key_sync_lock=sync_lock, + ) + ) + server._sessions[sid] = session + + def transfer(_sid, current, *, new_session_id): + assert current is session + assert new_session_id == "new-key" + entered.set() + assert release.wait(timeout=5) + return True + + monkeypatch.setattr(server, "_transfer_active_session_slot", transfer) + monkeypatch.setattr( + server, + "_restart_slash_worker", + lambda *_args, **_kwargs: restart_calls.append(True), + ) + monkeypatch.setattr(server, "_notify_session_boundary", lambda *_a, **_k: None) + monkeypatch.setattr(server, "_get_db", lambda: None) + monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_k: None) + monkeypatch.setattr( + "tools.async_delegation.interrupt_for_session", lambda *_a, **_k: None + ) + + sync_errors = [] + finalize_errors = [] + + def run_sync(): + try: + server._sync_session_key_after_compress(sid, session) + except BaseException as exc: + sync_errors.append(exc) + + def run_finalize(): + try: + server._finalize_session(session) + except BaseException as exc: + finalize_errors.append(exc) + + sync_thread = threading.Thread(target=run_sync) + finalize_thread = threading.Thread(target=run_finalize) + contended = False + try: + sync_thread.start() + assert entered.wait(timeout=5) + finalize_thread.start() + assert finalized_marked.wait(timeout=5) + assert session["_finalized"] is True + contended = sync_lock.contended.wait(timeout=5) + finally: + release.set() + sync_thread.join(timeout=5) + finalize_thread.join(timeout=5) + server._sessions.pop(sid, None) + + assert contended is True + assert not sync_thread.is_alive() + assert not finalize_thread.is_alive() + assert sync_errors == [] + assert finalize_errors == [] + assert session["session_key"] == "old-key" + assert restart_calls == [] + assert approval_calls == [] + assert lease.release_calls == 1 + assert "active_session_lease" not in session + + +def test_session_key_sync_replacement_during_transfer_aborts_without_side_effects( + monkeypatch, +): + sid = "sync-replacement-sid" + entered = threading.Event() + release = threading.Event() + lease = _SyncLease() + restart_calls = [] + approval_calls = _spy_session_key_approval_effects(monkeypatch) + session = _session( + sid=sid, + agent=types.SimpleNamespace(session_id="new-key"), + session_key="old-key", + active_session_lease=lease, + ) + replacement = _session(sid=sid) + server._sessions[sid] = session + + def transfer(_sid, current, *, new_session_id): + assert current is session + assert new_session_id == "new-key" + entered.set() + assert release.wait(timeout=5) + return True + + monkeypatch.setattr(server, "_transfer_active_session_slot", transfer) + monkeypatch.setattr( + server, + "_restart_slash_worker", + lambda *_args, **_kwargs: restart_calls.append(True), + ) + errors = [] + + def run_sync(): + try: + server._sync_session_key_after_compress(sid, session) + except BaseException as exc: + errors.append(exc) + + worker = threading.Thread(target=run_sync) + try: + worker.start() + assert entered.wait(timeout=5) + with server._sessions_lock: + server._sessions[sid] = replacement + release.set() + worker.join(timeout=5) + finally: + release.set() + worker.join(timeout=5) + server._sessions.pop(sid, None) + + assert not worker.is_alive() + assert errors == [] + assert session["session_key"] == "old-key" + assert restart_calls == [] + assert approval_calls == [] + assert lease.release_calls == 1 + assert "active_session_lease" not in session + + +def test_session_key_sync_live_session_preserves_transfer_approval_and_worker( + monkeypatch, +): + sid = "sync-live-sid" + lease = _SyncLease() + transfer_calls = [] + restart_calls = [] + approval_calls = _spy_session_key_approval_effects(monkeypatch) + session = _session( + sid=sid, + agent=types.SimpleNamespace(session_id="new-key"), + session_key="old-key", + pending_title="continuation title", + active_session_lease=lease, + ) + server._sessions[sid] = session + + def transfer(_sid, current, *, new_session_id): + transfer_calls.append((_sid, current, new_session_id)) + return True + + monkeypatch.setattr(server, "_transfer_active_session_slot", transfer) + monkeypatch.setattr( + server, + "_restart_slash_worker", + lambda _sid, current: restart_calls.append( + (_sid, current["session_key"]) + ), + ) + try: + server._sync_session_key_after_compress(sid, session) + finally: + server._sessions.pop(sid, None) + + assert transfer_calls == [(sid, session, "new-key")] + assert session["session_key"] == "new-key" + assert session["pending_title"] is None + assert ("enable_yolo", "new-key") in approval_calls + assert ("disable_yolo", "old-key") in approval_calls + assert ("register_notify", "new-key") in approval_calls + assert ("unregister_notify", "old-key") in approval_calls + assert restart_calls == [(sid, "new-key")] + assert lease.release_calls == 0 + + +def test_finalize_popped_session_uses_teardown_sid_to_stop_idle_worker(monkeypatch): + sid = "popped-sid" + session = _session() + session["agent"] = None + session["session_key"] = "" + server._sessions[sid] = session + popped = server._pop_session_by_id(sid) + assert popped is session + assert session.get("_sid") == sid + assert "sid" not in session + assert "tui_session_id" not in session + + stop_event = threading.Event() + server._IDLE_COMPRESSION_THREADS[sid] = (object(), stop_event) + monkeypatch.setattr(server, "_release_active_session_slot", lambda *_a: None) + monkeypatch.setattr(server, "_notify_session_boundary", lambda *_a, **_k: None) + + try: + server._finalize_session(session) + assert stop_event.is_set() + assert sid not in server._IDLE_COMPRESSION_THREADS + finally: + server._sessions.pop(sid, None) + server._IDLE_COMPRESSION_THREADS.pop(sid, None) + + +def test_context_override_resets_and_semaphore_releases_after_exception( + monkeypatch, tmp_path +): + named_home = tmp_path / "profiles" / "broken" + _write_idle_config(named_home, enabled=True) + monkeypatch.setattr( + "agent.model_metadata.estimate_request_tokens_rough", + lambda messages, system_prompt="", tools=None: 600, + ) + semaphore = threading.BoundedSemaphore(1) + monkeypatch.setattr(server, "_IDLE_COMPRESSION_GLOBAL_SEMAPHORE", semaphore) + seen = [] + + def fail(*_args, **_kwargs): + seen.append(get_hermes_home()) + raise RuntimeError("boom") + + monkeypatch.setattr(server, "_compress_session_history", fail) + session = _session( + sid="exception-sid", + agent=types.SimpleNamespace( + session_id="session-key", + context_compressor=types.SimpleNamespace(context_length=1000), + ), + history=_history(), + profile_home=str(named_home), + ) + server._sessions["exception-sid"] = session + parent_home = get_hermes_home() + thread_scope = {} + + def worker_call(): + thread_scope["before"] = get_hermes_home() + result = server._run_idle_compression_once("exception-sid", session) + thread_scope["after"] = get_hermes_home() + return result + + try: + assert _run_in_real_thread(worker_call) is False + assert seen == [named_home] + assert thread_scope == {"before": parent_home, "after": parent_home} + assert session["idle_compression_running"] is False + assert semaphore.acquire(blocking=False) + semaphore.release() + finally: + server._sessions.pop("exception-sid", None) + + +def test_session_steer_never_targets_idle_compactor(monkeypatch): + calls = [] + session = _session( + agent=types.SimpleNamespace(steer=lambda text: calls.append(text) or True), + idle_compression_running=True, + ) + server._sessions["steer-sid"] = session + try: + response = server.handle_request( + { + "id": "1", + "method": "session.steer", + "params": {"session_id": "steer-sid", "text": "not the compactor"}, + } + ) + assert "error" in response + assert calls == [] + finally: + server._sessions.pop("steer-sid", None) + + +def test_slash_mutation_is_blocked_during_idle_compression(monkeypatch): + calls = [] + session = _session( + agent=types.SimpleNamespace(), + idle_compression_running=True, + ) + monkeypatch.setattr( + server, + "_apply_model_switch", + lambda *_a, **_k: calls.append(True) or {}, + ) + result = server._mirror_slash_side_effects("slash-sid", session, "/model x") + assert "session busy" in result + assert calls == [] + + +def test_idle_compression_skips_agentless_session(monkeypatch): + _configure_ready_idle_pass(monkeypatch) + session = _session(agent=None, history=_history()) + session["agent"] = None + assert server._run_idle_compression_once("agentless-sid", session) is False + assert session.get("idle_compression_running") is not True diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 786587b07905..93bb22047051 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -5495,6 +5495,222 @@ def test_session_compress_uses_compress_helper(monkeypatch): emit.assert_any_call("status.update", "sid", {"kind": "status", "text": "ready"}) +def test_idle_compression_config_defaults_disabled(monkeypatch): + monkeypatch.setattr( + server, + "_load_cfg", + lambda: {"compression": {"enabled": True, "threshold": 0.6}}, + ) + + cfg = server._load_idle_compression_config() + + assert cfg["enabled"] is False + assert cfg["threshold"] == 0.54 + assert cfg["idle_after_seconds"] == 120.0 + assert cfg["min_interval_seconds"] == 1800.0 + + +def test_idle_compression_config_respects_parent_enabled(monkeypatch): + monkeypatch.setattr( + server, + "_load_cfg", + lambda: { + "compression": { + "enabled": False, + "threshold": 0.6, + "idle": {"enabled": True}, + } + }, + ) + + cfg = server._load_idle_compression_config() + + assert cfg["enabled"] is False + assert cfg["threshold"] == 0.54 + + +def test_idle_compression_config_rejects_non_finite_values(monkeypatch): + monkeypatch.setattr( + server, + "_load_cfg", + lambda: { + "compression": { + "enabled": True, + "threshold": 0.6, + "idle": { + "enabled": "yes", + "threshold": "nan", + "idle_after_seconds": "inf", + "min_interval_seconds": -1, + }, + } + }, + ) + + cfg = server._load_idle_compression_config() + + assert cfg["enabled"] is True + assert cfg["threshold"] == 0.54 + assert cfg["idle_after_seconds"] == 120.0 + assert cfg["min_interval_seconds"] == 60.0 + + +def test_session_info_reports_idle_compression_as_running(monkeypatch): + session = _session(idle_compression_running=True) + monkeypatch.setattr(server, "_load_cfg", lambda: {}) + monkeypatch.setattr(server, "_display_session_cwd", lambda _session: "/tmp") + monkeypatch.setattr(server, "_git_branch_for_cwd", lambda _cwd: "") + monkeypatch.setattr(server, "_project_info_for_cwd", lambda _cwd: None) + monkeypatch.setattr(server, "_session_usage_snapshot", lambda _session: {}) + + info = server._session_info(session["agent"], session) + + assert info["running"] is True + + +def test_finalize_session_cancels_pending_idle_compression(monkeypatch): + stop_event = threading.Event() + fake_thread = threading.Thread(target=lambda: None) + session = _session(agent=types.SimpleNamespace(session_id="session-key"), sid="sid") + server._IDLE_COMPRESSION_THREADS["sid"] = (fake_thread, stop_event) + monkeypatch.setattr(server, "_notify_session_boundary", lambda *args, **kwargs: None) + + try: + server._finalize_session(session) + + assert stop_event.is_set() + assert "sid" not in server._IDLE_COMPRESSION_THREADS + finally: + server._IDLE_COMPRESSION_THREADS.pop("sid", None) + + +class _IdleCompressAgent: + session_id = "session-key" + + def _compress_context(self, history, system_message, approx_tokens=None, focus_topic=None): + return history[:2], {} + + +def test_idle_compression_drops_result_if_prompt_started_during_compress(): + history = [ + {"role": "user", "content": "u1"}, + {"role": "assistant", "content": "a1"}, + {"role": "user", "content": "u2"}, + {"role": "assistant", "content": "a2"}, + ] + session = _session(agent=_IdleCompressAgent(), history=history, history_version=3) + original = list(history) + + def racing_compress(*args, **kwargs): + session["running"] = True + return original[:2], {} + + session["agent"]._compress_context = racing_compress + removed, _usage = server._compress_session_history( + session, + approx_tokens=600, + before_messages=original, + history_version=3, + abort_if_running=True, + ) + + assert removed == 0 + assert session["history"] == original + assert session["history_version"] == 3 + + +def test_idle_compression_attempt_cooldown_after_noop(monkeypatch): + history = [ + {"role": "user", "content": "u1"}, + {"role": "assistant", "content": "a1"}, + {"role": "user", "content": "u2"}, + {"role": "assistant", "content": "a2"}, + ] + session = _session( + agent=types.SimpleNamespace(session_id="session-key"), + history=history, + last_active=0.0, + ) + now = {"value": 250.0} + calls = [] + monkeypatch.setattr(server.time, "time", lambda: now["value"]) + monkeypatch.setattr( + server, + "_load_idle_compression_config", + lambda: { + "enabled": True, + "threshold": 0.5, + "idle_after_seconds": 120.0, + "min_interval_seconds": 1800.0, + "emit_status": False, + }, + ) + monkeypatch.setattr( + "agent.model_metadata.estimate_request_tokens_rough", + lambda messages, system_prompt="", tools=None: 600, + ) + monkeypatch.setattr( + server, "_session_compression_threshold_tokens", lambda _agent, _cfg: 500 + ) + monkeypatch.setattr( + server, + "_compress_session_history", + lambda *args, **kwargs: calls.append(kwargs) or (0, {}), + ) + + server._sessions["sid"] = session + try: + assert server._run_idle_compression_once("sid", session) is False + assert session["last_idle_compression_attempt_at"] == 250.0 + now["value"] = 300.0 + assert server._run_idle_compression_once("sid", session) is False + assert len(calls) == 1 + finally: + server._sessions.pop("sid", None) + + +def test_idle_compression_schedule_debounces_to_latest_activity(monkeypatch): + class FakeThread: + created = [] + + def __init__(self, target, **kwargs): + self.target = target + self._alive = False + self.__class__.created.append(self) + + def start(self): + self._alive = True + + def is_alive(self): + return self._alive + + now = {"value": 100.0} + session = _session(last_active=100.0, sid="sid") + monkeypatch.setattr( + server, + "_load_idle_compression_config", + lambda: { + "enabled": True, + "threshold": 0.5, + "idle_after_seconds": 120.0, + "min_interval_seconds": 1800.0, + "emit_status": False, + }, + ) + monkeypatch.setattr(server.time, "time", lambda: now["value"]) + monkeypatch.setattr(server.threading, "Thread", FakeThread) + server._IDLE_COMPRESSION_THREADS.clear() + + server._schedule_idle_compression("sid", session) + assert session["idle_compression_due_at"] == 220.0 + now["value"] = 150.0 + session["last_active"] = 150.0 + server._schedule_idle_compression("sid", session) + + assert len(FakeThread.created) == 1 + assert session["idle_compression_due_at"] == 270.0 + + def test_session_compress_reports_aborted_summary_without_success(monkeypatch): compression_state = types.SimpleNamespace( _last_compress_aborted=True, diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 801f855910ca..fc463fbebd1a 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -6,6 +6,7 @@ import inspect import json import logging +import math import os import queue import subprocess @@ -262,6 +263,9 @@ def _thread_panic_hook(args): max_workers=_rpc_pool_workers, thread_name_prefix="tui-rpc", ) +_IDLE_COMPRESSION_THREADS: dict[str, tuple[threading.Thread, threading.Event]] = {} +_IDLE_COMPRESSION_LOCK = threading.Lock() +_IDLE_COMPRESSION_GLOBAL_SEMAPHORE = threading.BoundedSemaphore(1) atexit.register(lambda: _pool.shutdown(wait=False, cancel_futures=True)) # Reserve real stdout for JSON-RPC only; redirect Python's stdout to stderr @@ -472,6 +476,59 @@ def _release_active_session_slot(session: dict | None) -> None: logger.debug("Failed to release active session slot", exc_info=True) +def _session_key_sync_lock(session: dict): + """Return the per-session lock serializing key sync with finalization.""" + lock = session.get("_session_key_sync_lock") + if lock is not None: + return lock + # CPython's dict.setdefault is atomic under the GIL, so concurrent lazy + # creators still converge on the same lock stored on this session record. + return session.setdefault("_session_key_sync_lock", threading.Lock()) + + +def _session_key_sync_is_live(sid: str, session: dict) -> bool: + """Atomically check registry identity and the history finalize fence.""" + history_lock = session.get("history_lock") + with _sessions_lock: + if not _session_has_live_identity(sid, session): + return False + if history_lock is not None: + with history_lock: + return not bool(session.get("_finalized")) + return not bool(session.get("_finalized")) + + +def _commit_session_key_sync( + sid: str, + session: dict, + new_session_id: str, + *, + clear_pending_title: bool, +) -> bool: + """Publish a rotated key iff identity and finalize fences are still live. + + This is deliberately field-only while holding the registry/history locks; + lease, approval, YOLO, and worker operations stay outside both locks. + """ + + def _commit_fields() -> bool: + if session.get("_finalized"): + return False + session["session_key"] = new_session_id + if clear_pending_title: + session["pending_title"] = None + return True + + history_lock = session.get("history_lock") + with _sessions_lock: + if not _session_has_live_identity(sid, session): + return False + if history_lock is not None: + with history_lock: + return _commit_fields() + return _commit_fields() + + def _transfer_active_session_slot( sid: str, session: dict, @@ -571,16 +628,46 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No force-quit (double Ctrl‑C, terminal‑close, SIGHUP) while the agent is mid‑turn. """ - if not session or session.get("_finalized"): + if not session: return - session["_finalized"] = True + sync_lock = _session_key_sync_lock(session) + lock = session.get("history_lock") + if lock is not None: + with lock: + if session.get("_finalized"): + return + # Fence any idle compressor before doing persistence, hooks, or + # worker cleanup. The compressor checks this under the same lock + # before publishing its snapshot. + session["_finalized"] = True + else: + if session.get("_finalized"): + return + session["_finalized"] = True + # Do not hold the history or registry lock while waiting. A sync already + # inside lease transfer must be able to return, observe the finalize fence, + # compensate its lease, and release this per-session handoff lock. + with sync_lock: + pass _release_active_session_slot(session) stop_event = session.get("_notif_stop") if stop_event is not None: stop_event.set() + sid = ( + session.get("_sid") + or session.get("sid") + or session.get("tui_session_id") + or "" + ) + if sid: + with _IDLE_COMPRESSION_LOCK: + worker = _IDLE_COMPRESSION_THREADS.pop(sid, None) + if worker: + _thread, idle_stop = worker + idle_stop.set() + agent = session.get("agent") - lock = session.get("history_lock") if lock is not None: with lock: history = list(session.get("history", [])) @@ -802,7 +889,7 @@ def _ws_session_is_orphaned(session: dict | None) -> bool: """ if not session or session.get("_finalized"): return False - if session.get("running"): + if _session_busy(session): return False return session.get("transport") is _detached_ws_transport @@ -905,7 +992,7 @@ def _transport_is_dead(transport) -> bool: def _session_is_evictable(sid: str, session: dict, now: float) -> bool: - if session.get("running") or _session_pending_kind(sid): + if _session_busy(session) or _session_pending_kind(sid): return False ready = session.get("agent_ready") # Lazy watch sessions (subagent spectator windows) never start a build, @@ -956,7 +1043,7 @@ def _session_is_lru_evictable(sid: str, session: dict) -> bool: # Same hard exemptions as the TTL reaper (never evict a session mid-turn, # awaiting input, or still building), but WITHOUT the hours-scale age gate: # a detached session is eligible the moment it loses its client. - if session.get("running") or _session_pending_kind(sid): + if _session_busy(session) or _session_pending_kind(sid): return False ready = session.get("agent_ready") if ready is not None and not ready.is_set() and not session.get("lazy"): @@ -3445,6 +3532,8 @@ def _compress_session_history( approx_tokens: int | None = None, before_messages: list | None = None, history_version: int | None = None, + abort_if_running: bool = False, + expected_sid: str | None = None, ) -> tuple[int, dict]: from agent.model_metadata import estimate_request_tokens_rough @@ -3481,6 +3570,15 @@ def _compress_session_history( focus_topic=focus_topic or None, ) with session["history_lock"]: + if session.get("_finalized") or ( + expected_sid is not None + and not _session_has_live_identity(expected_sid, session) + ): + usage = _get_usage(agent) + return 0, usage + if abort_if_running and session.get("running"): + usage = _get_usage(agent) + return 0, usage if int(session.get("history_version", 0)) != history_version: # External mutation during compaction — drop the compressed # result so we don't clobber concurrent edits. @@ -3492,6 +3590,300 @@ def _compress_session_history( return len(history) - len(compressed), usage +def _session_busy(session: dict | None) -> bool: + session = session or {} + return bool(session.get("running") or session.get("idle_compression_running")) + + +def _session_has_live_identity(sid: str, session: dict) -> bool: + """Return whether a registry-backed session is still this exact object. + + The live registry object is authoritative because create/deferred records + intentionally do not carry a long-lived sid. Teardown markers fence + detached real sessions, while marker-free utility sessions remain usable + outside the registry in focused tests and helpers. + """ + registered = _sessions.get(sid) + if registered is session: + return True + if registered is not None: + return False + if session.get("sid") or session.get("_sid"): + return False + return True + + +def _coerce_float(value: Any, default: float) -> float: + try: + coerced = float(value) + except (TypeError, ValueError): + return default + return coerced if math.isfinite(coerced) else default + + +def _coerce_bool(value: Any, default: bool = False) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + if isinstance(value, str): + return value.strip().lower() in {"1", "true", "yes", "on", "y"} + return default + + +def _load_idle_compression_config() -> dict: + """Read conservative Gateway idle pre-compression settings.""" + cfg = _load_cfg().get("compression") or {} + if not isinstance(cfg, dict): + cfg = {} + parent_enabled = _coerce_bool(cfg.get("enabled"), True) + idle = cfg.get("idle") or cfg.get("idle_precompression") or {} + if not isinstance(idle, dict): + idle = {} + base_threshold = _coerce_float(cfg.get("threshold"), 0.5) + default_threshold = max(0.1, base_threshold * 0.9) + threshold = _coerce_float(idle.get("threshold"), default_threshold) + threshold = max(0.1, min(threshold, 0.95)) + idle_after = _coerce_float( + idle.get("idle_after_seconds", idle.get("after_seconds", 120.0)), 120.0 + ) + min_interval = _coerce_float( + idle.get("min_interval_seconds", idle.get("cooldown_seconds", 1800.0)), + 1800.0, + ) + return { + "enabled": parent_enabled and _coerce_bool(idle.get("enabled"), False), + "threshold": threshold, + "idle_after_seconds": min(3600.0, max(10.0, idle_after)), + "min_interval_seconds": min(86400.0, max(60.0, min_interval)), + "emit_status": _coerce_bool(idle.get("emit_status"), False), + } + + +def _session_compression_threshold_tokens(agent: Any, cfg: dict) -> int: + compressor = getattr(agent, "context_compressor", None) + context_length = getattr(compressor, "context_length", None) + if not context_length: + try: + from agent.model_metadata import get_model_context_length + + context_length = get_model_context_length( + getattr(agent, "model", "") or _resolve_model(), + base_url=getattr(agent, "base_url", "") or "", + api_key=getattr(agent, "api_key", "") or "", + provider=getattr(agent, "provider", "") or "", + config_context_length=getattr(agent, "_config_context_length", None), + ) + except Exception: + context_length = 0 + if not context_length: + return 0 + return int(context_length * float(cfg.get("threshold", 0.5))) + + +def _run_idle_compression_once(sid: str, session: dict) -> bool: + home_token = set_hermes_home_override(session.get("profile_home")) + try: + return _run_idle_compression_once_scoped(sid, session) + finally: + reset_hermes_home_override(home_token) + + +def _run_idle_compression_once_scoped(sid: str, session: dict) -> bool: + """Attempt one safe idle compression pass for a Gateway session.""" + cfg = _load_idle_compression_config() + if not cfg.get("enabled"): + return False + now = time.time() + with session["history_lock"]: + if ( + _session_busy(session) + or session.get("_finalized") + or not _session_has_live_identity(sid, session) + ): + return False + last_active = float(session.get("last_active") or 0.0) + if now - last_active < float(cfg["idle_after_seconds"]): + return False + last_attempt = float( + session.get("last_idle_compression_attempt_at") + or session.get("last_idle_compression_at") + or 0.0 + ) + if last_attempt and now - last_attempt < float(cfg["min_interval_seconds"]): + return False + history = list(session.get("history", [])) + history_version = int(session.get("history_version", 0)) + if len(history) < 4: + return False + agent = session.get("agent") + if agent is None: + return False + try: + from agent.model_metadata import estimate_request_tokens_rough + + approx_tokens = estimate_request_tokens_rough( + history, + system_prompt=getattr(agent, "_cached_system_prompt", "") or "", + tools=getattr(agent, "tools", None) or None, + ) + except Exception as exc: + logger.debug("idle compression token estimate skipped for %s: %s", sid, exc) + return False + threshold_tokens = _session_compression_threshold_tokens(agent, cfg) + if not threshold_tokens or approx_tokens < threshold_tokens: + return False + if not _IDLE_COMPRESSION_GLOBAL_SEMAPHORE.acquire(blocking=False): + with session["history_lock"]: + if not session.get("_finalized") and not _session_busy(session): + session["idle_compression_due_at"] = time.time() + min( + 30.0, float(cfg["idle_after_seconds"]) + ) + return False + try: + with session["history_lock"]: + if ( + _session_busy(session) + or session.get("_finalized") + or not _session_has_live_identity(sid, session) + or int(session.get("history_version", 0)) != history_version + ): + return False + session["idle_compression_running"] = True + session["last_idle_compression_attempt_at"] = now + if cfg.get("emit_status"): + _emit( + "status.update", + sid, + {"kind": "status", "text": f"idle compacting ~{approx_tokens:,} tokens…"}, + ) + try: + removed, _usage = _compress_session_history( + session, + approx_tokens=approx_tokens, + before_messages=history, + history_version=history_version, + abort_if_running=True, + expected_sid=sid, + ) + if removed > 0: + with session["history_lock"]: + can_publish = bool( + not session.get("_finalized") + and not session.get("running") + and _session_has_live_identity(sid, session) + ) + if not can_publish: + return False + _sync_session_key_after_compress( + sid, + session, + clear_pending_title=False, + restart_slash_worker=True, + ) + with session["history_lock"]: + session["last_idle_compression_at"] = now + logger.info( + "idle compression completed: sid=%s approx_tokens=%s threshold=%s removed=%s", + sid, + approx_tokens, + threshold_tokens, + removed, + ) + return True + return False + finally: + with session["history_lock"]: + session["idle_compression_running"] = False + is_live = bool( + not session.get("_finalized") + and _session_has_live_identity(sid, session) + ) + should_drain = bool(is_live and not session.get("running")) + if cfg.get("emit_status") and is_live: + _emit("status.update", sid, {"kind": "status", "text": "ready"}) + if should_drain: + _drain_queued_prompt("__idle_compression__", sid, session) + except Exception as exc: + logger.warning("idle compression failed for sid=%s: %s", sid, exc) + return False + finally: + _IDLE_COMPRESSION_GLOBAL_SEMAPHORE.release() + + +def _schedule_idle_compression(sid: str, session: dict) -> None: + home_token = set_hermes_home_override(session.get("profile_home")) + try: + _schedule_idle_compression_scoped(sid, session) + finally: + reset_hermes_home_override(home_token) + + +def _schedule_idle_compression_scoped(sid: str, session: dict) -> None: + # Real create/deferred records are identified by the live registry object + # and intentionally do not carry a long-lived sid. Preserve the original + # PR's explicit-sid unit-test helper, but do not start daemon workers for + # unrelated marker-free synthetic sessions outside the registry. + if _sessions.get(sid) is not session and session.get("sid") != sid: + return + cfg = _load_idle_compression_config() + if not cfg.get("enabled"): + return + due_at = time.time() + float(cfg["idle_after_seconds"]) + with session["history_lock"]: + if session.get("_finalized") or _session_busy(session): + return + session["idle_compression_due_at"] = due_at + stop_event = threading.Event() + + def _worker() -> None: + with _IDLE_COMPRESSION_LOCK: + cur = _IDLE_COMPRESSION_THREADS.get(sid) + if not cur or cur[0] is not threading.current_thread(): + # A replaced worker must not run. Synchronous Thread fakes + # also land here instead of blocking their caller in wait(). + if cur and cur[0] is thread: + _IDLE_COMPRESSION_THREADS.pop(sid, None) + return + try: + while True: + with session["history_lock"]: + if session.get("_finalized"): + return + target_due_at = float(session.get("idle_compression_due_at") or 0.0) + remaining = max(0.0, target_due_at - time.time()) + if remaining > 0 and stop_event.wait(timeout=remaining): + return + _run_idle_compression_once(sid, session) + with session["history_lock"]: + next_due_at = float(session.get("idle_compression_due_at") or 0.0) + finalized = bool(session.get("_finalized")) + if finalized or next_due_at <= time.time(): + return + finally: + with _IDLE_COMPRESSION_LOCK: + cur = _IDLE_COMPRESSION_THREADS.get(sid) + if cur and cur[0] is threading.current_thread(): + _IDLE_COMPRESSION_THREADS.pop(sid, None) + + with _IDLE_COMPRESSION_LOCK: + existing = _IDLE_COMPRESSION_THREADS.get(sid) + if existing and existing[0].is_alive(): + return + thread = threading.Thread( + target=_worker, + daemon=True, + name=f"idle-compress-{sid[:12] or 'session'}", + ) + _IDLE_COMPRESSION_THREADS[sid] = (thread, stop_event) + # Start outside the registry lock. Real threads may otherwise only block + # briefly, but synchronous test/embedding thread adapters would re-enter + # _worker while this non-reentrant lock is still held and deadlock. + thread.start() + + def _sync_session_key_after_compress( sid: str, session: dict, @@ -3516,69 +3908,152 @@ def _sync_session_key_after_compress( auto-compression (worker holds stale session key). False only if the caller manages the worker lifecycle separately. """ - agent = session.get("agent") - new_session_id = getattr(agent, "session_id", None) or "" - old_key = session.get("session_key", "") or "" - if not new_session_id or new_session_id == old_key: - return + sync_lock = _session_key_sync_lock(session) + with sync_lock: + if not _session_key_sync_is_live(sid, session): + return + agent = session.get("agent") + new_session_id = getattr(agent, "session_id", None) or "" + old_key = session.get("session_key", "") or "" + if not new_session_id or new_session_id == old_key: + return - lease_reanchored = _transfer_active_session_slot( - sid, - session, - new_session_id=new_session_id, - ) - if not lease_reanchored: - logger.warning( - "Compression session lease did not re-anchor: sid=%s old_session_id=%s new_session_id=%s", + lease_reanchored = _transfer_active_session_slot( sid, - old_key, - new_session_id, + session, + new_session_id=new_session_id, ) + if not lease_reanchored: + logger.warning( + "Compression session lease did not re-anchor: sid=%s old_session_id=%s new_session_id=%s", + sid, + old_key, + new_session_id, + ) - try: - from tools.approval import ( - disable_session_yolo, - enable_session_yolo, - is_session_yolo_enabled, - register_gateway_notify, - unregister_gateway_notify, - ) + # Transfer can block. Finalize deliberately marks its fence before it + # waits on sync_lock, and registry replacement does not need sync_lock, + # so both races must be re-checked before publishing any new identity. + if not _session_key_sync_is_live(sid, session): + _release_active_session_slot(session) + return + disable_session_yolo = None + enable_session_yolo = None + is_session_yolo_enabled = None + register_gateway_notify = None + unregister_gateway_notify = None try: - unregister_gateway_notify(old_key) + from tools.approval import ( + disable_session_yolo, + enable_session_yolo, + is_session_yolo_enabled, + register_gateway_notify, + unregister_gateway_notify, + ) except Exception: pass - session["session_key"] = new_session_id - try: - yolo_was_on = is_session_yolo_enabled(old_key) - except Exception: - yolo_was_on = False - if yolo_was_on: + + new_yolo_attempted = False + new_notify_attempted = False + + def _compensate_stale_sync() -> None: + if new_notify_attempted and unregister_gateway_notify is not None: + try: + unregister_gateway_notify(new_session_id) + except Exception: + pass + if new_yolo_attempted and disable_session_yolo is not None: + try: + disable_session_yolo(new_session_id) + except Exception: + pass + _release_active_session_slot(session) + + def _continue_if_live() -> bool: + if _session_key_sync_is_live(sid, session): + return True + _compensate_stale_sync() + return False + + yolo_was_on = False + if is_session_yolo_enabled is not None: + if not _continue_if_live(): + return + try: + yolo_was_on = bool(is_session_yolo_enabled(old_key)) + except Exception: + yolo_was_on = False + if not _continue_if_live(): + return + + if not _commit_session_key_sync( + sid, + session, + new_session_id, + clear_pending_title=clear_pending_title, + ): + _compensate_stale_sync() + return + if not _continue_if_live(): + return + + if yolo_was_on and enable_session_yolo is not None: + if not _continue_if_live(): + return + new_yolo_attempted = True try: enable_session_yolo(new_session_id) - disable_session_yolo(old_key) except Exception: pass - try: - register_gateway_notify( - new_session_id, - lambda data: _emit_approval_request(sid, data), - ) - except Exception: - pass - except Exception: - # Even if the approval module fails to import, still anchor the - # session_key on the new continuation id so downstream lookups - # don't keep targeting the ended row. - session["session_key"] = new_session_id + if not _continue_if_live(): + return - if clear_pending_title: - session["pending_title"] = None - if restart_slash_worker: - try: - _restart_slash_worker(sid, session) - except Exception: - pass + if disable_session_yolo is not None: + if not _continue_if_live(): + return + try: + disable_session_yolo(old_key) + except Exception: + pass + if not _continue_if_live(): + return + + if register_gateway_notify is not None: + if not _continue_if_live(): + return + new_notify_attempted = True + try: + register_gateway_notify( + new_session_id, + lambda data: _emit_approval_request(sid, data), + ) + except Exception: + pass + if not _continue_if_live(): + return + + if unregister_gateway_notify is not None: + if not _continue_if_live(): + return + try: + unregister_gateway_notify(old_key) + except Exception: + pass + if not _continue_if_live(): + return + + if restart_slash_worker: + # This is the final live fence before the external worker restart; + # _restart_slash_worker retains _attach_worker's identity guard. + if not _continue_if_live(): + return + try: + _restart_slash_worker(sid, session) + except Exception: + pass + if not _continue_if_live(): + return def _get_usage(agent) -> dict: @@ -3801,7 +4276,7 @@ def _session_info(agent, session: dict | None = None) -> dict: "branch": _git_branch_for_cwd(cwd), "project": _project_info_for_cwd(cwd), "personality": str(personality or ""), - "running": bool((session or {}).get("running")), + "running": _session_busy(session or {}), "title": _session_live_title(session or {}, session_key) if session_key else "", "stored_session_id": session_key or "", "desktop_contract": DESKTOP_BACKEND_CONTRACT, @@ -5139,6 +5614,7 @@ def _init_session( with _sessions_lock: _sessions[sid] = { "agent": agent, + "sid": sid, "session_key": key, "history": history, "history_lock": threading.Lock(), @@ -5628,13 +6104,20 @@ def _handle_busy_submit( without interrupting; ``steer`` → inject into the live turn if accepted, else queue. """ - mode = _load_busy_input_mode() agent = session.get("agent") with session["history_lock"]: + if session.get("idle_compression_running"): + # An idle compactor already owns the agent. Accept every busy + # input mode through the lossless queue, without steering or + # interrupting the compressor and without discarding its result. + _enqueue_prompt(session, text, transport) + session["last_active"] = time.time() + return _ok(rid, {"status": "queued"}) if not session.get("running"): # The turn ended between prompt.submit's first busy check and this # helper. Let the caller retry and claim the now-idle session. return None + mode = _load_busy_input_mode() if mode == "steer" and agent is not None and hasattr(agent, "steer"): try: if agent.steer(text): @@ -5647,6 +6130,10 @@ def _handle_busy_submit( # provider or compute-host method while holding history_lock: an interrupt # can wait behind the very operation it is trying to cancel. with session["history_lock"]: + if session.get("idle_compression_running"): + _enqueue_prompt(session, text, transport) + session["last_active"] = time.time() + return _ok(rid, {"status": "queued"}) if not session.get("running"): return None _enqueue_prompt(session, text, transport) @@ -5666,7 +6153,12 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool: """ with session["history_lock"]: queued = session.get("queued_prompt") - if not queued or session.get("running"): + if ( + not queued + or _session_busy(session) + or session.get("_finalized") + or not _session_has_live_identity(sid, session) + ): return False session["queued_prompt"] = None session["running"] = True @@ -6500,7 +6992,7 @@ def _(rid, params: dict) -> dict: session, err = _sess_nowait(params, rid) if err: return err - if session.get("running"): + if _session_busy(session): return _err(rid, 4009, "session busy") raw = str(params.get("cwd", "") or "").strip() if not raw: @@ -6537,7 +7029,7 @@ def _session_live_status(sid: str, session: dict) -> str: # session stuck mid-construction. if ready is not None and not ready.is_set() and session.get("agent_build_started"): return "starting" - if session.get("running"): + if _session_busy(session): return "working" return "idle" @@ -6644,7 +7136,7 @@ def _live_session_payload( ) inflight = _inflight_snapshot(session) queued = _queued_prompt_snapshot(session) - running = bool(session.get("running")) + running = _session_busy(session) payload = { "info": _fallback_session_info(session), "message_count": len(history), @@ -6942,7 +7434,7 @@ def _(rid, params: dict) -> dict: session, err = _sess_nowait(params, rid) if err: return err - if session.get("running"): + if _session_busy(session): return _err( rid, 4009, @@ -8739,7 +9231,7 @@ def _(rid, params: dict) -> dict: # write would either clobber the undo (version matches) or # silently drop the agent's output (version mismatch, see below). # Neither is what the user wants — make them /interrupt first. - if session.get("running"): + if _session_busy(session): return _err( rid, 4009, "session busy — /interrupt the current turn before /undo" ) @@ -8782,7 +9274,7 @@ def _(rid, params: dict) -> dict: session, err = _sess(params, rid) if err: return err - if session.get("running"): + if _session_busy(session): return _err( rid, 4009, "session busy — /interrupt the current turn before /compress" ) @@ -9333,6 +9825,9 @@ def _(rid, params: dict) -> dict: session, err = _sess_nowait(params, rid) if err: return err + with session["history_lock"]: + if session.get("idle_compression_running"): + return _err(rid, 4009, "session busy — idle compression in progress") agent = session.get("agent") if agent is None or not hasattr(agent, "steer"): return _err(rid, 4010, "agent does not support steer") @@ -9376,11 +9871,11 @@ def _(rid, params: dict) -> dict: while True: busy_transport = None with session["history_lock"]: - if session.get("running"): + if _session_busy(session): # Don't reject a mid-turn prompt — queue it (and, by default, - # interrupt the live turn) so it runs as the next turn. The - # provider interrupt itself must happen after this lock is - # released: a non-interruptible tool may keep it waiting. + # interrupt the live turn) so it runs as the next turn. Idle + # compaction uses the same lossless busy-input contract, but + # finishes and syncs before the queued turn is drained. busy_transport = t or session.get("transport") else: break @@ -9715,7 +10210,7 @@ def _notification_poller_loop( _requeued = False with session["history_lock"]: - if session.get("running"): + if _session_busy(session): process_registry.completion_queue.put(evt) _requeued = True else: @@ -9790,7 +10285,7 @@ def _notification_poller_loop( _emitted.add(_dedup_key) with session["history_lock"]: - if session.get("running"): + if _session_busy(session): process_registry.completion_queue.put(evt) break session["running"] = True @@ -10354,7 +10849,7 @@ def _stream(delta): # we check that guard before re-firing. if goal_followup: with session["history_lock"]: - if session.get("running"): + if _session_busy(session): # User already sent something — their turn wins, # the judge will re-run on the next turn anyway. return @@ -10394,7 +10889,7 @@ def _stream(delta): ) for index, (_evt, synth) in enumerate(drained): with session["history_lock"]: - if session.get("running"): + if _session_busy(session): for pending_evt, _pending_synth in drained[index:]: process_registry.completion_queue.put(pending_evt) break @@ -10425,6 +10920,8 @@ def _stream(delta): file=sys.stderr, ) + _schedule_idle_compression(sid, session) + run_thread = threading.Thread(target=run, daemon=True) session["_run_thread"] = run_thread run_thread.start() @@ -11284,7 +11781,7 @@ def _(rid, params: dict) -> dict: # with the new base_url but old model (or vice versa), # producing 400/404s the user never asked for. Parity # with the gateway's running-agent /model guard. - if session.get("running"): + if _session_busy(session): return _err( rid, 4009, @@ -13145,7 +13642,7 @@ def _(rid, params: dict) -> dict: if name == "retry": if not session: return _err(rid, 4001, "no active session to retry") - if session.get("running"): + if _session_busy(session): return _err( rid, 4009, "session busy — /interrupt the current turn before /retry" ) @@ -13275,7 +13772,7 @@ def _(rid, params: dict) -> dict: # /undo 3 backs up three user turns at once. See issue #21910. if not session: return _err(rid, 4001, "no active session to undo") - if session.get("running"): + if _session_busy(session): return _err( rid, 4009, "session busy — /interrupt the current turn before /undo" ) @@ -13390,7 +13887,7 @@ def _(rid, params: dict) -> dict: if name in {"compress", "compact"}: if not session: return _err(rid, 4001, "no active session to compress") - if session.get("running"): + if _session_busy(session): return _err( rid, 4009, "session busy — /interrupt the current turn before /compress" ) @@ -14459,7 +14956,7 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str: return str(ack.get("message") or f"compute-host {route_name} failed") _apply_compute_host_metadata_mirror(session, ack) return str(ack.get("output") or "") - if name in _MUTATES_WHILE_RUNNING and session.get("running"): + if name in _MUTATES_WHILE_RUNNING and _session_busy(session): return f"session busy — /interrupt the current turn before running /{name}" try: @@ -14993,7 +15490,7 @@ def _(rid, params: dict) -> dict: # the agent's output (version mismatch path) or clobbering the # rollback (version-matches path). A file-scoped rollback only # touches disk, so we allow it. - if not file_path and session.get("running"): + if not file_path and _session_busy(session): return _err( rid, 4009,