fix(cli): stop new sessions adopting dead sessions' delegation completions (#64484)

cli.py

@@ -9148,6 +9148,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``.
 
@@ -15321,18 +15370,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
@@ -15492,16 +15530,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",
+        }
+    )

tests/tools/test_restored_delegation_ownership.py

@@ -0,0 +1,149 @@
+"""Regression coverage for #64484 — durable-restored delegation completions
+must never be adopted by a session that cannot positively prove ownership.
+
+Layers under test:
+1. ``restore_undelivered_completions`` stamps every restored event with
+   ``restored=True`` (in-memory only).
+2. ``ProcessRegistry.drain_notifications`` with NO filter (legacy
+   consume-everything CLI path) re-queues restored events instead of
+   consuming them.
+3. Same-process (non-restored) keyless events keep the legacy behavior.
+4. An owner with a matching session_key still receives its restored event.
+"""
+
+import json
+import queue
+
+from tools.process_registry import ProcessRegistry
+
+
+def _make_registry():
+    reg = ProcessRegistry.__new__(ProcessRegistry)
+    import threading
+
+    reg._running = {}
+    reg._finished = {}
+    reg._lock = threading.Lock()
+    reg.completion_queue = queue.Queue()
+    reg._completion_consumed = set()
+    reg._poll_observed = set()
+    return reg
+
+
+def _delegation_event(session_key="", restored=False, delegation_id="d1"):
+    evt = {
+        "type": "async_delegation",
+        "delegation_id": delegation_id,
+        "session_key": session_key,
+        "origin_ui_session_id": "",
+        "goal": "secret goal",
+        "status": "success",
+        "summary": "SECRET RESULT",
+        "api_calls": 3,
+        "duration_seconds": 1.5,
+        "dispatched_at": 1.0,
+        "completed_at": 2.0,
+    }
+    if restored:
+        evt["restored"] = True
+    return evt
+
+
+def test_restore_stamps_restored_flag(tmp_path, monkeypatch):
+    """Every durable completion re-enqueued at startup carries restored=True."""
+    import tools.async_delegation as ad
+
+    monkeypatch.setattr(ad, "_db_path", lambda: tmp_path / "async_delegations.db")
+    record = {
+        "delegation_id": "d-old",
+        "goal": "old goal",
+        "context": None,
+        "toolsets": None,
+        "role": "leaf",
+        "model": "m",
+        "session_key": "OLD_SESSION_A",
+        "origin_ui_session_id": "",
+        "parent_session_id": "OLD_SESSION_A",
+        "status": "running",
+        "dispatched_at": 1.0,
+        "completed_at": None,
+        "interrupt_fn": None,
+    }
+    ad._persist_dispatch(record)
+    evt = _delegation_event(session_key="OLD_SESSION_A", delegation_id="d-old")
+    ad._persist_completion(evt, {"summary": "SECRET RESULT"})
+
+    q = queue.Queue()
+    restored = ad.restore_undelivered_completions(q)
+    assert restored == 1
+    got = q.get_nowait()
+    assert got["restored"] is True
+    assert got["session_key"] == "OLD_SESSION_A"
+
+    # The stamp is in-memory only — the durable payload is unchanged.
+    with ad._connect() as conn:
+        row = conn.execute(
+            "SELECT event_json FROM async_delegations WHERE delegation_id='d-old'"
+        ).fetchone()
+    assert "restored" not in json.loads(row[0])
+
+
+def test_unfiltered_drain_never_consumes_restored_events():
+    """The legacy consume-everything branch must fail closed on restored events."""
+    reg = _make_registry()
+    reg.completion_queue.put(_delegation_event(session_key="DEAD_SESSION", restored=True))
+
+    results = reg.drain_notifications()  # no filter — legacy CLI post-turn shape
+
+    assert results == []
+    # Still queued for its real owner.
+    assert reg.completion_queue.qsize() == 1
+    assert reg.completion_queue.get_nowait()["session_key"] == "DEAD_SESSION"
+
+
+def test_unfiltered_drain_keeps_legacy_behavior_for_same_process_events():
+    """Non-restored keyless events (created by this process) are still consumed."""
+    reg = _make_registry()
+    reg.completion_queue.put(_delegation_event(session_key=""))
+
+    results = reg.drain_notifications()
+
+    assert len(results) == 1
+    assert results[0][0]["delegation_id"] == "d1"
+    assert reg.completion_queue.empty()
+
+
+def test_owner_session_key_drain_consumes_restored_event():
+    """The owning session (key match) still receives its restored completion."""
+    reg = _make_registry()
+    reg.completion_queue.put(_delegation_event(session_key="OWNER", restored=True))
+
+    results = reg.drain_notifications(session_key="OWNER")
+
+    assert len(results) == 1
+    assert results[0][0]["session_key"] == "OWNER"
+    assert reg.completion_queue.empty()
+
+
+def test_foreign_session_key_drain_requeues_restored_event():
+    """A different session's keyed drain must not claim the restored event."""
+    reg = _make_registry()
+    reg.completion_queue.put(_delegation_event(session_key="OWNER", restored=True))
+
+    results = reg.drain_notifications(session_key="SOMEONE_ELSE")
+
+    assert results == []
+    assert reg.completion_queue.qsize() == 1
+
+
+def test_owns_event_callback_beats_restored_flag():
+    """A positive-proof ownership callback consumes restored events it owns."""
+    reg = _make_registry()
+    reg.completion_queue.put(_delegation_event(session_key="OWNER", restored=True))
+
+    results = reg.drain_notifications(
+        owns_event=lambda e: e.get("session_key") == "OWNER"
+    )
+
+    assert len(results) == 1
+    assert reg.completion_queue.empty()

