From 20c16e91671f4fd25d43ae2e3a0fcc306f55a093 Mon Sep 17 00:00:00 2001 From: Anthony Ruiz Date: Thu, 17 Sep 2026 22:27:12 +0000 Subject: [PATCH] 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 | 11 +++++-- tests/test_real_steer.py | 32 +++++++++++++++++--- tests/test_steer_active_run_identity.py | 39 +++++++++++++++++++++++++ 3 files changed, 75 insertions(+), 7 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index ef50fd60..88e5c6e0 100644 --- a/api/streaming.py +++ b/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", diff --git a/tests/test_real_steer.py b/tests/test_real_steer.py index 42c971bb..e131f166 100644 --- a/tests/test_real_steer.py +++ b/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): diff --git a/tests/test_steer_active_run_identity.py b/tests/test_steer_active_run_identity.py index 19de2f51..50a25359 100644 --- a/tests/test_steer_active_run_identity.py +++ b/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