fix(cli): scope async delegation delivery to session

cli.py

@@ -9147,6 +9147,55 @@ def _get_goal_manager(self):
 
 
 
+    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``.
 
@@ -15185,18 +15234,7 @@ def process_loop():
                             # 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
@@ -15356,16 +15394,7 @@ def _restart_recording():
                         # 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
 

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",
+        }
+    )