diff --git a/AGENTS.md b/AGENTS.md index 5dc841ba..badaff7a 100644 --- a/AGENTS.md +++ b/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/`). diff --git a/api/streaming.py b/api/streaming.py index f7fa9f96..ef50fd60 100644 --- a/api/streaming.py +++ b/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: diff --git a/tests/test_steer_cancel_ordering.py b/tests/test_steer_cancel_ordering.py index 1596686d..b6750b16 100644 --- a/tests/test_steer_cancel_ordering.py +++ b/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()