fix(steer): make cache-only delivery atomic with cancellation

The cache-only compatibility path selected the matching cached agent,
checked stream liveness under STREAMS_LOCK, released the lock, and only
then called agent.steer(). A Stop running in that gap published
phase='cancelling' and detached STREAMS/AGENT_INSTANCES while the
earlier Steer still enqueued into the cached worker, so the HTTP
response reported accepted guidance that cancellation already owned.

Revalidate stream membership, owner, and active-run
session/backend/phase and call agent.steer() under the same
STREAMS_LOCK -> ACTIVE_RUNS_LOCK edge Stop uses; interrupt,
persistence, and HTTP writes stay outside the lock. Cache eviction and
agent closing remain absent.

Extend tests/test_steer_cancel_ordering.py with a cache-only reverse-
order test (AGENT_INSTANCES absent, matching cached worker, decoy cache
entry) gating both unlocked edges around the stream-lock reacquire.
Stop-first must return stream_dead without calling agent.steer(); the
selected cache object is asserted to be the intended worker. Proven
red on the pre-fix code (order ['cancel','steer']) and green after.

AGENTS.md

@@ -75,7 +75,12 @@ Follow that checklist's safety rules:
   even when no in-process worker is registered for that stream. Stop publishes
   cancellation and detaches stream/agent entries under the same stream lock
   (STREAMS_LOCK -> ACTIVE_RUNS_LOCK); interrupt and session persistence remain
-  outside it. Test both Stop/Steer orderings with deterministic barriers.
+  outside it. The retained cache-only path (no registered worker) revalidates
+  stream membership, owner, and active-run session/backend/phase and enqueues
+  agent.steer() under that same lock edge, so a Stop that claims cancellation
+  never strands guidance an earlier Steer response reported as accepted. Test
+  both Stop/Steer orderings — registered and cache-only — with deterministic
+  barriers.
 - 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

@@ -13678,22 +13678,39 @@ def _handle_chat_steer(handler, body: dict) -> bool:
     if not active_stream_id:
         return j(handler, {"accepted": False, "fallback": "not_running",
                            "stream_id": None})
+
+    # Cache-only compatibility: no registered worker for this stream. The
+    # liveness/ownership revalidation and the enqueue must share the same
+    # stream-ownership edge as Stop (STREAMS_LOCK -> ACTIVE_RUNS_LOCK). A Stop
+    # that claims cancellation between an unlocked check and agent.steer()
+    # would strand guidance this response still reports as accepted. steer()
+    # only stashes input; interrupt, persistence, and HTTP writes stay outside
+    # the lock, and the cached agent is never evicted or closed here.
+    result = {"accepted": False, "fallback": "stream_dead", "stream_id": None}
     with _cfg.STREAMS_LOCK:
-        stream_alive = active_stream_id in _cfg.STREAMS
-    if not stream_alive:
-        # Active stream id is stale — stream has ended; caller falls back
-        return j(handler, {"accepted": False, "fallback": "stream_dead",
-                           "stream_id": None})
+        if active_stream_id in _cfg.STREAMS:
+            owner = _cfg.stream_owner_session_id(active_stream_id)
+            with _cfg.ACTIVE_RUNS_LOCK:
+                run = dict((_cfg.ACTIVE_RUNS or {}).get(str(active_stream_id)) or {})
+            conflicting_owner = bool(owner and owner != sid)
+            conflicting_run = bool(run.get("session_id") and run["session_id"] != sid)
+            if not conflicting_owner and not conflicting_run and run.get("phase") != "cancelling":
+                if run.get("backend") == "gateway":
+                    # Gateway owns transport; a local cache object is never steered.
+                    result = {"accepted": False, "fallback": "gateway_steer_queued",
+                              "stream_id": active_stream_id}
+                else:
+                    try:
+                        accepted = bool(agent.steer(text))
+                    except Exception as exc:
+                        logger.debug("agent.steer() raised for session=%s: %s", sid, exc)
+                        result = {"accepted": False, "fallback": "steer_error",
+                                  "stream_id": active_stream_id}
+                    else:
+                        result = {"accepted": accepted, "fallback": None,
+                                  "stream_id": active_stream_id}
 
-    try:
-        accepted = bool(agent.steer(text))
-    except Exception as exc:
-        logger.debug("agent.steer() raised for session=%s: %s", sid, exc)
-        return j(handler, {"accepted": False, "fallback": "steer_error",
-                           "stream_id": active_stream_id})
-
-    return j(handler, {"accepted": accepted, "fallback": None,
-                       "stream_id": active_stream_id})
+    return j(handler, result)
 
 
 def cancel_stream(stream_id: str) -> bool:

tests/test_steer_cancel_ordering.py

@@ -19,8 +19,13 @@ class ObservedLock:
         self.lock = threading.Lock()
         self.owner = None
         self.contender = threading.Event()
+        # Optional unlocked hooks: before acquisition and after release.
+        self.before_acquire = None
+        self.after_release = None
 
     def __enter__(self):
