Release PF (v0.51.445): scope gateway watcher to active profile (#2848) (#4275)

* stage-3629: scope gateway watcher to active profile (profile-keyed registry), rebased onto master

* stage-3629: + stop-race-safe subscribe() (Codex CORE) + regression test

Codex caught a TOCTOU between get_watcher() and subscribe(): a watcher reaped as
subscriber-free during a concurrent profile switch left a late subscriber's queue
without the None sentinel -> SSE loop hangs open (keepalives, no events). Opus
independently found the same race (rated follow-up). subscribe() now checks
_stop_event under _sub_lock after appending and self-enqueues the sentinel.
test_subscribe_after_stop_gets_sentinel_immediately pins it.

* stage-3629: + reconnect gateway SSE to new profile on switch (Codex CORE #2)

Second Codex finding: after a profile switch this tab's _gatewaySSE stayed
subscribed to the PREVIOUS profile's watcher (switchToProfile never restarted it;
the probe reattach is gated on !_gatewaySSE which is false while the old ES is
open). switchToProfile() now calls startGatewaySSE() after S.activeProfile is set
— it closes the old ES + reconnects with the new profile cookie, self-gating on
_showCliSessions.

* Release PF: scope gateway watcher to active profile (#2848)

rodboev's #3629, re-pushed to fix both bounce findings (client SSE reconnect +
singleton→profile-keyed registry). Two Codex CORE races caught + fixed inline:
(1) stop/subscribe sentinel race, (2) tab's own SSE staying on the old profile
after switch. Codex SAFE + Opus SHIP (architecture) + gateway-watcher tests green.

Co-authored-by: rodboev <rodboev@users.noreply.github.com>

---------

Co-authored-by: nesquena-hermes <agent@nesquena-hermes>
Co-authored-by: rodboev <rodboev@users.noreply.github.com>
This commit is contained in:
authored and GitHub committed 2026-06-15 15:41:21 -07:00
1 parent fe62f445cf
commit 1bf2ef9f5e
8 files changed
+517 -28

No files matched your search

+6
View File
@@ -7,6 +7,12 @@
- **Windows pytest-harness compatibility (#3664).** Hardened the test suite to run on Windows: profile-home fallback paths are path-normalized, strict POSIX file-mode (`0o600`) assertions are gated behind `os.name != "nt"` (Linux still asserts them at full strictness), the conftest cleanup handles Windows process-tree/port teardown and the Py3.12+ `shutil.rmtree` `onexc` shim, and tests that require `fork`/`fcntl` carry `@requires_fork` / `@requires_fcntl` markers (which never skip on Linux). Test-only — no runtime or app behavior change, no Linux CI behavior change. (#4254, #4255, #4256, #4257, #4259, #4263, #4266)
## [v0.51.445] — 2026-06-15 — Release PF (scope gateway watcher to the active profile)
### Fixed
- **The gateway session watcher is now per-profile, so switching profiles in one tab no longer breaks gateway live-updates in another.** The watcher that drives gateway session SSE updates was a single process-wide instance: when one browser tab switched profiles it stopped and replaced that global watcher, silently stealing the gateway event stream from tabs on other profiles. Gateway watchers are now keyed by profile (a registry, not a singleton), each request's SSE stream subscribes to its own active profile's watcher, a profile switch only restarts that profile's watcher (never one another tab is observing), and idle watchers with no subscribers are reaped. Switching profiles in a tab now also reconnects that tab's own gateway stream to the new profile. (#2848)
## [v0.51.444] — 2026-06-15 — Release PE (refresh sidebar session list on uncommitted writes + settings changes)
### Fixed
+142 -23
View File
@@ -145,8 +145,10 @@ def _cheap_change_fingerprint(db_path: Path) -> str | None:
# ── DB resolution (shared pattern with state_sync.py) ──────────────────────
def _get_state_db_path() -> Path:
def _get_state_db_path(hermes_home: Path | None = None) -> Path:
"""Resolve state.db path for the active profile."""
if hermes_home is not None:
return Path(hermes_home).expanduser().resolve() / 'state.db'
try:
from api.profiles import get_active_hermes_home
hermes_home = Path(get_active_hermes_home()).expanduser().resolve()
@@ -155,11 +157,11 @@ def _get_state_db_path() -> Path:
return hermes_home / 'state.db'
def _get_agent_sessions_from_db() -> list:
def _get_agent_sessions_from_db(db_path: Path | None = None) -> list:
"""Read all non-webui sessions from state.db.
Returns list of session dicts, or empty list on any error.
"""
db_path = _get_state_db_path()
db_path = Path(db_path) if db_path is not None else _get_state_db_path()
if not db_path.exists():
return []
@@ -200,11 +202,24 @@ class GatewayWatcher:
POLL_INTERVAL = 5 # seconds between polls
SUBSCRIBER_TIMEOUT = 30 # seconds before sending keepalive comment
def __init__(self):
def __init__(
self,
*,
hermes_home: Path | None = None,
profile_name: str | None = None,
state_db_path: Path | None = None,
):
self._subscribers: list[queue.Queue] = []
self._sub_lock = threading.Lock()
self._stop_event = threading.Event()
self._thread: threading.Thread | None = None
self._hermes_home = Path(hermes_home).expanduser().resolve() if hermes_home else None
self._state_db_path = (
Path(state_db_path).expanduser().resolve()
if state_db_path is not None
else _get_state_db_path(self._hermes_home) if self._hermes_home is not None else _get_state_db_path()
)
self.profile_name = profile_name or ""
self._last_hash: str = ''
self._last_sessions: list = []
# Cheap sessions-only fingerprint from the previous poll. When it is
@@ -255,6 +270,16 @@ class GatewayWatcher:
q = queue.Queue(maxsize=10)
with self._sub_lock:
self._subscribers.append(q)
# Stop-race safety: if stop() already ran (set _stop_event and drained
# the then-current subscriber list) before we appended, this queue would
# never receive the sentinel and the SSE loop would hang open with
# keepalives but no events. Enqueue the sentinel ourselves so the handler
# closes and reconnects to the live registry watcher. (#3629 / Codex gate)
if self._stop_event.is_set():
try:
q.put_nowait(None)
except Exception:
logger.debug("Failed to send stop sentinel to late subscriber")
return q
def unsubscribe(self, q: queue.Queue):
@@ -303,7 +328,7 @@ class GatewayWatcher:
# of message rows every 5 seconds (issue #3506). A None
# fingerprint (error / unreadable db) forces the full read so we
# never silently skip a real change.
db_path = _get_state_db_path()
db_path = self._state_db_path
cheap_fp = _cheap_change_fingerprint(db_path) if db_path.exists() else ''
if cheap_fp is not None and cheap_fp == self._last_cheap_fp:
# Nothing changed in the sidebar-visible session set; skip
@@ -311,7 +336,7 @@ class GatewayWatcher:
pass
else:
# Phase 2: only now pay for the full projection.
sessions = _get_agent_sessions_from_db()
sessions = _get_agent_sessions_from_db(db_path)
current_hash = _snapshot_hash(sessions)
if cheap_fp is not None:
self._last_cheap_fp = cheap_fp
@@ -330,31 +355,125 @@ class GatewayWatcher:
time.sleep(0.1)
# ── Module-level singleton ─────────────────────────────────────────────────
# ── Module-level watcher registry ──────────────────────────────────────────
_watcher: GatewayWatcher | None = None
_watchers: dict[str, GatewayWatcher] = {}
_watcher_lock = threading.Lock()
def _resolve_watcher_target(
*,
profile_name: str | None = None,
hermes_home: Path | None = None,
) -> tuple[str, Path | None]:
"""Resolve the watcher profile/home pair for the current request context."""
resolved_profile = str(profile_name or "").strip()
resolved_home = Path(hermes_home).expanduser().resolve() if hermes_home is not None else None
def start_watcher():
"""Start the global gateway watcher (idempotent)."""
global _watcher
try:
from api.profiles import get_active_profile_name, get_hermes_home_for_profile
if not resolved_profile:
resolved_profile = get_active_profile_name() or "default"
if resolved_home is None and resolved_profile:
resolved_home = Path(get_hermes_home_for_profile(resolved_profile)).expanduser().resolve()
except Exception:
if resolved_home is None:
try:
resolved_home = _get_state_db_path().parent.resolve()
except Exception:
resolved_home = None
return resolved_profile, resolved_home
def _watcher_registry_key(profile_name: str | None = None, hermes_home: Path | None = None) -> str:
"""Return the stable registry key for a watcher target."""
if hermes_home is not None:
return str(Path(hermes_home).expanduser().resolve())
return str(profile_name or "").strip() or "__default__"
def _watcher_has_subscribers(watcher: GatewayWatcher) -> bool:
subscribers = getattr(watcher, "_subscribers", None)
sub_lock = getattr(watcher, "_sub_lock", None)
if subscribers is None or sub_lock is None:
return False
with sub_lock:
return bool(subscribers)
def _pop_idle_watchers_locked(*, exclude_key: str) -> list[GatewayWatcher]:
stale: list[GatewayWatcher] = []
for key, watcher in list(_watchers.items()):
if key == exclude_key or _watcher_has_subscribers(watcher):
continue
if _watchers.get(key) is watcher:
stale.append(_watchers.pop(key))
return stale
def start_watcher(*, profile_name: str | None = None, hermes_home: Path | None = None):
"""Start the watcher for the resolved profile home (idempotent)."""
resolved_profile, resolved_home = _resolve_watcher_target(
profile_name=profile_name,
hermes_home=hermes_home,
)
key = _watcher_registry_key(resolved_profile, resolved_home)
with _watcher_lock:
if _watcher is None:
_watcher = GatewayWatcher()
_watcher.start()
watcher = _watchers.get(key)
if watcher is None or not watcher.is_alive():
if watcher is not None:
watcher.stop()
watcher = GatewayWatcher(profile_name=resolved_profile, hermes_home=resolved_home)
watcher.start()
_watchers[key] = watcher
return watcher
def stop_watcher():
"""Stop the global gateway watcher."""
global _watcher
def stop_watcher(*, profile_name: str | None = None, hermes_home: Path | None = None):
"""Stop either one profile watcher or the entire registry."""
with _watcher_lock:
if _watcher is not None:
_watcher.stop()
_watcher = None
if profile_name is None and hermes_home is None:
watchers = list(_watchers.values())
_watchers.clear()
else:
resolved_profile, resolved_home = _resolve_watcher_target(
profile_name=profile_name,
hermes_home=hermes_home,
)
key = _watcher_registry_key(resolved_profile, resolved_home)
watcher = _watchers.pop(key, None)
watchers = [watcher] if watcher is not None else []
for watcher in watchers:
watcher.stop()
def get_watcher() -> GatewayWatcher | None:
"""Get the global watcher instance (or None if not started)."""
def restart_watcher_for_profile(name: str):
"""Restart only the watcher pinned to the target profile home."""
from api.profiles import get_hermes_home_for_profile
hermes_home = Path(get_hermes_home_for_profile(name)).expanduser().resolve()
key = _watcher_registry_key(name, hermes_home)
watcher = GatewayWatcher(profile_name=name, hermes_home=hermes_home)
watcher.start()
with _watcher_lock:
return _watcher
existing = _watchers.pop(key, None)
stale_watchers = [] if existing is not None else _pop_idle_watchers_locked(exclude_key=key)
_watchers[key] = watcher
for old_watcher in ([existing] if existing is not None else stale_watchers):
old_watcher.stop()
return watcher
def get_watcher(*, profile_name: str | None = None, hermes_home: Path | None = None) -> GatewayWatcher | None:
"""Get or lazily start the watcher for the resolved request profile."""
resolved_profile, resolved_home = _resolve_watcher_target(
profile_name=profile_name,
hermes_home=hermes_home,
)
key = _watcher_registry_key(resolved_profile, resolved_home)
with _watcher_lock:
watcher = _watchers.get(key)
if watcher is None or not watcher.is_alive():
watcher = start_watcher(profile_name=resolved_profile, hermes_home=resolved_home)
return watcher
+5
View File
@@ -9206,6 +9206,11 @@ def handle_post(handler, parsed) -> bool:
# the old profile's cached model list (#1200 — profile-switch model bug).
from api.config import invalidate_models_cache
invalidate_models_cache()
try:
from api.gateway_watcher import restart_watcher_for_profile
restart_watcher_for_profile(name)
except Exception as exc:
logger.warning("Failed to restart gateway watcher for profile %s: %s", name, exc)
return j(handler, result, extra_headers={
'Set-Cookie': build_profile_cookie(name, handler),
})
+8
View File
@@ -5591,6 +5591,14 @@ async function switchToProfile(name) {
S.activeProfile = data.active || name;
S.activeProfileIsDefault = !!data.is_default;
// Reconnect the gateway SSE to the NEW profile's watcher. The backend watcher
// registry is now profile-keyed (#3629), but this tab's existing EventSource is
// still subscribed to the PREVIOUS profile's watcher — and the probe-based
// reattach is gated on `!_gatewaySSE`, which can't fire while the old stream is
// open. startGatewaySSE() closes the old ES (stopGatewaySSE) and reconnects with
// the new profile cookie; it self-gates on window._showCliSessions internally.
if (typeof startGatewaySSE === 'function') startGatewaySSE();
// Update composer placeholder and title bar while the core profile-switch
// state is still close to the profile API response.
if (typeof applyBotName === 'function') applyBotName();
+1
View File
@@ -3940,6 +3940,7 @@ async function probeGatewaySSEStatus(){
if(resp.ok && data.watcher_running){
stopGatewayPollFallback();
_gatewaySSEWarningShown = false;
if(!_gatewaySSE && typeof EventSource!=='undefined' && !(document&&document.hidden)) startGatewaySSE();
return;
}
if(resp.status === 503 || data.watcher_running === false){
+9 -1
View File
@@ -22,7 +22,7 @@ def _block(src: str, start: str, end: str) -> str:
def test_gateway_watcher_remains_hash_only():
"""The watcher should not try to infer restarts from state.db mtime."""
src = _read(GATEWAY_WATCHER)
poll = _block(src, " def _poll_loop(self):", "\n_watcher:")
poll = _block(src, " def _poll_loop(self):", "\n_watchers:")
assert "_get_db_mtime" not in src
assert "_detect_gateway_restart" not in src
@@ -46,6 +46,14 @@ def test_gateway_sse_dedupes_reconnect_snapshot_before_refresh():
assert "renderSessionList({deferWhileInteracting:true}); // re-fetch and re-render" in handler
def test_gateway_probe_reattaches_sse_after_profile_switch_restart():
"""A healthy probe must revive the EventSource when the watcher restarted."""
src = _read(SESSIONS_JS)
probe = _block(src, "async function probeGatewaySSEStatus()", "\n\nfunction startGatewaySSE")
assert "if(!_gatewaySSE && typeof EventSource!=='undefined' && !(document&&document.hidden)) startGatewaySSE();" in probe
def test_gateway_snapshot_key_matches_backend_hash_fields():
"""Frontend dedupe must compare the same fields that drive watcher events."""
src = _read(SESSIONS_JS)
+342
View File
@@ -0,0 +1,342 @@
from __future__ import annotations
import json
import threading
from urllib.parse import urlparse
class _FakeHandler:
def __init__(self):
self.status = None
self.sent_headers: list[tuple[str, str]] = []
self.body = bytearray()
self.wfile = self
self.headers = {}
def send_response(self, code):
self.status = code
def send_header(self, key, value):
self.sent_headers.append((key, value))
def end_headers(self):
pass
def write(self, data):
self.body.extend(data if isinstance(data, (bytes, bytearray)) else data.encode("utf-8"))
def get_json(self):
return json.loads(self.body.decode("utf-8"))
def test_gateway_watcher_pins_explicit_profile_home(tmp_path, monkeypatch):
from api import gateway_watcher as gw
profile_a = tmp_path / "profile-a"
profile_b = tmp_path / "profile-b"
profile_a.mkdir()
profile_b.mkdir()
(profile_a / "state.db").touch()
seen = []
def fake_rows(db_path, **kwargs):
seen.append(db_path)
return [
{
"id": "session-a",
"title": "Profile A",
"model": None,
"message_count": 1,
"actual_message_count": 1,
"started_at": 1,
"last_activity": 2,
"source": "telegram",
"raw_source": "telegram",
"session_source": "gateway",
"source_label": "Telegram",
}
]
monkeypatch.setattr(gw, "read_importable_agent_session_rows", fake_rows)
monkeypatch.setattr(
gw,
"_get_state_db_path",
lambda *args: args[0].resolve() / "state.db" if args and args[0] is not None else profile_b / "state.db",
)
watcher = gw.GatewayWatcher(hermes_home=profile_a, profile_name="a")
sessions = gw._get_agent_sessions_from_db(watcher._state_db_path)
assert watcher.profile_name == "a"
assert watcher._state_db_path == profile_a.resolve() / "state.db"
assert seen == [profile_a.resolve() / "state.db"]
assert sessions[0]["session_id"] == "session-a"
def test_restart_watcher_for_profile_replaces_singleton_with_profile_home(tmp_path, monkeypatch):
from api import gateway_watcher as gw
from api import profiles
created = []
class FakeWatcher:
def __init__(self, *, profile_name="", hermes_home=None, state_db_path=None):
self.profile_name = profile_name
self.hermes_home = hermes_home
self.state_db_path = state_db_path
self.started = False
self.stopped = False
created.append(self)
def start(self):
self.started = True
def is_alive(self):
return self.started
def stop(self):
self.stopped = True
old_home = (tmp_path / "old").resolve()
old = FakeWatcher(profile_name="old", hermes_home=old_home)
monkeypatch.setattr(gw, "_watchers", {str(old_home): old})
monkeypatch.setattr(gw, "GatewayWatcher", FakeWatcher)
monkeypatch.setattr(profiles, "get_hermes_home_for_profile", lambda name: tmp_path / name)
monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "target")
watcher = gw.restart_watcher_for_profile("target")
assert old.stopped is True
assert watcher is created[-1]
assert watcher.profile_name == "target"
assert watcher.hermes_home == tmp_path / "target"
assert watcher.started is True
assert gw.get_watcher() is watcher
def test_restart_watcher_for_profile_keeps_subscribed_other_profile(tmp_path, monkeypatch):
from api import gateway_watcher as gw
from api import profiles
default_home = (tmp_path / "default").resolve()
work_home = (tmp_path / "work").resolve()
created = []
class FakeWatcher:
def __init__(self, *, profile_name="", hermes_home=None, state_db_path=None):
self.profile_name = profile_name
self.hermes_home = hermes_home
self.state_db_path = state_db_path
self.started = False
self.stopped = False
self._subscribers = []
self._sub_lock = threading.Lock()
created.append(self)
def start(self):
self.started = True
def is_alive(self):
return self.started
def stop(self):
self.stopped = True
self.started = False
default_watcher = FakeWatcher(profile_name="default", hermes_home=default_home)
default_watcher.started = True
default_watcher._subscribers.append(object())
monkeypatch.setattr(gw, "_watchers", {str(default_home): default_watcher})
monkeypatch.setattr(gw, "GatewayWatcher", FakeWatcher)
monkeypatch.setattr(profiles, "get_hermes_home_for_profile", lambda name: work_home)
watcher = gw.restart_watcher_for_profile("work")
assert default_watcher.stopped is False
assert gw._watchers[str(default_home)] is default_watcher
assert gw._watchers[str(work_home)] is watcher
assert watcher.profile_name == "work"
def test_restart_watcher_for_profile_swaps_atomically(tmp_path, monkeypatch):
from api import gateway_watcher as gw
from api import profiles
target_home = (tmp_path / "target").resolve()
created = []
seen = []
class FakeWatcher:
def __init__(self, *, profile_name="", hermes_home=None, state_db_path=None):
self.profile_name = profile_name
self.hermes_home = hermes_home
self.state_db_path = state_db_path
self.started = False
self.stopped = False
created.append(self)
def start(self):
self.started = True
if len(seen) == 0:
seen.append(gw.get_watcher(profile_name="target", hermes_home=target_home))
def is_alive(self):
return self.started
def stop(self):
self.stopped = True
self.started = False
existing = FakeWatcher(profile_name="target", hermes_home=target_home)
existing.started = True
monkeypatch.setattr(gw, "_watchers", {str(target_home): existing})
monkeypatch.setattr(gw, "GatewayWatcher", FakeWatcher)
monkeypatch.setattr(profiles, "get_hermes_home_for_profile", lambda name: target_home)
watcher = gw.restart_watcher_for_profile("target")
assert len(created) == 2
assert seen == [existing]
assert existing.stopped is True
assert watcher is created[-1]
assert gw.get_watcher(profile_name="target", hermes_home=target_home) is watcher
def test_watcher_registry_key_uses_concrete_values(tmp_path, monkeypatch):
from api import gateway_watcher as gw
def fail_resolve(**kwargs):
raise AssertionError("_watcher_registry_key should not resolve profile state")
monkeypatch.setattr(gw, "_resolve_watcher_target", fail_resolve)
assert gw._watcher_registry_key("work", tmp_path / "work") == str((tmp_path / "work").resolve())
assert gw._watcher_registry_key("default", None) == "default"
def test_start_watcher_pins_active_profile_home(tmp_path, monkeypatch):
from api import gateway_watcher as gw
from api import profiles
profile_home = tmp_path / "active-profile"
created = []
class FakeWatcher:
def __init__(self, *, profile_name="", hermes_home=None, state_db_path=None):
self.profile_name = profile_name
self.hermes_home = hermes_home
self.state_db_path = state_db_path
self.started = False
created.append(self)
def start(self):
self.started = True
def is_alive(self):
return self.started
monkeypatch.setattr(gw, "_watchers", {})
monkeypatch.setattr(gw, "GatewayWatcher", FakeWatcher)
monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "work")
monkeypatch.setattr(profiles, "get_hermes_home_for_profile", lambda name: profile_home)
gw.start_watcher()
assert len(created) == 1
assert created[0].profile_name == "work"
assert created[0].hermes_home == profile_home
assert created[0].started is True
assert gw.get_watcher() is created[0]
def test_get_watcher_scopes_lookup_to_active_profile(tmp_path, monkeypatch):
from api import gateway_watcher as gw
from api import profiles
default_home = (tmp_path / "default").resolve()
work_home = (tmp_path / "work").resolve()
class FakeWatcher:
def __init__(self, *, profile_name="", hermes_home=None, state_db_path=None):
self.profile_name = profile_name
self.hermes_home = hermes_home
self.started = True
def is_alive(self):
return True
default_watcher = FakeWatcher(profile_name="default", hermes_home=default_home)
work_watcher = FakeWatcher(profile_name="work", hermes_home=work_home)
monkeypatch.setattr(
gw,
"_watchers",
{
str(default_home): default_watcher,
str(work_home): work_watcher,
},
)
monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "work")
monkeypatch.setattr(
profiles,
"get_hermes_home_for_profile",
lambda name: work_home if name == "work" else default_home,
)
assert gw.get_watcher() is work_watcher
def test_profile_switch_restarts_watcher_best_effort(monkeypatch):
from api import config, gateway_watcher, profiles, routes
calls = []
monkeypatch.setattr(routes, "_check_csrf", lambda handler: True)
monkeypatch.setattr(routes, "read_body", lambda handler: {"name": "demo"})
monkeypatch.setattr(profiles, "_validate_profile_name", lambda name: None)
monkeypatch.setattr(profiles, "switch_profile", lambda name, process_wide=False: {"ok": True, "name": name})
monkeypatch.setattr(config, "invalidate_models_cache", lambda: calls.append("cache"))
monkeypatch.setattr(gateway_watcher, "restart_watcher_for_profile", lambda name: calls.append(("watcher", name)))
handler = _FakeHandler()
routes.handle_post(handler, urlparse("/api/profile/switch"))
assert handler.status == 200
assert handler.get_json() == {"ok": True, "name": "demo"}
assert calls == ["cache", ("watcher", "demo")]
assert any(k == "Set-Cookie" for k, _v in handler.sent_headers)
def test_profile_switch_response_survives_watcher_restart_failure(monkeypatch):
from api import config, gateway_watcher, profiles, routes
monkeypatch.setattr(routes, "_check_csrf", lambda handler: True)
monkeypatch.setattr(routes, "read_body", lambda handler: {"name": "demo"})
monkeypatch.setattr(profiles, "_validate_profile_name", lambda name: None)
monkeypatch.setattr(profiles, "switch_profile", lambda name, process_wide=False: {"ok": True, "name": name})
monkeypatch.setattr(config, "invalidate_models_cache", lambda: None)
monkeypatch.setattr(
gateway_watcher,
"restart_watcher_for_profile",
lambda name: (_ for _ in ()).throw(RuntimeError("restart failed")),
)
handler = _FakeHandler()
routes.handle_post(handler, urlparse("/api/profile/switch"))
assert handler.status == 200
assert handler.get_json() == {"ok": True, "name": "demo"}
def test_subscribe_after_stop_gets_sentinel_immediately():
"""Stop-race safety (#3629 / Codex gate): a subscriber added AFTER stop() has
already set _stop_event and drained the then-current subscriber list must still
receive the None sentinel — otherwise the SSE loop hangs open with keepalives but
no events (it never learns the watcher it attached to was reaped during a
concurrent profile switch)."""
from api import gateway_watcher as gw
watcher = gw.GatewayWatcher(hermes_home=None, profile_name="race")
# Simulate the reaped/stopped watcher: stop() ran before this subscribe().
watcher.stop()
q = watcher.subscribe()
# The late subscriber must find the sentinel already queued (no block / hang).
assert q.get_nowait() is None
+4 -4
View File
@@ -368,9 +368,9 @@ def test_poll_loop_skips_projection_when_unchanged(tmp_path, monkeypatch):
calls = {"n": 0}
real = gw._get_agent_sessions_from_db
def counting():
def counting(db_path=None):
calls["n"] += 1
return real()
return real(db_path)
monkeypatch.setattr(gw, "_get_agent_sessions_from_db", counting)
@@ -378,11 +378,11 @@ def test_poll_loop_skips_projection_when_unchanged(tmp_path, monkeypatch):
# Run the change-detection body directly (one iteration) without the thread.
def one_iteration():
db_path = gw._get_state_db_path()
db_path = w._state_db_path
cheap_fp = gw._cheap_change_fingerprint(db_path) if db_path.exists() else ''
if cheap_fp is not None and cheap_fp == w._last_cheap_fp:
return
sessions = gw._get_agent_sessions_from_db()
sessions = gw._get_agent_sessions_from_db(db_path)
if cheap_fp is not None:
w._last_cheap_fp = cheap_fp
_ = gw._snapshot_hash(sessions)