fix(cli): scope async delegation delivery to session

This commit is contained in:
rabadaki 2026-07-14 15:47:22 +09:00 committed by Teknium
parent 51580e192b
commit 8ff3b67e6d
2 changed files with 127 additions and 22 deletions

73
cli.py
View file

@ -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

View file

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