tools/async_delegation.py

@@ -264,7 +264,17 @@ def recover_abandoned_delegations() -> int:
 
 
 def restore_undelivered_completions(target_queue) -> int:
-    """Enqueue durable pending completions as fresh turns after process start."""
+    """Enqueue durable pending completions as fresh turns after process start.
+
+    Every restored event is stamped ``restored=True`` (in-memory only — the
+    stamp is added after the durable payload is deserialized and is never
+    persisted). Restored events originate from a *previous* process, so no
+    consumer in THIS process implicitly owns them: drain paths that run
+    without an ownership filter (the legacy single-session behavior) must
+    leave them queued for a consumer that can positively prove ownership,
+    otherwise a brand-new session adopts a dead session's delegation
+    results seconds after boot (#64484).
+    """
     recover_abandoned_delegations()
     with _DB_LOCK, _connect() as conn:
         rows = conn.execute(
@@ -273,7 +283,10 @@ def restore_undelivered_completions(target_queue) -> int:
                ORDER BY completed_at, delegation_id"""
         ).fetchall()
         for _delegation_id, payload in rows:
-            target_queue.put(json.loads(payload))
+            evt = json.loads(payload)
+            if isinstance(evt, dict):
+                evt["restored"] = True
+            target_queue.put(evt)
     return len(rows)
 
 

tools/delegate_tool.py

@@ -2844,6 +2844,20 @@ def _execute_and_aggregate() -> dict:
                     _session_key = _agent_session_id
         except Exception:
             _origin_ui_session_id = ""
+        if not _session_key:
+            # CLI (single-process) path: the approval contextvar is only bound
+            # during gateway/TUI turns and HERMES_SESSION_KEY is not in the CLI
+            # environment, so the key resolves empty here. Since #64240 the CLI
+            # drains completions through a positive-ownership filter keyed on
+            # the durable AIAgent.session_id — an empty session_key would fail
+            # closed and the CLI could never claim its own completions, while
+            # a restored foreign event with an empty key could leak into any
+            # unfiltered consumer (#64484). Stamp the parent's durable session
+            # id instead; compression rotations are handled on the drain side
+            # via resolve_resume_session_id lineage resolution.
+            _agent_session_id = str(getattr(parent_agent, "session_id", "") or "")
+            if _agent_session_id:
+                _session_key = _agent_session_id
         _parent_session_id = getattr(parent_agent, "session_id", None)
         _child_agents = [c for (_, _, c) in children]
 

tools/process_registry.py

@@ -1211,6 +1211,19 @@ def drain_notifications(
                     if evt_session_key != session_key:
                         requeue.append(evt)
                         continue
+                elif evt.get("restored"):
+                    # Legacy unfiltered drain (no ownership callback, no
+                    # session key). That behavior was safe when the in-memory
+                    # queue could only hold events created by this very
+                    # process — but durable restore (#63494) re-enqueues
+                    # completions from PREVIOUS processes at startup, so an
+                    # unfiltered consumer here would adopt a dead, unrelated
+                    # session's conversation payload (#64484). Fail closed:
+                    # leave restored events queued (still 'pending' on disk)
+                    # for a consumer that can positively prove ownership,
+                    # e.g. the owning session's --resume.
+                    requeue.append(evt)
+                    continue
             text = format_process_notification(evt)
             if text:
                 results.append((evt, text))