fix(steer): target the owned active worker after compression

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/`).

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 <text>` 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)

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()

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()