From c6152e4182657ec8dbb401b438e8438f0e2c52a5 Mon Sep 17 00:00:00 2001 From: Anthony Ruiz Date: Sat, 12 Sep 2026 20:02:40 +0000 Subject: [PATCH] fix(steer): target the owned active worker after compression --- AGENTS.md | 4 + api/streaming.py | 72 ++++++++--- tests/test_steer_active_run_identity.py | 112 ++++++++++++++++++ tests/test_streaming_agent_cache_lifecycle.py | 2 - 4 files changed, 170 insertions(+), 20 deletions(-) create mode 100644 tests/test_steer_active_run_identity.py diff --git a/AGENTS.md b/AGENTS.md index d1b785bc..d59c8cee 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -67,6 +67,10 @@ Follow that checklist's safety rules: the manual verification performed. - For runtime, streaming, recovery, replay, compression, or sidebar metadata changes, name the state layer being mutated and prove the relevant invariant. +- Active-run Steer resolves the stream-bound agent with explicit stream and + worker ownership before consulting the reusable session cache. Compression + may rotate the agent identity; steering must never evict or close an agent. + Keep HTTP response writes outside runtime registry locks. - For Docker build changes in `docker_init.bash`, mirror directory exclusions in both the `rsync` and `cp -a` paths — `/opt/hermes` may contain subdirectories with restricted permissions (e.g. `.playwright/`). diff --git a/api/streaming.py b/api/streaming.py index 71cf8208..c6770ddd 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -13536,13 +13536,50 @@ def _run_agent_streaming( # ============================================================ +def _steer_bound_stream(sid: str, stream_id: str, text: str) -> dict | None: + """Deliver to a verified live worker; None retains cache-only compatibility. + + Never perform HTTP writes or cache/database teardown under stream locks. + """ + from api import config as cfg + + with cfg.STREAMS_LOCK: + agent = cfg.AGENT_INSTANCES.get(stream_id) + owner = cfg.stream_owner_session_id(stream_id) + with cfg.ACTIVE_RUNS_LOCK: + run = dict(cfg.ACTIVE_RUNS.get(stream_id) or {}) + if agent is None: + if ((owner and owner != sid) + or (run.get("session_id") and run["session_id"] != sid) + or run.get("phase") == "cancelling"): + return {"accepted": False, "fallback": "stream_dead", "stream_id": None} + return None + # A cache hit cannot override missing or conflicting worker ownership. + if (stream_id not in cfg.STREAMS + or owner != sid or run.get("session_id") != sid + or run.get("phase") == "cancelling" + or run.get("backend") == "gateway"): + return {"accepted": False, "fallback": "stream_dead", "stream_id": None} + if not callable(getattr(agent, "steer", None)): + return {"accepted": False, "fallback": "agent_lacks_steer", "stream_id": None} + # steer() only stashes input; serializing it with stream teardown keeps + # cancel from detaching the selected worker before delivery. + try: + accepted = bool(agent.steer(text)) + except Exception: + logger.debug("Stream-bound steer failed for session %s", sid, exc_info=True) + return {"accepted": False, "fallback": "steer_error", "stream_id": stream_id} + return {"accepted": accepted, "fallback": None, "stream_id": stream_id} + + def _handle_chat_steer(handler, body: dict) -> bool: """Inject a /steer payload into the active agent for a session. Mirrors the CLI's `/steer ` command (cli.py:6140-6155): - - Look up the cached AIAgent for the session (PR #1051's - SESSION_AGENT_CACHE). + - Prefer the active stream's registered AIAgent, with explicit stream + and worker ownership; use the cache only for legacy cache-only runs. - Verify a stream is currently active for this session. + - Never evict or close an agent while delivering steering input. - Call agent.steer(text) — thread-safe, stashes text in _pending_steer for application at the next tool-result boundary. @@ -13571,24 +13608,23 @@ def _handle_chat_steer(handler, body: dict) -> bool: if not text: return bad(handler, "text required") - evicted_cached_entry = None + # A compression rotates agent.session_id without replacing the running + # worker. Resolve the stream-bound instance BEFORE consulting the reusable + # cache. Steer never owns cache eviction or agent/database teardown. + try: + session = get_session(sid) + except KeyError: + session = None + stream_id = getattr(session, "active_stream_id", None) or None + if stream_id: + result = _steer_bound_stream(sid, stream_id, text) + if result is not None: + return j(handler, result) + with _cfg.SESSION_AGENT_CACHE_LOCK: cached = _cfg.SESSION_AGENT_CACHE.get(sid) - if cached: - agent = cached[0] - if not _cached_agent_matches_session(agent, sid): - evicted_cached_entry = _cfg.SESSION_AGENT_CACHE.pop(sid, None) - logger.warning( - '[webui] Evicted cached agent before steer due to mismatched session identity: cache_key=%s agent_session_id=%s', - sid, - _cached_agent_session_identity(agent), - ) - cached = None - if evicted_cached_entry is not None: - try: - _close_cached_agent_entry_at_session_boundary(sid, evicted_cached_entry) - except Exception: - logger.debug("Failed to close steer identity-mismatched cached agent for session %s", sid, exc_info=True) + if cached and not _cached_agent_matches_session(cached[0], sid): + cached = None if not cached: try: s = get_session(sid) diff --git a/tests/test_steer_active_run_identity.py b/tests/test_steer_active_run_identity.py new file mode 100644 index 00000000..c875cea8 --- /dev/null +++ b/tests/test_steer_active_run_identity.py @@ -0,0 +1,112 @@ +"""Steer must use the active worker, not a compression-sensitive cache key.""" +import queue +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from api import config, streaming +from tests.test_real_steer import _make_handler, _captured_response + + +@pytest.fixture +def scene(monkeypatch): + agent = MagicMock() + agent.session_id = "compressed-child" + agent.steer.return_value = True + other = MagicMock() + other.session_id = "other-session" + monkeypatch.setattr(config, "SESSION_AGENT_CACHE", {"original": (agent, "sig")}) + monkeypatch.setattr(config, "STREAMS", {"run": queue.Queue(), "other-run": queue.Queue()}) + monkeypatch.setattr(config, "AGENT_INSTANCES", {"run": agent, "other-run": other}) + monkeypatch.setattr(config, "STREAM_SESSION_OWNERS", {"run": "original", "other-run": "other-session"}) + monkeypatch.setattr(config, "ACTIVE_RUNS", { + "run": {"session_id": "original", "phase": "running", "backend": "legacy"}, + "other-run": {"session_id": "other-session", "phase": "running", "backend": "legacy"}, + }) + sess = SimpleNamespace(active_stream_id="run") + monkeypatch.setattr(streaming, "get_session", lambda sid: sess) + return agent, other, sess + + +def steer(): + h = _make_handler() + streaming._handle_chat_steer(h, {"session_id": "original", "text": "updated guidance"}) + return _captured_response(h) + + +@pytest.mark.parametrize("cache", ["rotated", "missing", "different"]) +def test_active_worker_accepts_after_compression_without_cache_mutation(scene, cache): + agent, other, _ = scene + if cache == "missing": + config.SESSION_AGENT_CACHE.clear() + elif cache == "different": + config.SESSION_AGENT_CACHE["original"] = (other, "sig") + before = dict(config.SESSION_AGENT_CACHE) + assert steer() == {"accepted": True, "fallback": None, "stream_id": "run"} + agent.steer.assert_called_once_with("updated guidance") + other.steer.assert_not_called() + agent._session_db.close.assert_not_called() + other._session_db.close.assert_not_called() + agent.interrupt.assert_not_called() + assert config.SESSION_AGENT_CACHE == before + + +@pytest.mark.parametrize("bad", ["owner", "run-owner", "missing-owner", "missing-run", "cancelling", "dead", "gateway"]) +def test_ambiguous_or_inactive_worker_is_never_steered_or_closed(scene, bad): + agent, other, _ = scene + # Matching cache must not bypass conflicting active-run ownership. + agent.session_id = "original" + if bad == "owner": + config.STREAM_SESSION_OWNERS["run"] = "other-session" + elif bad == "run-owner": + config.ACTIVE_RUNS["run"]["session_id"] = "other-session" + elif bad == "missing-owner": + config.STREAM_SESSION_OWNERS.pop("run") + elif bad == "missing-run": + config.ACTIVE_RUNS.pop("run") + elif bad == "cancelling": + config.ACTIVE_RUNS["run"]["phase"] = "cancelling" + elif bad == "dead": + config.STREAMS.pop("run") + elif bad == "gateway": + config.ACTIVE_RUNS["run"]["backend"] = "gateway" + assert steer()["accepted"] is False + agent.steer.assert_not_called() + other.steer.assert_not_called() + agent._session_db.close.assert_not_called() + + +def test_cache_only_mismatch_does_not_close_an_agent_owned_by_another_run(scene): + agent, other, _ = scene + config.AGENT_INSTANCES.pop("run") + config.SESSION_AGENT_CACHE["original"] = (other, "sig") + assert steer()["accepted"] is False + other._session_db.close.assert_not_called() + other.steer.assert_not_called() + + +def test_http_response_is_written_after_stream_lock_release(scene, monkeypatch): + from api import helpers + + def respond(handler, payload): + assert config.STREAMS_LOCK.acquire(blocking=False) + config.STREAMS_LOCK.release() + return payload + + monkeypatch.setattr(helpers, "j", respond) + result = streaming._handle_chat_steer(None, {"session_id": "original", "text": "hint"}) + assert result["accepted"] is True + + +@pytest.mark.parametrize("behavior", ["reject", "raise", "unsupported"]) +def test_live_worker_reports_real_acceptance(scene, behavior): + agent, _, _ = scene + if behavior == "reject": + agent.steer.return_value = False + elif behavior == "raise": + agent.steer.side_effect = RuntimeError("fixture") + else: + agent.steer = None + assert steer()["accepted"] is False + agent.interrupt.assert_not_called() diff --git a/tests/test_streaming_agent_cache_lifecycle.py b/tests/test_streaming_agent_cache_lifecycle.py index 1cf513be..9cf247c8 100644 --- a/tests/test_streaming_agent_cache_lifecycle.py +++ b/tests/test_streaming_agent_cache_lifecycle.py @@ -96,7 +96,6 @@ def test_identity_mismatch_cache_evictions_close_entries_outside_cache_lock(): "_identity_mismatch_entry = SESSION_AGENT_CACHE.pop(session_id, None)", "_stale_runtime_entry = SESSION_AGENT_CACHE.pop(session_id, None)", "_skipped_agent_migration_entry = _cached_entry", - "evicted_cached_entry = _cfg.SESSION_AGENT_CACHE.pop(sid, None)", "_evicted_entry = SESSION_AGENT_CACHE.pop(session_id, None)", ] for marker in expected_markers: @@ -106,7 +105,6 @@ def test_identity_mismatch_cache_evictions_close_entries_outside_cache_lock(): "_close_cached_agent_entry_at_session_boundary(session_id, _identity_mismatch_entry)", "_close_cached_agent_entry_at_session_boundary(session_id, _stale_runtime_entry)", "_close_cached_agent_entry_at_session_boundary(old_sid, _skipped_agent_migration_entry)", - "_close_cached_agent_entry_at_session_boundary(sid, evicted_cached_entry)", "_close_cached_agent_entry_at_session_boundary(session_id, _evicted_entry)", ] lines = src.splitlines()