fix(steer): require positive ownership on the cache-only path

The cache-only Steer branch (no AGENT_INSTANCES worker) only rejected
*conflicting* ownership, so a missing stream owner or a missing
active-run session_id was treated as non-conflicting and let
agent.steer() run and report accepted without proving the requesting
session owns the run.

Fail closed instead: both the stream owner and the active-run session
must be present and strictly equal to the requesting session (backend
and phase are still revalidated) under the same STREAMS_LOCK ->
ACTIVE_RUNS_LOCK edge; any absent identity returns stream_dead without
calling agent.steer(). No cache eviction, no agent closing, HTTP write
still outside the locks.

Regressions: missing owner, missing/empty run session, both missing,
missing run entry -> stream_dead + steer never called; both present and
equal -> accepted, cache untouched. Legacy happy-path fixtures now
register the ownership they rely on.

api/streaming.py

@@ -13686,15 +13686,20 @@ def _handle_chat_steer(handler, body: dict) -> bool:
     # 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.
+    #
+    # Ownership must be proven positively, not merely unrefuted: BOTH the
+    # stream owner AND the active-run session must equal the requesting
+    # session. Missing metadata is ambiguous and fails closed (stream_dead);
+    # an unfenced cache object is never steered on absent ownership.
     result = {"accepted": False, "fallback": "stream_dead", "stream_id": None}
     with _cfg.STREAMS_LOCK:
         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":
+            owned_stream = bool(owner) and owner == sid
+            owned_run = bool(run.get("session_id")) and run["session_id"] == sid
+            if owned_stream and owned_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",

tests/test_real_steer.py

@@ -101,9 +101,17 @@ def _captured_status(handler):
 class TestHandleChatSteerHappyPath:
     """Endpoint accepts text and calls agent.steer() when all gates pass."""
 
-    def test_accepts_when_agent_cached_and_running(self, _clear_caches):
+    def test_accepts_when_agent_cached_and_running(self, _clear_caches, monkeypatch):
         from api.streaming import _handle_chat_steer
-        from api.config import SESSION_AGENT_CACHE, SESSION_AGENT_CACHE_LOCK, STREAMS, STREAMS_LOCK
+        from api import config
+        from api.config import (
+            ACTIVE_RUNS,
+            ACTIVE_RUNS_LOCK,
+            SESSION_AGENT_CACHE,
+            SESSION_AGENT_CACHE_LOCK,
+            STREAMS,
+            STREAMS_LOCK,
+        )
         sid, stream_id = "sid_happy", "stream_happy"
         agent = MagicMock()
         agent.steer = MagicMock(return_value=True)
@@ -112,6 +120,11 @@ class TestHandleChatSteerHappyPath:
         with STREAMS_LOCK:
             import queue as _q
             STREAMS[stream_id] = _q.Queue()
+        # Cache-only Steer requires positive ownership: stream owner AND
+        # active-run session must both equal the requesting session.
+        monkeypatch.setattr(config, "STREAM_SESSION_OWNERS", {stream_id: sid})
+        with ACTIVE_RUNS_LOCK:
+            ACTIVE_RUNS[stream_id] = {"session_id": sid, "backend": "legacy", "phase": "running"}
 
         sess = MagicMock()
         sess.active_stream_id = stream_id
@@ -228,10 +241,18 @@ class TestHandleChatSteerFallbacks:
         assert body["fallback"] == "stream_dead"
         agent.steer.assert_not_called()
 
-    def test_steer_raises(self, _clear_caches):
+    def test_steer_raises(self, _clear_caches, monkeypatch):
         """If agent.steer() raises, return steer_error rather than 500."""
         from api.streaming import _handle_chat_steer
-        from api.config import SESSION_AGENT_CACHE, SESSION_AGENT_CACHE_LOCK, STREAMS, STREAMS_LOCK
+        from api import config
+        from api.config import (
+            ACTIVE_RUNS,
+            ACTIVE_RUNS_LOCK,
+            SESSION_AGENT_CACHE,
+            SESSION_AGENT_CACHE_LOCK,
+            STREAMS,
+            STREAMS_LOCK,
+        )
         sid, stream_id = "sid_throws", "stream_throws"
         agent = MagicMock()
         agent.steer = MagicMock(side_effect=RuntimeError("boom"))
@@ -240,6 +261,9 @@ class TestHandleChatSteerFallbacks:
         with STREAMS_LOCK:
             import queue as _q
             STREAMS[stream_id] = _q.Queue()
+        monkeypatch.setattr(config, "STREAM_SESSION_OWNERS", {stream_id: sid})
+        with ACTIVE_RUNS_LOCK:
+            ACTIVE_RUNS[stream_id] = {"session_id": sid, "backend": "legacy", "phase": "running"}
         sess = MagicMock()
         sess.active_stream_id = stream_id
         with patch("api.streaming.get_session", return_value=sess):

tests/test_steer_active_run_identity.py

@@ -118,6 +118,45 @@ def test_cache_only_mismatch_does_not_close_an_agent_owned_by_another_run(scene)
     other.steer.assert_not_called()
 
 
+@pytest.mark.parametrize("ownership", [
+    "both", "missing-owner", "missing-run-session", "empty-run-session",
+    "missing-both", "missing-run-entry",
+])
+def test_cache_only_steer_requires_positive_stream_and_run_ownership(scene, ownership):
+    """Cache-only Steer must prove ownership, not merely fail to refute it.
+
+    With no registered worker and a matching cached agent on a live stream, a
+    missing stream owner or a missing active-run session is ambiguous and must
+    fail closed with ``stream_dead`` without calling ``agent.steer()``. Only
+    both identities present and equal to the requesting session may enqueue.
+    """
+    agent, other, _ = scene
+    agent.session_id = "original"
+    config.AGENT_INSTANCES.pop("run")  # Cache-only compatibility path.
+    if ownership in ("missing-owner", "missing-both"):
+        config.STREAM_SESSION_OWNERS.pop("run")
+    if ownership in ("missing-run-session", "missing-both"):
+        config.ACTIVE_RUNS["run"].pop("session_id")
+    elif ownership == "empty-run-session":
+        config.ACTIVE_RUNS["run"]["session_id"] = ""
+    elif ownership == "missing-run-entry":
+        config.ACTIVE_RUNS.pop("run")
+    before = dict(config.SESSION_AGENT_CACHE)
+    assert "run" in config.STREAMS
+    result = steer()
+    if ownership == "both":
+        assert result == {"accepted": True, "fallback": None, "stream_id": "run"}
+        agent.steer.assert_called_once_with("updated guidance")
+    else:
+        assert result == {"accepted": False, "fallback": "stream_dead", "stream_id": None}
+        agent.steer.assert_not_called()
+    other.steer.assert_not_called()
+    agent.interrupt.assert_not_called()
+    agent._session_db.close.assert_not_called()
+    assert config.SESSION_AGENT_CACHE == before
+    assert "run" in config.STREAMS
+
+
 def test_http_response_is_written_after_stream_lock_release(scene, monkeypatch):
     from api import helpers