+        if self.before_acquire is not None:
+            self.before_acquire()
         if self.owner is not None and self.owner != threading.get_ident():
             self.contender.set()
         assert self.lock.acquire(timeout=5), "stream lock acquisition timed out"
@@ -30,6 +35,8 @@ class ObservedLock:
     def __exit__(self, *args):
         self.owner = None
         self.lock.release()
+        if self.after_release is not None:
+            self.after_release()
 
 
 @pytest.fixture
@@ -115,8 +122,11 @@ def test_cancel_snapshot_claims_stream_before_later_steer(scene, monkeypatch, re
     assert config.STREAM_LIVE_TOOL_CALLS["run"] == [{"name": "example"}]
 
 
-def test_steer_claimed_first_enqueues_before_cancel(scene):
+@pytest.mark.parametrize("registered", [True, False])
+def test_steer_claimed_first_enqueues_before_cancel(scene, registered):
     lock, agent = scene
+    if not registered:
+        config.AGENT_INSTANCES.clear()  # Matching-cache compatibility path.
     reached = threading.Event()
     release = threading.Event()
     order = []
@@ -149,3 +159,89 @@ def test_steer_claimed_first_enqueues_before_cancel(scene):
         assert guidance.result(timeout=5)["accepted"] is True
         assert cancel.result(timeout=5) is True
     assert order == ["steer", "cancel"]
+
+
+@pytest.mark.parametrize("gap", ["before_acquire", "after_release"])
+def test_cache_only_steer_is_atomic_with_stop(scene, monkeypatch, gap):
+    """Cache-only Steer (no AGENT_INSTANCES entry) must never enqueue after Stop claims.
+
+    Steer selects the matching cached worker, then pauses at an unlocked gap
+    around its next stream-lock edge while Stop runs to completion. Exactly one
+    complete outcome is allowed: Steer already queued under the stream lock
+    before Stop claimed, or Stop claimed first and Steer reports ``stream_dead``
+    without ever calling ``agent.steer()`` on the cached worker.
+    """
+    lock, agent = scene
+    config.AGENT_INSTANCES.clear()
+    decoy = Mock(session_id="other-session")
+    config.SESSION_AGENT_CACHE["other-session"] = (decoy, "sig")
+    reached = threading.Event()
+    release = threading.Event()
+    state = {"steer_ident": None, "selected": None, "paused": False, "enqueue": None}
+    order = []
+
+    matches = streaming._cached_agent_matches_session
+
+    def select(candidate, sid):
+        result = matches(candidate, sid)
+        if threading.get_ident() == state["steer_ident"] and result:
+            state["selected"] = candidate
+        return result
+
+    def gate(edge):
+        # Pause the Steer thread once, at the requested unlocked edge, only
+        # after it has selected its cache candidate.
+        if (threading.get_ident() != state["steer_ident"]
+                or state["selected"] is None or state["paused"] or edge != gap):
+            return
+        state["paused"] = True
+        reached.set()
+        assert release.wait(5), "cache-only steer barrier timed out"
+
+    def enqueue(text):
+        state["enqueue"] = {
+            "held": lock.owner == threading.get_ident(),
+            "alive": "run" in config.STREAMS,
+            "phase": config.ACTIVE_RUNS["run"]["phase"],
+        }
+        order.append("steer")
+        return True
+
+    def interrupt(reason):
+        assert lock.owner != threading.get_ident(), "interrupt must be outside registry lock"
+        order.append("cancel")
+
+    def cache_only_steer():
+        state["steer_ident"] = threading.get_ident()
+        return steer()
+
+    agent.steer.side_effect = enqueue
+    agent.interrupt.side_effect = interrupt
+    monkeypatch.setattr(streaming, "_cached_agent_matches_session", select)
+    lock.before_acquire = lambda: gate("before_acquire")
+    lock.after_release = lambda: gate("after_release")
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        guidance = pool.submit(cache_only_steer)
+        try:
+            assert reached.wait(5), "steer never reached the requested gap"
+            assert lock.owner is None, "steer must not pause while holding the stream lock"
+            assert streaming.cancel_stream("run") is True
+        finally:
+            release.set()
+        result = guidance.result(timeout=5)
+    assert state["selected"] is agent, "the selected cache object must be the intended worker"
+    decoy.steer.assert_not_called()
+    assert config.ACTIVE_RUNS["run"]["phase"] == "cancelling"
+    assert "run" not in config.STREAMS
+    assert "run" not in config.AGENT_INSTANCES
+    agent.interrupt.assert_called_once()
+    if order == ["steer", "cancel"]:
+        # Steer queued before Stop claimed cancellation, under the stream lock.
+        assert result == {"accepted": True, "fallback": None, "stream_id": "run"}
+        agent.steer.assert_called_once_with("guidance")
+        assert state["enqueue"] == {"held": True, "alive": True, "phase": "running"}
+    else:
+        # Stop claimed first: the cached worker must not be steered afterwards.
+        assert order == ["cancel"]
+        assert result == {"accepted": False, "fallback": "stream_dead", "stream_id": None}
+        agent.steer.assert_not_called()