mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-23 16:36:23 +00:00
417 lines
14 KiB
Python
417 lines
14 KiB
Python
"""Gateway-side session binding for async delegations (#57498, #55578).
|
|
|
|
Three invariants on the messaging-gateway surface, mirroring the TUI rules:
|
|
|
|
1. Completions are pinned to the spawning session (contributor commit).
|
|
2. A dead/ended spawning session is never resurrected: the injection is
|
|
dropped, fail-closed (never rerouted to the peer's current session).
|
|
3. /new interrupts the old conversation's in-flight async delegations.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import tools.async_delegation as ad
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_async_delegation():
|
|
ad._reset_for_tests()
|
|
yield
|
|
ad._reset_for_tests()
|
|
|
|
|
|
def _seed_record(delegation_id, session_key="", parent_session_id="", status="running"):
|
|
fn = MagicMock()
|
|
with ad._records_lock:
|
|
ad._records[delegation_id] = {
|
|
"delegation_id": delegation_id,
|
|
"status": status,
|
|
"session_key": session_key,
|
|
"parent_session_id": parent_session_id,
|
|
"interrupt_fn": fn,
|
|
}
|
|
return fn
|
|
|
|
|
|
class TestInterruptForSessionByParentId:
|
|
def test_parent_session_id_selector(self):
|
|
mine = _seed_record("d1", session_key="agent:main:telegram:dm:1", parent_session_id="sess_old")
|
|
other = _seed_record("d2", session_key="agent:main:telegram:dm:2", parent_session_id="sess_other")
|
|
n = ad.interrupt_for_session(parent_session_id="sess_old")
|
|
assert n == 1
|
|
mine.assert_called_once()
|
|
other.assert_not_called()
|
|
|
|
def test_reset_interrupts_by_key_and_parent(self):
|
|
"""A /new reset passes both selectors — either match claims the record."""
|
|
by_key = _seed_record("d1", session_key="agent:main:telegram:dm:1", parent_session_id="")
|
|
by_parent = _seed_record("d2", session_key="", parent_session_id="sess_old")
|
|
unrelated = _seed_record("d3", session_key="other", parent_session_id="other")
|
|
n = ad.interrupt_for_session(
|
|
session_key="agent:main:telegram:dm:1",
|
|
parent_session_id="sess_old",
|
|
reason="session_reset",
|
|
)
|
|
assert n == 2
|
|
by_key.assert_called_once()
|
|
by_parent.assert_called_once()
|
|
unrelated.assert_not_called()
|
|
|
|
|
|
class TestGatewayPinningFailsClosed:
|
|
"""The gateway must follow only verified compression continuations."""
|
|
|
|
@staticmethod
|
|
def _entry(session_id):
|
|
from datetime import datetime
|
|
|
|
from gateway.config import Platform
|
|
from gateway.session import SessionEntry
|
|
|
|
return SessionEntry(
|
|
session_key="agent:main:telegram:group:-100:4",
|
|
session_id=session_id,
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
platform=Platform.TELEGRAM,
|
|
chat_type="group",
|
|
)
|
|
|
|
def _make_runner(
|
|
self,
|
|
rows,
|
|
*,
|
|
compression_tip=None,
|
|
compression_error=None,
|
|
switched_entry=None,
|
|
):
|
|
from gateway.run import GatewayRunner
|
|
from gateway.session import AsyncSessionStore
|
|
|
|
runner = object.__new__(GatewayRunner)
|
|
db = MagicMock()
|
|
db.get_session = AsyncMock(side_effect=lambda session_id: rows.get(session_id))
|
|
db.get_compression_tip = AsyncMock(
|
|
return_value=compression_tip,
|
|
side_effect=compression_error,
|
|
)
|
|
runner._session_db = db
|
|
runner.session_store = MagicMock()
|
|
runner.session_store.switch_session = MagicMock(return_value=switched_entry)
|
|
runner.session_store.advance_compression_session = MagicMock(
|
|
return_value=switched_entry
|
|
)
|
|
runner._async_session_store = AsyncSessionStore(runner.session_store)
|
|
return runner
|
|
|
|
@staticmethod
|
|
def _assert_no_route_change(runner):
|
|
getattr(runner.session_store, "switch_session").assert_not_called()
|
|
getattr(
|
|
runner.session_store, "advance_compression_session"
|
|
).assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_live_spawning_session_stays_pinned(self):
|
|
current = self._entry("sess_live")
|
|
runner = self._make_runner(
|
|
{"sess_live": {"id": "sess_live", "ended_at": None}}
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_live"
|
|
)
|
|
|
|
assert resolved is current
|
|
self._assert_no_route_change(runner)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_live_spawning_session_rebinds_from_different_route(self):
|
|
current = self._entry("sess_current")
|
|
pinned = self._entry("sess_live")
|
|
runner = self._make_runner(
|
|
{"sess_live": {"id": "sess_live", "ended_at": None}},
|
|
switched_entry=pinned,
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_live"
|
|
)
|
|
|
|
assert resolved is pinned
|
|
getattr(runner.session_store, "switch_session").assert_called_once_with(
|
|
current.session_key, "sess_live"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_compression_ended_parent_drops(self):
|
|
current = self._entry("sess_old")
|
|
runner = self._make_runner(
|
|
{
|
|
"sess_old": {
|
|
"id": "sess_old",
|
|
"ended_at": "2026-07-08T00:00:00",
|
|
"end_reason": "session_reset",
|
|
}
|
|
}
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_old"
|
|
)
|
|
|
|
assert resolved is None
|
|
self._assert_no_route_change(runner)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compression_parent_advances_stale_route_to_live_tip(self):
|
|
current = self._entry("sess_parent")
|
|
tip = self._entry("sess_tip")
|
|
runner = self._make_runner(
|
|
{
|
|
"sess_parent": {
|
|
"id": "sess_parent",
|
|
"ended_at": "2026-07-08T00:00:00",
|
|
"end_reason": "compression",
|
|
},
|
|
"sess_tip": {
|
|
"id": "sess_tip",
|
|
"ended_at": None,
|
|
"parent_session_id": "sess_parent",
|
|
},
|
|
},
|
|
compression_tip="sess_tip",
|
|
switched_entry=tip,
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_parent"
|
|
)
|
|
|
|
assert resolved is tip
|
|
getattr(
|
|
runner.session_store, "advance_compression_session"
|
|
).assert_called_once_with(current.session_key, "sess_parent", "sess_tip")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compression_cas_losing_to_new_drops(self):
|
|
current = self._entry("sess_parent")
|
|
runner = self._make_runner(
|
|
{
|
|
"sess_parent": {
|
|
"id": "sess_parent",
|
|
"ended_at": "2026-07-08T00:00:00",
|
|
"end_reason": "compression",
|
|
},
|
|
"sess_tip": {
|
|
"id": "sess_tip",
|
|
"ended_at": None,
|
|
"parent_session_id": "sess_parent",
|
|
},
|
|
},
|
|
compression_tip="sess_tip",
|
|
switched_entry=None,
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_parent"
|
|
)
|
|
|
|
assert resolved is None
|
|
getattr(
|
|
runner.session_store, "advance_compression_session"
|
|
).assert_called_once_with(current.session_key, "sess_parent", "sess_tip")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_intermediate_compression_route_advances_to_same_live_tip(self):
|
|
current = self._entry("sess_middle")
|
|
tip = self._entry("sess_tip")
|
|
runner = self._make_runner(
|
|
{
|
|
"sess_parent": {
|
|
"id": "sess_parent",
|
|
"ended_at": "2026-07-08T00:00:00",
|
|
"end_reason": "compression",
|
|
},
|
|
"sess_middle": {
|
|
"id": "sess_middle",
|
|
"ended_at": "2026-07-08T00:01:00",
|
|
"end_reason": "compression",
|
|
"parent_session_id": "sess_parent",
|
|
},
|
|
"sess_tip": {
|
|
"id": "sess_tip",
|
|
"ended_at": None,
|
|
"parent_session_id": "sess_middle",
|
|
},
|
|
},
|
|
compression_tip="sess_tip",
|
|
switched_entry=tip,
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_parent"
|
|
)
|
|
|
|
assert resolved is tip
|
|
getattr(
|
|
runner.session_store, "advance_compression_session"
|
|
).assert_called_once_with(current.session_key, "sess_middle", "sess_tip")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compression_parent_follows_real_sessiondb_lineage(self, tmp_path):
|
|
from gateway.run import GatewayRunner
|
|
from gateway.session import AsyncSessionStore
|
|
from hermes_state import AsyncSessionDB, SessionDB
|
|
|
|
session_db = SessionDB(db_path=tmp_path / "state.db")
|
|
session_db.create_session("sess_parent", source="telegram")
|
|
session_db.end_session("sess_parent", end_reason="compression")
|
|
session_db.create_session(
|
|
"sess_tip",
|
|
source="telegram",
|
|
parent_session_id="sess_parent",
|
|
)
|
|
|
|
current = self._entry("sess_parent")
|
|
tip = self._entry("sess_tip")
|
|
runner = object.__new__(GatewayRunner)
|
|
runner._session_db = AsyncSessionDB(session_db)
|
|
runner.session_store = MagicMock()
|
|
runner.session_store.switch_session = MagicMock(return_value=tip)
|
|
runner.session_store.advance_compression_session = MagicMock(return_value=tip)
|
|
runner._async_session_store = AsyncSessionStore(runner.session_store)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_parent"
|
|
)
|
|
|
|
assert resolved is tip
|
|
getattr(
|
|
runner.session_store, "advance_compression_session"
|
|
).assert_called_once_with(current.session_key, "sess_parent", "sess_tip")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ended_compression_tip_drops(self):
|
|
current = self._entry("sess_parent")
|
|
runner = self._make_runner(
|
|
{
|
|
"sess_parent": {
|
|
"id": "sess_parent",
|
|
"ended_at": "2026-07-08T00:00:00",
|
|
"end_reason": "compression",
|
|
},
|
|
"sess_tip": {
|
|
"id": "sess_tip",
|
|
"ended_at": "2026-07-08T00:01:00",
|
|
"end_reason": "session_reset",
|
|
"parent_session_id": "sess_parent",
|
|
},
|
|
},
|
|
compression_tip="sess_tip",
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_parent"
|
|
)
|
|
|
|
assert resolved is None
|
|
self._assert_no_route_change(runner)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compression_lookup_failure_drops(self):
|
|
current = self._entry("sess_parent")
|
|
runner = self._make_runner(
|
|
{
|
|
"sess_parent": {
|
|
"id": "sess_parent",
|
|
"ended_at": "2026-07-08T00:00:00",
|
|
"end_reason": "compression",
|
|
}
|
|
},
|
|
compression_error=RuntimeError("db unavailable"),
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_parent"
|
|
)
|
|
|
|
assert resolved is None
|
|
self._assert_no_route_change(runner)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compression_parent_accepts_already_current_tip(self):
|
|
current = self._entry("sess_tip")
|
|
runner = self._make_runner(
|
|
{
|
|
"sess_parent": {
|
|
"id": "sess_parent",
|
|
"ended_at": "2026-07-08T00:00:00",
|
|
"end_reason": "compression",
|
|
},
|
|
"sess_tip": {
|
|
"id": "sess_tip",
|
|
"ended_at": None,
|
|
"parent_session_id": "sess_parent",
|
|
},
|
|
},
|
|
compression_tip="sess_tip",
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_parent"
|
|
)
|
|
|
|
assert resolved is current
|
|
self._assert_no_route_change(runner)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compression_parent_does_not_override_new_route(self):
|
|
current = self._entry("sess_after_new")
|
|
runner = self._make_runner(
|
|
{
|
|
"sess_parent": {
|
|
"id": "sess_parent",
|
|
"ended_at": "2026-07-08T00:00:00",
|
|
"end_reason": "compression",
|
|
},
|
|
"sess_tip": {
|
|
"id": "sess_tip",
|
|
"ended_at": None,
|
|
"parent_session_id": "sess_parent",
|
|
},
|
|
},
|
|
compression_tip="sess_tip",
|
|
)
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_parent"
|
|
)
|
|
|
|
assert resolved is None
|
|
self._assert_no_route_change(runner)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_spawning_session_drops(self):
|
|
current = self._entry("sess_current")
|
|
runner = self._make_runner({})
|
|
|
|
resolved = await runner._resolve_async_delegation_session(
|
|
current, "sess_gone"
|
|
)
|
|
|
|
assert resolved is None
|
|
self._assert_no_route_change(runner)
|
|
|
|
|
|
class TestResetHandlerInterruptsDelegations:
|
|
def test_reset_command_calls_interrupt_for_session(self):
|
|
"""The /new handler must sever the old conversation's delegations."""
|
|
import inspect
|
|
from gateway import slash_commands
|
|
|
|
src = inspect.getsource(slash_commands.GatewaySlashCommandsMixin._handle_reset_command)
|
|
assert "interrupt_for_session" in src
|
|
assert "session_reset" in src
|