diff --git a/cli.py b/cli.py index 25cce4f95d05..6b2f00cf936a 100644 --- a/cli.py +++ b/cli.py @@ -9148,6 +9148,55 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): + def _owns_process_notification(self, event: dict) -> bool: + """Return whether this CLI session provably owns a delegation event. + + Delegations dispatched before context compression retain the original + session key, so resolve that key to its continuation before comparing. + Missing or foreign keys fail closed and remain queued for their owner. + """ + event_key = str(event.get("session_key") or "") + current_key = str(getattr(self, "session_id", "") or "") + if not event_key or not current_key: + return False + if event_key == current_key: + return True + try: + session_db = getattr(self, "_session_db", None) + resolved_key = ( + session_db.resolve_resume_session_id(event_key) + if session_db is not None + else event_key + ) or event_key + except Exception: + resolved_key = event_key + return str(resolved_key) == current_key + + def _drain_process_notifications(self, consumer: str) -> None: + """Queue background notifications owned by this visible CLI session. + + ``process_registry`` restores durable delegation completions into every + process using the same Hermes profile. Always pass this CLI's stable + session identity when draining so another window cannot claim and mark + delivered a completion that belongs to this one. + """ + from tools.process_registry import process_registry + from tools.async_delegation import ( + claim_event_delivery, + complete_event_delivery, + ) + + session_key = getattr(self, "session_id", "") or "" + for event, synthetic_message in process_registry.drain_notifications( + session_key=session_key, + owns_event=self._owns_process_notification, + ): + claim = claim_event_delivery(event, consumer) + if claim is None: + continue + self._pending_input.put(synthetic_message) + complete_event_delivery(event, claim) + def _drain_interrupt_queue_to_pending_input(self) -> None: """Move stray messages from ``_interrupt_queue`` into ``_pending_input``. @@ -15321,18 +15370,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): # Check for background process notifications (completions # and watch pattern matches) while agent is idle. try: - from tools.process_registry import process_registry - from tools.approval import get_current_session_key - _drain_sk = get_current_session_key(default="") - for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk): - from tools.async_delegation import ( - claim_event_delivery, complete_event_delivery, - ) - _claim = claim_event_delivery(_evt, "cli-idle") - if _claim is None: - continue - self._pending_input.put(_synth) - complete_event_delivery(_evt, _claim) + self._drain_process_notifications("cli-idle") except Exception: pass continue @@ -15492,16 +15530,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): # Drain process notifications (completions + watch matches) # that arrived while the agent was running. try: - from tools.process_registry import process_registry - for _evt, _synth in process_registry.drain_notifications(): - from tools.async_delegation import ( - claim_event_delivery, complete_event_delivery, - ) - _claim = claim_event_delivery(_evt, "cli-post-turn") - if _claim is None: - continue - self._pending_input.put(_synth) - complete_event_delivery(_evt, _claim) + self._drain_process_notifications("cli-post-turn") except Exception: pass # Non-fatal — don't break the main loop diff --git a/tests/cli/test_cli_async_delegation_delivery.py b/tests/cli/test_cli_async_delegation_delivery.py new file mode 100644 index 000000000000..b970aca56a7f --- /dev/null +++ b/tests/cli/test_cli_async_delegation_delivery.py @@ -0,0 +1,76 @@ +"""Regression coverage for CLI async-delegation completion ownership.""" + +import queue + +from cli import HermesCLI + + +def test_cli_completion_drain_uses_visible_session_identity(monkeypatch): + """A CLI window must not claim another window's restored completion.""" + cli = HermesCLI.__new__(HermesCLI) + cli.session_id = "visible-session" + cli._pending_input = queue.Queue() + + event = { + "type": "async_delegation", + "delegation_id": "deleg_visible", + "session_key": "visible-session", + } + calls = [] + + class FakeRegistry: + def drain_notifications(self, *, session_key="", owns_event=None): + calls.append((session_key, owns_event(event))) + return [(event, "completion payload")] + + claimed = [] + completed = [] + + monkeypatch.setattr( + "tools.process_registry.process_registry", + FakeRegistry(), + ) + monkeypatch.setattr( + "tools.async_delegation.claim_event_delivery", + lambda evt, consumer: claimed.append((evt, consumer)) or "claim-token", + ) + monkeypatch.setattr( + "tools.async_delegation.complete_event_delivery", + lambda evt, token: completed.append((evt, token)), + ) + + cli._drain_process_notifications("cli-idle") + + assert calls == [("visible-session", True)] + assert cli._pending_input.get_nowait() == "completion payload" + assert claimed == [(event, "cli-idle")] + assert completed == [(event, "claim-token")] + + +def test_cli_completion_ownership_rejects_foreign_session(): + cli = HermesCLI.__new__(HermesCLI) + cli.session_id = "visible-session" + cli._session_db = None + + assert not cli._owns_process_notification( + {"type": "async_delegation", "session_key": "foreign-session"} + ) + + +def test_cli_completion_ownership_accepts_compression_lineage(): + cli = HermesCLI.__new__(HermesCLI) + cli.session_id = "visible-session" + + class FakeSessionDB: + def resolve_resume_session_id(self, session_id): + assert session_id == "pre-compression-session" + return "visible-session" + + cli._session_db = FakeSessionDB() + + assert cli._owns_process_notification( + { + "type": "async_delegation", + "session_key": "pre-compression-session", + } + )