mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-10 13:31:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4
hermes-agent/tests/gateway/test_agent_cache.py
Hermes Trismegistus 90b618f48a fix(gateway): keep idle cached agents alive until session actually expires 
The idle-TTL sweep (_sweep_idle_cached_agents) was evicting agents
as soon as they passed _AGENT_CACHE_IDLE_TTL_SECS, even when the
session hadn't expired yet. In daily-reset mode the reset can fire
hours after the last user message — evicting the agent early means
the session-expiry watcher has no agent in cache to call
on_session_end() with, so memory providers miss the live transcript.

Now the sweep checks the session store before evicting: if the
session still exists and hasn't expired, the agent stays in cache
so the expiry watcher can tear it down properly later.
When the session store is unavailable or throws, falls back to the
original eviction behavior (safe default).

Fixes: #11205
2026-07-03 03:46:43 +05:30

1956 lines
80 KiB
Python
Raw Blame History

"""Integration tests for gateway AIAgent caching.
Verifies that the agent cache correctly:
- Reuses agents across messages (same config → same instance)
- Rebuilds agents when config changes (model, provider, toolsets)
- Updates reasoning_config in-place without rebuilding
- Evicts on session reset
- Evicts on fallback activation
- Preserves frozen system prompt across turns
"""
import threading
from unittest.mock import MagicMock, patch
import pytest
def _make_runner():
"""Create a minimal GatewayRunner with just the cache infrastructure."""
from gateway.run import GatewayRunner
runner = GatewayRunner.__new__(GatewayRunner)
runner._agent_cache = {}
runner._agent_cache_lock = threading.Lock()
return runner
class TestAgentConfigSignature:
"""Config signature produces stable, distinct keys."""
def test_same_config_same_signature(self):
from gateway.run import GatewayRunner
runtime = {"api_key": "sk-test12345678", "base_url": "https://openrouter.ai/api/v1",
"provider": "openrouter", "api_mode": "chat_completions"}
sig1 = GatewayRunner._agent_config_signature("claude-sonnet-4", runtime, ["hermes-telegram"], "")
sig2 = GatewayRunner._agent_config_signature("claude-sonnet-4", runtime, ["hermes-telegram"], "")
assert sig1 == sig2
def test_model_change_different_signature(self):
from gateway.run import GatewayRunner
runtime = {"api_key": "sk-test12345678", "base_url": "https://openrouter.ai/api/v1",
"provider": "openrouter"}
sig1 = GatewayRunner._agent_config_signature("claude-sonnet-4", runtime, ["hermes-telegram"], "")
sig2 = GatewayRunner._agent_config_signature("claude-opus-4.6", runtime, ["hermes-telegram"], "")
assert sig1 != sig2
def test_same_token_prefix_different_full_token_changes_signature(self):
"""Tokens sharing a JWT-style prefix must not collide."""
from gateway.run import GatewayRunner
rt1 = {
"api_key": "eyJhbGci.token-for-account-a",
"base_url": "https://chatgpt.com/backend-api/codex",
"provider": "openai-codex",
"api_mode": "codex_responses",
}
rt2 = {
"api_key": "eyJhbGci.token-for-account-b",
"base_url": "https://chatgpt.com/backend-api/codex",
"provider": "openai-codex",
"api_mode": "codex_responses",
}
assert rt1["api_key"][:8] == rt2["api_key"][:8]
sig1 = GatewayRunner._agent_config_signature("gpt-5.3-codex", rt1, ["hermes-telegram"], "")
sig2 = GatewayRunner._agent_config_signature("gpt-5.3-codex", rt2, ["hermes-telegram"], "")
assert sig1 != sig2
def test_provider_change_different_signature(self):
from gateway.run import GatewayRunner
rt1 = {"api_key": "sk-test12345678", "base_url": "https://openrouter.ai/api/v1", "provider": "openrouter"}
rt2 = {"api_key": "sk-test12345678", "base_url": "https://api.anthropic.com", "provider": "anthropic"}
sig1 = GatewayRunner._agent_config_signature("claude-sonnet-4", rt1, ["hermes-telegram"], "")
sig2 = GatewayRunner._agent_config_signature("claude-sonnet-4", rt2, ["hermes-telegram"], "")
assert sig1 != sig2
def test_toolset_change_different_signature(self):
from gateway.run import GatewayRunner
runtime = {"api_key": "sk-test12345678", "base_url": "https://openrouter.ai/api/v1", "provider": "openrouter"}
sig1 = GatewayRunner._agent_config_signature("claude-sonnet-4", runtime, ["hermes-telegram"], "")
sig2 = GatewayRunner._agent_config_signature("claude-sonnet-4", runtime, ["hermes-discord"], "")
assert sig1 != sig2
def test_reasoning_not_in_signature(self):
"""Reasoning config is set per-message, not part of the signature."""
from gateway.run import GatewayRunner
runtime = {"api_key": "sk-test12345678", "base_url": "https://openrouter.ai/api/v1", "provider": "openrouter"}
# Same config — signature should be identical regardless of what
# reasoning_config the caller might have (it's not passed in)
sig1 = GatewayRunner._agent_config_signature("claude-sonnet-4", runtime, ["hermes-telegram"], "")
sig2 = GatewayRunner._agent_config_signature("claude-sonnet-4", runtime, ["hermes-telegram"], "")
assert sig1 == sig2
# ---------------------------------------------------------------
# cache_keys (compression/context config cache-busting)
# ---------------------------------------------------------------
def test_cache_keys_default_omitted_matches_empty(self):
"""Omitted cache_keys must produce the same signature as empty {}."""
from gateway.run import GatewayRunner
runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
sig_omitted = GatewayRunner._agent_config_signature("m", runtime, [], "")
sig_empty = GatewayRunner._agent_config_signature("m", runtime, [], "", cache_keys={})
sig_none = GatewayRunner._agent_config_signature("m", runtime, [], "", cache_keys=None)
assert sig_omitted == sig_empty == sig_none
def test_context_length_change_busts_cache(self):
"""Editing model.context_length in config must produce a new signature."""
from gateway.run import GatewayRunner
runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
sig1 = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"model.context_length": 200_000},
)
sig2 = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"model.context_length": 400_000},
)
assert sig1 != sig2
def test_max_tokens_change_busts_cache(self):
"""Editing model.max_tokens in config must produce a new signature."""
from gateway.run import GatewayRunner
runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
sig1 = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"model.max_tokens": 4096},
)
sig2 = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"model.max_tokens": 8192},
)
assert sig1 != sig2
def test_compression_threshold_change_busts_cache(self):
from gateway.run import GatewayRunner
runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
sig1 = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"compression.threshold": 0.50},
)
sig2 = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"compression.threshold": 0.75},
)
assert sig1 != sig2
def test_compression_enabled_toggle_busts_cache(self):
from gateway.run import GatewayRunner
runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
sig_on = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"compression.enabled": True},
)
sig_off = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"compression.enabled": False},
)
assert sig_on != sig_off
def test_cache_keys_key_order_does_not_matter(self):
"""Signature must be stable regardless of dict key insertion order."""
from gateway.run import GatewayRunner
runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
sig_a = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"model.context_length": 200_000, "compression.threshold": 0.5},
)
sig_b = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys={"compression.threshold": 0.5, "model.context_length": 200_000},
)
assert sig_a == sig_b
def test_tool_registry_generation_change_busts_cache(self):
"""MCP reloads mutate the tool registry, so cached agents must rebuild."""
from gateway.run import GatewayRunner
runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
sig_before = GatewayRunner._agent_config_signature(
"m", runtime, ["telegram"], "",
cache_keys={"tools.registry_generation": 10},
)
sig_after = GatewayRunner._agent_config_signature(
"m", runtime, ["telegram"], "",
cache_keys={"tools.registry_generation": 11},
)
assert sig_before != sig_after
class TestExtractCacheBustingConfig:
"""Verify _extract_cache_busting_config pulls the documented subset of
config values that must invalidate the cached agent on change."""
def test_reads_model_context_length(self):
from gateway.run import GatewayRunner
out = GatewayRunner._extract_cache_busting_config(
{
"model": {
"context_length": 272_000,
"max_tokens": 4096,
"provider": "openrouter",
}
}
)
assert out["model.context_length"] == 272_000
assert out["model.max_tokens"] == 4096
def test_reads_compression_subkeys(self):
from gateway.run import GatewayRunner
out = GatewayRunner._extract_cache_busting_config(
{
"compression": {
"enabled": False,
"threshold": 0.6,
"target_ratio": 0.3,
"protect_last_n": 25,
"some_other_key": "ignored",
}
}
)
assert out["compression.enabled"] is False
assert out["compression.threshold"] == 0.6
assert out["compression.target_ratio"] == 0.3
assert out["compression.protect_last_n"] == 25
def test_missing_keys_yield_none(self):
"""Absent config keys must produce None values (still contribute to signature)."""
from gateway.run import GatewayRunner
out = GatewayRunner._extract_cache_busting_config({})
# Every documented cache-busting key must be present, even if None
for section, key in GatewayRunner._CACHE_BUSTING_CONFIG_KEYS:
assert f"{section}.{key}" in out
assert out[f"{section}.{key}"] is None
def test_non_dict_section_treated_as_missing(self):
from gateway.run import GatewayRunner
# compression is a string — should not crash, all compression.* keys None
out = GatewayRunner._extract_cache_busting_config(
{"compression": "broken", "model": {"context_length": 100_000}}
)
assert out["compression.enabled"] is None
assert out["compression.threshold"] is None
assert out["model.context_length"] == 100_000
def test_none_config_is_safe(self):
from gateway.run import GatewayRunner
out = GatewayRunner._extract_cache_busting_config(None)
for section, key in GatewayRunner._CACHE_BUSTING_CONFIG_KEYS:
assert out[f"{section}.{key}"] is None
assert "tools.registry_generation" in out
def test_extract_includes_live_tool_registry_generation(self, monkeypatch):
from gateway.run import GatewayRunner
from tools.registry import registry
monkeypatch.setattr(registry, "_generation", 12345)
out = GatewayRunner._extract_cache_busting_config({})
assert out["tools.registry_generation"] == 12345
def test_skips_honcho_config_read_when_provider_is_not_honcho(self, monkeypatch):
"""Non-Honcho gateways must not read/parse honcho.json on every message."""
from gateway.run import GatewayRunner
called = False
def _boom():
nonlocal called
called = True
raise AssertionError("should not read Honcho config")
monkeypatch.setattr(GatewayRunner, "_extract_honcho_cache_busting_config", _boom)
out = GatewayRunner._extract_cache_busting_config({"memory": {"provider": "mem0"}})
assert called is False
assert out["honcho.peer_name"] is None
assert out["honcho.user_peer_aliases"] is None
def test_reads_honcho_config_only_when_provider_is_honcho(self, monkeypatch):
from gateway.run import GatewayRunner
calls = []
def _fake():
calls.append(True)
return {
"honcho.peer_name": "eri",
"honcho.ai_peer": "hermes",
"honcho.pin_peer_name": True,
"honcho.runtime_peer_prefix": "tg_",
"honcho.user_peer_aliases": [("123", "eri")],
}
monkeypatch.setattr(GatewayRunner, "_extract_honcho_cache_busting_config", _fake)
out = GatewayRunner._extract_cache_busting_config({"memory": {"provider": "honcho"}})
assert calls == [True]
assert out["honcho.peer_name"] == "eri"
assert out["honcho.user_peer_aliases"] == [("123", "eri")]
def test_memory_provider_change_busts_signature(self, monkeypatch):
"""Switching memory.provider must itself change the cache-busting
signature, so the agent is rebuilt when a user swaps providers
mid-gateway (independent of the honcho.json identity keys)."""
from gateway.run import GatewayRunner
# Neutralize honcho.json reads so the only varying input is the
# provider value itself.
monkeypatch.setattr(
GatewayRunner,
"_extract_honcho_cache_busting_config",
classmethod(lambda cls: cls._empty_honcho_cache_busting_config()),
)
sig_honcho = GatewayRunner._extract_cache_busting_config({"memory": {"provider": "honcho"}})
sig_mem0 = GatewayRunner._extract_cache_busting_config({"memory": {"provider": "mem0"}})
assert sig_honcho["memory.provider"] == "honcho"
assert sig_mem0["memory.provider"] == "mem0"
assert sig_honcho != sig_mem0
def test_honcho_cache_busting_config_memoized_by_mtime(self, monkeypatch, tmp_path):
"""Repeated Honcho extraction for unchanged honcho.json should reuse parse result."""
from types import SimpleNamespace
from gateway.run import GatewayRunner
config_path = tmp_path / "honcho.json"
config_path.write_text("{}")
parse_calls = []
class FakeConfig:
peer_name = "eri"
ai_peer = "hermes"
pin_peer_name = False
runtime_peer_prefix = "tg_"
user_peer_aliases = {"123": "eri"}
@classmethod
def from_global_config(cls, config_path=None):
parse_calls.append(config_path)
return cls()
fake_client = SimpleNamespace(
HonchoClientConfig=FakeConfig,
resolve_config_path=lambda: config_path,
)
monkeypatch.setitem(__import__("sys").modules, "plugins.memory.honcho.client", fake_client)
monkeypatch.setattr(GatewayRunner, "_HONCHO_CACHE_BUSTING_MEMO", {})
first = GatewayRunner._extract_honcho_cache_busting_config()
second = GatewayRunner._extract_honcho_cache_busting_config()
assert first == second
assert first["honcho.user_peer_aliases"] == [("123", "eri")]
assert parse_calls == [config_path]
config_path.write_text("{\n \"changed\": true\n}")
third = GatewayRunner._extract_honcho_cache_busting_config()
assert third == first
assert parse_calls == [config_path, config_path]
def test_full_round_trip_busts_cache_on_real_edit(self):
"""End-to-end: simulate a config edit on main and verify the
extracted cache_keys change produces a new signature."""
from gateway.run import GatewayRunner
runtime = {"api_key": "k", "base_url": "u", "provider": "p"}
cfg_before = {
"model": {"context_length": 200_000},
"compression": {"threshold": 0.50, "enabled": True},
}
cfg_after = {
"model": {"context_length": 200_000},
"compression": {"threshold": 0.75, "enabled": True}, # user raised threshold
}
sig_before = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys=GatewayRunner._extract_cache_busting_config(cfg_before),
)
sig_after = GatewayRunner._agent_config_signature(
"m", runtime, [], "",
cache_keys=GatewayRunner._extract_cache_busting_config(cfg_after),
)
assert sig_before != sig_after, (
"Editing compression.threshold in config.yaml must bust the "
"gateway's cached agent so the new threshold takes effect."
)
class TestAgentCacheLifecycle:
"""End-to-end cache behavior with real AIAgent construction."""
def test_cache_hit_returns_same_agent(self):
"""Second message with same config reuses the cached agent instance."""
from run_agent import AIAgent
runner = _make_runner()
session_key = "telegram:12345"
runtime = {"api_key": "test", "base_url": "https://openrouter.ai/api/v1",
"provider": "openrouter", "api_mode": "chat_completions"}
sig = runner._agent_config_signature("anthropic/claude-sonnet-4", runtime, ["hermes-telegram"], "")
# First message — create and cache
agent1 = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True, skip_context_files=True,
skip_memory=True, platform="telegram",
)
with runner._agent_cache_lock:
runner._agent_cache[session_key] = (agent1, sig)
# Second message — cache hit
with runner._agent_cache_lock:
cached = runner._agent_cache.get(session_key)
assert cached is not None
assert cached[1] == sig
assert cached[0] is agent1 # same instance
def test_cache_miss_on_model_change(self):
"""Model change produces different signature → cache miss."""
from run_agent import AIAgent
runner = _make_runner()
session_key = "telegram:12345"
runtime = {"api_key": "test", "base_url": "https://openrouter.ai/api/v1",
"provider": "openrouter", "api_mode": "chat_completions"}
old_sig = runner._agent_config_signature("anthropic/claude-sonnet-4", runtime, ["hermes-telegram"], "")
agent1 = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True, skip_context_files=True,
skip_memory=True, platform="telegram",
)
with runner._agent_cache_lock:
runner._agent_cache[session_key] = (agent1, old_sig)
# New model → different signature
new_sig = runner._agent_config_signature("anthropic/claude-opus-4.6", runtime, ["hermes-telegram"], "")
assert new_sig != old_sig
with runner._agent_cache_lock:
cached = runner._agent_cache.get(session_key)
assert cached[1] != new_sig # signature mismatch → would create new agent
def test_evict_on_session_reset(self):
"""_evict_cached_agent removes the entry."""
from run_agent import AIAgent
runner = _make_runner()
session_key = "telegram:12345"
agent = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True, skip_context_files=True,
skip_memory=True,
)
with runner._agent_cache_lock:
runner._agent_cache[session_key] = (agent, "sig123")
runner._evict_cached_agent(session_key)
with runner._agent_cache_lock:
assert session_key not in runner._agent_cache
def test_evict_does_not_affect_other_sessions(self):
"""Evicting one session leaves other sessions cached."""
runner = _make_runner()
with runner._agent_cache_lock:
runner._agent_cache["session-A"] = ("agent-A", "sig-A")
runner._agent_cache["session-B"] = ("agent-B", "sig-B")
runner._evict_cached_agent("session-A")
with runner._agent_cache_lock:
assert "session-A" not in runner._agent_cache
assert "session-B" in runner._agent_cache
def test_reasoning_config_updates_in_place(self):
"""Reasoning config can be set on a cached agent without eviction."""
from run_agent import AIAgent
agent = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True, skip_context_files=True,
skip_memory=True,
reasoning_config={"enabled": True, "effort": "medium"},
)
# Simulate per-message reasoning update
agent.reasoning_config = {"enabled": True, "effort": "high"}
assert agent.reasoning_config["effort"] == "high"
# System prompt should not be affected by reasoning change
prompt1 = agent._build_system_prompt()
agent._cached_system_prompt = prompt1 # simulate run_conversation caching
agent.reasoning_config = {"enabled": True, "effort": "low"}
prompt2 = agent._cached_system_prompt
assert prompt1 is prompt2 # same object — not invalidated by reasoning change
def test_system_prompt_frozen_across_cache_reuse(self):
"""The cached agent's system prompt stays identical across turns."""
from run_agent import AIAgent
agent = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True, skip_context_files=True,
skip_memory=True, platform="telegram",
)
# Build system prompt (simulates first run_conversation)
prompt1 = agent._build_system_prompt()
agent._cached_system_prompt = prompt1
# Simulate second turn — prompt should be frozen
prompt2 = agent._cached_system_prompt
assert prompt1 is prompt2 # same object, not rebuilt
def test_callbacks_update_without_cache_eviction(self):
"""Per-message callbacks can be set on cached agent."""
from run_agent import AIAgent
agent = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True, skip_context_files=True,
skip_memory=True,
)
# Set callbacks like the gateway does per-message
cb1 = lambda *a: None
cb2 = lambda *a: None
agent.tool_progress_callback = cb1
agent.step_callback = cb2
agent.stream_delta_callback = None
agent.status_callback = None
assert agent.tool_progress_callback is cb1
assert agent.step_callback is cb2
# Update for next message
cb3 = lambda *a: None
agent.tool_progress_callback = cb3
assert agent.tool_progress_callback is cb3
class TestAgentCacheBoundedGrowth:
"""LRU cap and idle-TTL eviction prevent unbounded cache growth."""
def _bounded_runner(self):
"""Runner with an OrderedDict cache (matches real gateway init)."""
from collections import OrderedDict
from gateway.run import GatewayRunner
runner = GatewayRunner.__new__(GatewayRunner)
runner._agent_cache = OrderedDict()
runner._agent_cache_lock = threading.Lock()
return runner
def _fake_agent(self, last_activity: float | None = None):
"""Lightweight stand-in; real AIAgent is heavy to construct."""
m = MagicMock()
if last_activity is not None:
m._last_activity_ts = last_activity
else:
import time as _t
m._last_activity_ts = _t.time()
return m
def test_cap_evicts_lru_when_exceeded(self, monkeypatch):
"""Inserting past _AGENT_CACHE_MAX_SIZE pops the oldest entry."""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", 3)
runner = self._bounded_runner()
runner._cleanup_agent_resources = MagicMock()
for i in range(3):
runner._agent_cache[f"s{i}"] = (self._fake_agent(), f"sig{i}")
# Insert a 4th — oldest (s0) must be evicted.
with runner._agent_cache_lock:
runner._agent_cache["s3"] = (self._fake_agent(), "sig3")
runner._enforce_agent_cache_cap()
assert "s0" not in runner._agent_cache
assert "s3" in runner._agent_cache
assert len(runner._agent_cache) == 3
def test_cap_respects_move_to_end(self, monkeypatch):
"""Entries refreshed via move_to_end are NOT evicted as 'oldest'."""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", 3)
runner = self._bounded_runner()
runner._cleanup_agent_resources = MagicMock()
for i in range(3):
runner._agent_cache[f"s{i}"] = (self._fake_agent(), f"sig{i}")
# Touch s0 — it is now MRU, so s1 becomes LRU.
runner._agent_cache.move_to_end("s0")
with runner._agent_cache_lock:
runner._agent_cache["s3"] = (self._fake_agent(), "sig3")
runner._enforce_agent_cache_cap()
assert "s0" in runner._agent_cache # rescued by move_to_end
assert "s1" not in runner._agent_cache # now oldest → evicted
assert "s3" in runner._agent_cache
def test_cap_triggers_cleanup_thread(self, monkeypatch):
"""Evicted agent has release_clients() called for it (soft cleanup).
Uses the soft path (_release_evicted_agent_soft), NOT the hard
_cleanup_agent_resources — cache eviction must not tear down
per-task state (terminal/browser/bg procs).
"""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", 1)
runner = self._bounded_runner()
release_calls: list = []
cleanup_calls: list = []
# Intercept both paths; only release_clients path should fire.
def _soft(agent):
release_calls.append(agent)
runner._release_evicted_agent_soft = _soft
runner._cleanup_agent_resources = lambda a: cleanup_calls.append(a)
old_agent = self._fake_agent()
new_agent = self._fake_agent()
with runner._agent_cache_lock:
runner._agent_cache["old"] = (old_agent, "sig_old")
runner._agent_cache["new"] = (new_agent, "sig_new")
runner._enforce_agent_cache_cap()
# Cleanup is dispatched to a daemon thread; join briefly to observe.
import time as _t
deadline = _t.time() + 2.0
while _t.time() < deadline and not release_calls:
_t.sleep(0.02)
assert old_agent in release_calls
assert new_agent not in release_calls
# Hard-cleanup path must NOT have fired — that's for session expiry only.
assert cleanup_calls == []
def test_idle_ttl_sweep_evicts_stale_agents(self, monkeypatch):
"""_sweep_idle_cached_agents removes agents idle past the TTL."""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_IDLE_TTL_SECS", 0.05)
runner = self._bounded_runner()
runner._cleanup_agent_resources = MagicMock()
import time as _t
fresh = self._fake_agent(last_activity=_t.time())
stale = self._fake_agent(last_activity=_t.time() - 10.0)
runner._agent_cache["fresh"] = (fresh, "s1")
runner._agent_cache["stale"] = (stale, "s2")
evicted = runner._sweep_idle_cached_agents()
assert evicted == 1
assert "stale" not in runner._agent_cache
assert "fresh" in runner._agent_cache
def test_idle_sweep_skips_agents_without_activity_ts(self, monkeypatch):
"""Agents missing _last_activity_ts are left alone (defensive)."""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_IDLE_TTL_SECS", 0.01)
runner = self._bounded_runner()
runner._cleanup_agent_resources = MagicMock()
no_ts = MagicMock(spec=[]) # no _last_activity_ts attribute
runner._agent_cache["s"] = (no_ts, "sig")
assert runner._sweep_idle_cached_agents() == 0
assert "s" in runner._agent_cache
def test_idle_sweep_keeps_agent_when_session_not_expired(self, monkeypatch):
"""Agents past idle TTL are kept if the session hasn't expired yet.
In daily-reset mode the reset can fire hours after the last
user message — evicting the agent early means the
session-expiry watcher has nothing to call on_session_end()
with, and memory providers miss the live transcript.
"""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_IDLE_TTL_SECS", 0.01)
runner = self._bounded_runner()
runner._cleanup_agent_resources = MagicMock()
import time as _t
stale = self._fake_agent(last_activity=_t.time() - 10.0)
# Session store says the session is still alive.
session_entry = MagicMock()
runner.session_store = MagicMock()
runner.session_store._entries = {"stale-session": session_entry}
runner.session_store._is_session_expired.return_value = False
runner._agent_cache["stale-session"] = (stale, "sig")
evicted = runner._sweep_idle_cached_agents()
assert evicted == 0
assert "stale-session" in runner._agent_cache
def test_idle_sweep_evicts_when_session_is_expired(self, monkeypatch):
"""Agent IS evicted when past idle TTL AND session store says expired."""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_IDLE_TTL_SECS", 0.01)
runner = self._bounded_runner()
runner._cleanup_agent_resources = MagicMock()
import time as _t
stale = self._fake_agent(last_activity=_t.time() - 10.0)
# Session store says the session has expired.
session_entry = MagicMock()
runner.session_store = MagicMock()
runner.session_store._entries = {"stale-session": session_entry}
runner.session_store._is_session_expired.return_value = True
runner._agent_cache["stale-session"] = (stale, "sig")
evicted = runner._sweep_idle_cached_agents()
assert evicted == 1
assert "stale-session" not in runner._agent_cache
def test_plain_dict_cache_is_tolerated(self):
"""Test fixtures using plain {} don't crash _enforce_agent_cache_cap."""
from gateway.run import GatewayRunner
runner = GatewayRunner.__new__(GatewayRunner)
runner._agent_cache = {} # plain dict, not OrderedDict
runner._agent_cache_lock = threading.Lock()
runner._cleanup_agent_resources = MagicMock()
# Should be a no-op rather than raising.
with runner._agent_cache_lock:
for i in range(200):
runner._agent_cache[f"s{i}"] = (MagicMock(), f"sig{i}")
runner._enforce_agent_cache_cap() # no crash, no eviction
assert len(runner._agent_cache) == 200
def test_main_lookup_updates_lru_order(self, monkeypatch):
"""Cache hit via the main-lookup path refreshes LRU position."""
runner = self._bounded_runner()
a0 = self._fake_agent()
a1 = self._fake_agent()
a2 = self._fake_agent()
runner._agent_cache["s0"] = (a0, "sig0")
runner._agent_cache["s1"] = (a1, "sig1")
runner._agent_cache["s2"] = (a2, "sig2")
# Simulate what _process_message_background does on a cache hit
# (minus the agent-state reset which isn't relevant here).
with runner._agent_cache_lock:
cached = runner._agent_cache.get("s0")
if cached and hasattr(runner._agent_cache, "move_to_end"):
runner._agent_cache.move_to_end("s0")
# After the hit, insertion order should be s1, s2, s0.
assert list(runner._agent_cache.keys()) == ["s1", "s2", "s0"]
class TestAgentCacheActiveSafety:
"""Safety: eviction must not tear down agents currently mid-turn.
AIAgent.close() kills process_registry entries for the task, cleans
the terminal sandbox, closes the OpenAI client, and cascades
.close() into active child subagents. Calling it while the agent
is still processing would crash the in-flight request. These tests
pin that eviction skips any agent present in _running_agents.
"""
def _runner(self):
from collections import OrderedDict
from gateway.run import GatewayRunner
runner = GatewayRunner.__new__(GatewayRunner)
runner._agent_cache = OrderedDict()
runner._agent_cache_lock = threading.Lock()
runner._running_agents = {}
return runner
def _fake_agent(self, idle_seconds: float = 0.0):
import time as _t
m = MagicMock()
m._last_activity_ts = _t.time() - idle_seconds
return m
def test_cap_skips_active_lru_entry(self, monkeypatch):
"""Active LRU entry is skipped; cache stays over cap rather than
compensating by evicting a newer entry.
Rationale: evicting a more-recent entry just because the oldest
slot is temporarily locked would punish the most recently-
inserted session (which has no cache to preserve) to protect
one that happens to be mid-turn. Better to let the cache stay
transiently over cap and re-check on the next insert.
"""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", 2)
runner = self._runner()
runner._cleanup_agent_resources = MagicMock()
active = self._fake_agent()
idle_a = self._fake_agent()
idle_b = self._fake_agent()
# Insertion order: active (oldest), idle_a, idle_b.
runner._agent_cache["session-active"] = (active, "sig")
runner._agent_cache["session-idle-a"] = (idle_a, "sig")
runner._agent_cache["session-idle-b"] = (idle_b, "sig")
# Mark `active` as mid-turn — it's LRU, but protected.
runner._running_agents["session-active"] = active
with runner._agent_cache_lock:
runner._enforce_agent_cache_cap()
# All three remain; no eviction ran, no cleanup dispatched.
assert "session-active" in runner._agent_cache
assert "session-idle-a" in runner._agent_cache
assert "session-idle-b" in runner._agent_cache
assert runner._cleanup_agent_resources.call_count == 0
def test_cap_evicts_when_multiple_excess_and_some_inactive(self, monkeypatch):
"""Mixed active/idle in the LRU excess window: only the idle ones go.
With CAP=2 and 4 entries, excess=2 (the two oldest). If the
oldest is active and the next is idle, we evict exactly one.
Cache ends at CAP+1, which is still better than unbounded.
"""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", 2)
runner = self._runner()
runner._cleanup_agent_resources = MagicMock()
oldest_active = self._fake_agent()
idle_second = self._fake_agent()
idle_third = self._fake_agent()
idle_fourth = self._fake_agent()
runner._agent_cache["s1"] = (oldest_active, "sig")
runner._agent_cache["s2"] = (idle_second, "sig") # in excess window, idle
runner._agent_cache["s3"] = (idle_third, "sig")
runner._agent_cache["s4"] = (idle_fourth, "sig")
runner._running_agents["s1"] = oldest_active # oldest is mid-turn
with runner._agent_cache_lock:
runner._enforce_agent_cache_cap()
# s1 protected (active), s2 evicted (idle + in excess window),
# s3 and s4 untouched (outside excess window).
assert "s1" in runner._agent_cache
assert "s2" not in runner._agent_cache
assert "s3" in runner._agent_cache
assert "s4" in runner._agent_cache
def test_cap_leaves_cache_over_limit_if_all_active(self, monkeypatch, caplog):
"""If every over-cap entry is mid-turn, the cache stays over cap.
Better to temporarily exceed the cap than to crash an in-flight
turn by tearing down its clients.
"""
from gateway import run as gw_run
import logging as _logging
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", 1)
runner = self._runner()
runner._cleanup_agent_resources = MagicMock()
a1 = self._fake_agent()
a2 = self._fake_agent()
a3 = self._fake_agent()
runner._agent_cache["s1"] = (a1, "sig")
runner._agent_cache["s2"] = (a2, "sig")
runner._agent_cache["s3"] = (a3, "sig")
# All three are mid-turn.
runner._running_agents["s1"] = a1
runner._running_agents["s2"] = a2
runner._running_agents["s3"] = a3
with caplog.at_level(_logging.WARNING, logger="gateway.run"):
with runner._agent_cache_lock:
runner._enforce_agent_cache_cap()
# Cache unchanged because eviction had to skip every candidate.
assert len(runner._agent_cache) == 3
# _cleanup_agent_resources must NOT have been scheduled.
assert runner._cleanup_agent_resources.call_count == 0
# And we logged a warning so operators can see the condition.
assert any("mid-turn" in r.message for r in caplog.records)
def test_cap_pending_sentinel_does_not_block_eviction(self, monkeypatch):
"""_AGENT_PENDING_SENTINEL in _running_agents is treated as 'not active'.
The sentinel is set while an agent is being CONSTRUCTED, before the
real AIAgent instance exists. Cached agents from other sessions
can still be evicted safely.
"""
from gateway import run as gw_run
from gateway.run import _AGENT_PENDING_SENTINEL
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", 1)
runner = self._runner()
runner._cleanup_agent_resources = MagicMock()
a1 = self._fake_agent()
a2 = self._fake_agent()
runner._agent_cache["s1"] = (a1, "sig")
runner._agent_cache["s2"] = (a2, "sig")
# Another session is mid-creation — sentinel, no real agent yet.
runner._running_agents["s3-being-created"] = _AGENT_PENDING_SENTINEL
with runner._agent_cache_lock:
runner._enforce_agent_cache_cap()
assert "s1" not in runner._agent_cache # evicted normally
assert "s2" in runner._agent_cache
def test_idle_sweep_skips_active_agent(self, monkeypatch):
"""Idle-TTL sweep must not tear down an active agent even if 'stale'."""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_IDLE_TTL_SECS", 0.01)
runner = self._runner()
runner._cleanup_agent_resources = MagicMock()
old_but_active = self._fake_agent(idle_seconds=10.0)
runner._agent_cache["s1"] = (old_but_active, "sig")
runner._running_agents["s1"] = old_but_active
evicted = runner._sweep_idle_cached_agents()
assert evicted == 0
assert "s1" in runner._agent_cache
assert runner._cleanup_agent_resources.call_count == 0
def test_eviction_does_not_close_active_agent_client(self, monkeypatch):
"""Live test: evicting an active agent does NOT null its .client.
This reproduces the original concern — if eviction fired while an
agent was mid-turn, `agent.close()` would set `self.client = None`
and the next API call inside the loop would crash. With the
active-agent skip, the client stays intact.
"""
from gateway import run as gw_run
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", 1)
runner = self._runner()
# Build a proper fake agent whose close() matches AIAgent's contract.
active = MagicMock()
active._last_activity_ts = __import__("time").time()
active.client = MagicMock() # simulate an OpenAI client
def _real_close():
active.client = None # mirrors run_agent.py:3299
active.close = _real_close
active.shutdown_memory_provider = MagicMock()
idle = self._fake_agent()
runner._agent_cache["active-session"] = (active, "sig")
runner._agent_cache["idle-session"] = (idle, "sig")
runner._running_agents["active-session"] = active
# Real cleanup function, not mocked — we want to see whether close()
# runs on the active agent. (It shouldn't.)
with runner._agent_cache_lock:
runner._enforce_agent_cache_cap()
# Let any eviction cleanup threads drain.
import time as _t
_t.sleep(0.2)
# The ACTIVE agent's client must still be usable.
assert active.client is not None, (
"Active agent's client was closed by eviction — "
"running turn would crash on its next API call."
)
class TestAgentCacheSpilloverLive:
"""Live E2E: fill cache with real AIAgent instances and stress it."""
def _runner(self):
from collections import OrderedDict
from gateway.run import GatewayRunner
runner = GatewayRunner.__new__(GatewayRunner)
runner._agent_cache = OrderedDict()
runner._agent_cache_lock = threading.Lock()
runner._running_agents = {}
return runner
def _real_agent(self):
"""A genuine AIAgent; no API calls are made during these tests."""
from run_agent import AIAgent
return AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True,
skip_context_files=True, skip_memory=True,
platform="telegram",
)
def test_fill_to_cap_then_spillover(self, monkeypatch):
"""Fill to cap with real agents, insert one more, oldest evicted."""
from gateway import run as gw_run
CAP = 8
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", CAP)
runner = self._runner()
agents = [self._real_agent() for _ in range(CAP)]
for i, a in enumerate(agents):
with runner._agent_cache_lock:
runner._agent_cache[f"s{i}"] = (a, "sig")
runner._enforce_agent_cache_cap()
assert len(runner._agent_cache) == CAP
# Spillover insertion.
newcomer = self._real_agent()
with runner._agent_cache_lock:
runner._agent_cache["new"] = (newcomer, "sig")
runner._enforce_agent_cache_cap()
# Oldest (s0) evicted, cap still CAP.
assert "s0" not in runner._agent_cache
assert "new" in runner._agent_cache
assert len(runner._agent_cache) == CAP
# Clean up so pytest doesn't leak resources.
for a in agents + [newcomer]:
try:
a.close()
except Exception:
pass
def test_spillover_all_active_keeps_cache_over_cap(self, monkeypatch, caplog):
"""Every slot active: cache goes over cap, no one gets torn down."""
from gateway import run as gw_run
import logging as _logging
CAP = 4
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", CAP)
runner = self._runner()
agents = [self._real_agent() for _ in range(CAP)]
for i, a in enumerate(agents):
runner._agent_cache[f"s{i}"] = (a, "sig")
runner._running_agents[f"s{i}"] = a # every session mid-turn
newcomer = self._real_agent()
with caplog.at_level(_logging.WARNING, logger="gateway.run"):
with runner._agent_cache_lock:
runner._agent_cache["new"] = (newcomer, "sig")
runner._enforce_agent_cache_cap()
assert len(runner._agent_cache) == CAP + 1 # temporarily over cap
# All existing agents still usable.
for i, a in enumerate(agents):
assert a.client is not None, f"s{i} got closed while active!"
# And we warned operators.
assert any("mid-turn" in r.message for r in caplog.records)
for a in agents + [newcomer]:
try:
a.close()
except Exception:
pass
def test_evicted_session_next_turn_gets_fresh_agent(self, monkeypatch):
"""After eviction, the same session_key can insert a fresh agent.
Simulates the real spillover flow: evicted session sends another
message, which builds a new AIAgent and re-enters the cache.
"""
from gateway import run as gw_run
CAP = 2
monkeypatch.setattr(gw_run, "_AGENT_CACHE_MAX_SIZE", CAP)
runner = self._runner()
a0 = self._real_agent()
a1 = self._real_agent()
runner._agent_cache["sA"] = (a0, "sig")
runner._agent_cache["sB"] = (a1, "sig")
# 3rd session forces sA (oldest) out.
a2 = self._real_agent()
with runner._agent_cache_lock:
runner._agent_cache["sC"] = (a2, "sig")
runner._enforce_agent_cache_cap()
assert "sA" not in runner._agent_cache
# Let the eviction cleanup thread run.
import time as _t
_t.sleep(0.3)
# Now sA's user sends another message → a fresh agent goes in.
a0_new = self._real_agent()
with runner._agent_cache_lock:
runner._agent_cache["sA"] = (a0_new, "sig")
runner._enforce_agent_cache_cap()
assert "sA" in runner._agent_cache
assert runner._agent_cache["sA"][0] is a0_new # the new one, not stale
# Fresh agent is usable.
assert a0_new.client is not None
for a in (a0, a1, a2, a0_new):
try:
a.close()
except Exception:
pass
class TestAgentCacheIdleResume:
"""End-to-end: idle-TTL-evicted session resumes cleanly with task state.
Real-world scenario: user leaves a Telegram session open for 2+ hours.
Idle-TTL evicts their cached agent. They come back and send a message.
The new agent built for the same session_id must inherit:
- Conversation history (from SessionStore — outside cache concern)
- Terminal sandbox (same task_id → same _active_environments entry)
- Browser daemon (same task_id → same browser session)
- Background processes (same task_id → same process_registry entries)
The ONLY thing that should reset is the LLM client pool (rebuilt fresh).
"""
def _runner(self):
from collections import OrderedDict
from gateway.run import GatewayRunner
runner = GatewayRunner.__new__(GatewayRunner)
runner._agent_cache = OrderedDict()
runner._agent_cache_lock = threading.Lock()
runner._running_agents = {}
return runner
def test_release_clients_does_not_touch_process_registry(self, monkeypatch):
"""release_clients must not call process_registry.kill_all for task_id."""
from run_agent import AIAgent
agent = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True,
skip_context_files=True, skip_memory=True,
session_id="idle-resume-test-session",
)
# Spy on process_registry.kill_all — it MUST NOT be called.
from tools import process_registry as _pr
kill_all_calls: list = []
original_kill_all = _pr.process_registry.kill_all
_pr.process_registry.kill_all = lambda **kw: kill_all_calls.append(kw)
try:
agent.release_clients()
finally:
_pr.process_registry.kill_all = original_kill_all
try:
agent.close()
except Exception:
pass
assert kill_all_calls == [], (
f"release_clients() called process_registry.kill_all — would "
f"kill user's bg processes on cache eviction. Calls: {kill_all_calls}"
)
def test_release_clients_does_not_touch_terminal_or_browser(self, monkeypatch):
"""release_clients must not call cleanup_vm or cleanup_browser."""
from run_agent import AIAgent
from tools import terminal_tool as _tt
from tools import browser_tool as _bt
agent = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True,
skip_context_files=True, skip_memory=True,
session_id="idle-resume-test-2",
)
vm_calls: list = []
browser_calls: list = []
original_vm = _tt.cleanup_vm
original_browser = _bt.cleanup_browser
_tt.cleanup_vm = lambda tid: vm_calls.append(tid)
_bt.cleanup_browser = lambda tid: browser_calls.append(tid)
try:
agent.release_clients()
finally:
_tt.cleanup_vm = original_vm
_bt.cleanup_browser = original_browser
try:
agent.close()
except Exception:
pass
assert vm_calls == [], (
f"release_clients() tore down terminal sandbox — user's cwd, "
f"env, and bg shells would be gone on resume. Calls: {vm_calls}"
)
assert browser_calls == [], (
f"release_clients() tore down browser session — user's open "
f"tabs and cookies gone on resume. Calls: {browser_calls}"
)
def test_release_clients_closes_llm_client(self):
"""release_clients IS expected to close the OpenAI/httpx client."""
from run_agent import AIAgent
agent = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True,
skip_context_files=True, skip_memory=True,
)
# Clients are lazy-built; force one to exist so we can verify close.
assert agent.client is not None # __init__ builds it
agent.release_clients()
# Post-release: client reference is dropped (memory freed).
assert agent.client is None
def test_close_vs_release_full_teardown_difference(self, monkeypatch):
"""close() tears down task state; release_clients() does not.
This pins the semantic contract: session-expiry path uses close()
(full teardown — session is done), cache-eviction path uses
release_clients() (soft — session may resume).
"""
from run_agent import AIAgent
import run_agent as _ra
# Agent A: evicted from cache (soft) — terminal survives.
# Agent B: session expired (hard) — terminal torn down.
agent_a = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True,
skip_context_files=True, skip_memory=True,
session_id="soft-session",
)
agent_b = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True,
skip_context_files=True, skip_memory=True,
session_id="hard-session",
)
vm_calls: list = []
# AIAgent.close() calls the ``cleanup_vm`` name bound into
# ``run_agent`` at import time, not ``tools.terminal_tool.cleanup_vm``
# directly — so patch the ``run_agent`` reference.
original_vm = _ra.cleanup_vm
_ra.cleanup_vm = lambda tid: vm_calls.append(tid)
try:
agent_a.release_clients() # cache eviction
agent_b.close() # session expiry
finally:
_ra.cleanup_vm = original_vm
try:
agent_a.close()
except Exception:
pass
# Only agent_b's task_id should appear in cleanup calls.
assert "hard-session" in vm_calls
assert "soft-session" not in vm_calls
def test_idle_evicted_session_rebuild_inherits_task_id(self, monkeypatch):
"""After idle-TTL eviction, a fresh agent with the same session_id
gets the same task_id — so tool state (terminal/browser/bg procs)
that persisted across eviction is reachable via the new agent.
"""
from gateway import run as gw_run
from run_agent import AIAgent
monkeypatch.setattr(gw_run, "_AGENT_CACHE_IDLE_TTL_SECS", 0.01)
runner = self._runner()
# Build an agent representing a stale (idle) session.
SESSION_ID = "long-lived-user-session"
old = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True,
skip_context_files=True, skip_memory=True,
session_id=SESSION_ID,
)
old._last_activity_ts = 0.0 # force idle
runner._agent_cache["sKey"] = (old, "sig")
# Simulate the idle-TTL sweep firing.
runner._sweep_idle_cached_agents()
assert "sKey" not in runner._agent_cache
# Wait for the daemon thread doing release_clients() to finish.
import time as _t
_t.sleep(0.3)
# Old agent's client is gone (soft cleanup fired).
assert old.client is None
# User comes back — new agent built for the SAME session_id.
new_agent = AIAgent(
model="anthropic/claude-sonnet-4", api_key="test",
base_url="https://openrouter.ai/api/v1", provider="openrouter",
max_iterations=5, quiet_mode=True,
skip_context_files=True, skip_memory=True,
session_id=SESSION_ID,
)
# Same session_id means same task_id routed to tools. The new
# agent inherits any per-task state (terminal sandbox etc.) that
# was preserved across eviction.
assert new_agent.session_id == old.session_id == SESSION_ID
# And it has a fresh working client.
assert new_agent.client is not None
try:
new_agent.close()
except Exception:
pass
_FAKE_NOW = 10_000.0 # Fixed epoch for deterministic time assertions
class TestCachedAgentInactivityReset:
"""Inactivity-clock reset must be gated on _interrupt_depth == 0.
On interrupt-recursive turns (_interrupt_depth > 0) the clock must
keep accumulating so the inactivity watchdog can fire when a turn is
stuck in an interrupt loop. Resetting unconditionally prevented the
30-min timeout from triggering (#15654). The depth-0 reset is still
needed: a session idle for 29 min must not trip the watchdog before
the new turn makes its first API call (#9051).
"""
def _fake_agent(self, stale_seconds: float = 1800.0):
m = MagicMock()
m._last_activity_ts = _FAKE_NOW - stale_seconds
m._api_call_count = 10
m._last_activity_desc = "previous turn activity"
return m
def test_fresh_turn_resets_idle_clock(self):
"""interrupt_depth=0: clock resets so a post-idle turn gets a
fresh 30-min inactivity window (guard for #9051)."""
from gateway.run import GatewayRunner
agent = self._fake_agent(stale_seconds=1800.0)
old_ts = agent._last_activity_ts
with patch("gateway.run.time") as mock_time:
mock_time.time.return_value = _FAKE_NOW
GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0)
assert agent._last_activity_ts == _FAKE_NOW, (
"_last_activity_ts was not reset on a fresh turn (interrupt_depth=0)"
)
assert agent._last_activity_ts > old_ts, (
"Stale idle time should be cleared so the new turn gets a fresh window"
)
def test_fresh_turn_resets_desc(self):
"""interrupt_depth=0: description is updated to reflect the new turn."""
from gateway.run import GatewayRunner
agent = self._fake_agent()
with patch("gateway.run.time") as mock_time:
mock_time.time.return_value = _FAKE_NOW
GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0)
assert agent._last_activity_desc == "starting new turn (cached)"
def test_interrupt_turn_preserves_idle_clock(self):
"""interrupt_depth=1: clock preserved so accumulated stuck-turn
idle time is not discarded by an interrupt-recursive re-entry (#15654)."""
from gateway.run import GatewayRunner
agent = self._fake_agent(stale_seconds=1200.0)
old_ts = agent._last_activity_ts
GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=1)
assert agent._last_activity_ts == old_ts, (
"_last_activity_ts must not be reset on interrupt-recursive turns "
"(interrupt_depth>0) — the watchdog needs the accumulated idle time"
)
def test_interrupt_turn_preserves_desc(self):
"""interrupt_depth=1: desc preserved — it is semantically paired with ts."""
from gateway.run import GatewayRunner
agent = self._fake_agent(stale_seconds=1200.0)
GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=1)
assert agent._last_activity_desc == "previous turn activity", (
"_last_activity_desc must not change on interrupt-recursive turns; "
"it describes the activity *at* _last_activity_ts"
)
def test_deep_interrupt_recursion_preserves_idle_clock(self):
"""interrupt_depth=MAX-1: clock still preserved at any non-zero depth."""
from gateway.run import GatewayRunner
agent = self._fake_agent(stale_seconds=600.0)
old_ts = agent._last_activity_ts
GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=4)
assert agent._last_activity_ts == old_ts
def test_fresh_turn_resets_flush_cursor(self):
"""interrupt_depth=0: _last_flushed_db_idx resets so new-turn
messages are fully persisted to the session DB (#44327)."""
from gateway.run import GatewayRunner
agent = self._fake_agent()
agent._last_flushed_db_idx = 42 # stale from previous turn
with patch("gateway.run.time") as mock_time:
mock_time.time.return_value = _FAKE_NOW
GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0)
assert agent._last_flushed_db_idx == 0, (
"_last_flushed_db_idx must be reset on a fresh turn so that "
"_flush_messages_to_session_db starts from index 0"
)
def test_interrupt_turn_preserves_flush_cursor(self):
"""interrupt_depth=1: _last_flushed_db_idx preserved so an
in-progress flush is not disrupted by interrupt re-entry."""
from gateway.run import GatewayRunner
agent = self._fake_agent()
agent._last_flushed_db_idx = 42
GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=1)
assert agent._last_flushed_db_idx == 42, (
"_last_flushed_db_idx must not be reset on interrupt-recursive "
"turns — the flush cursor tracks in-progress writes"
)
def test_api_call_count_reset_regardless_of_depth(self):
"""_api_call_count is always reset to 0 for the new turn, at any depth."""
from gateway.run import GatewayRunner
agent_fresh = self._fake_agent()
agent_interrupted = self._fake_agent()
with patch("gateway.run.time") as mock_time:
mock_time.time.return_value = _FAKE_NOW
GatewayRunner._init_cached_agent_for_turn(agent_fresh, interrupt_depth=0)
GatewayRunner._init_cached_agent_for_turn(agent_interrupted, interrupt_depth=1)
assert agent_fresh._api_call_count == 0
assert agent_interrupted._api_call_count == 0
def test_watchdog_accumulation_across_recursive_turns(self):
"""Scenario: stuck turn + user interrupt → recursive turn.
The idle time seen by the watchdog must reflect the full stuck
duration, not restart from zero on the recursive re-entry.
"""
from gateway.run import GatewayRunner
STUCK_FOR = 1750.0
agent = self._fake_agent(stale_seconds=STUCK_FOR)
# Simulate: user sees "Still working..." and sends another message.
# That triggers an interrupt → _run_agent recurses at depth=1.
GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=1)
# Watchdog sees time.time() - _last_activity_ts ≥ STUCK_FOR.
idle_secs = _FAKE_NOW - agent._last_activity_ts
assert idle_secs >= STUCK_FOR - 1.0, (
f"Watchdog would see {idle_secs:.0f}s idle, expected ~{STUCK_FOR}s. "
"Inactivity timeout could not fire for a stuck interrupted turn."
)
class TestAgentConfigSignatureUserId:
"""Shared-thread cache must not reuse an agent across users.
HonchoSessionManager freezes the resolved runtime user identity at
first-message init. When the gateway session_key omits the participant
ID (``thread_sessions_per_user=False``), a cached AIAgent created by
user A would otherwise be reused for user B, attributing B's writes to
A's resolved peer. Including ``user_id`` / ``user_id_alt`` in the
signature forces per-user agent builds in shared threads.
Tradeoff: cold prompt cache for each user's first turn in a shared
thread, in exchange for correct memory attribution.
"""
def test_signature_changes_with_user_id(self):
from gateway.run import GatewayRunner
runtime = {"provider": "anthropic", "api_key": "k", "base_url": "", "api_mode": "chat_completions"}
sig_a = GatewayRunner._agent_config_signature(
"claude-sonnet-4", runtime, ["hermes-telegram"], "", user_id="7654321"
)
sig_b = GatewayRunner._agent_config_signature(
"claude-sonnet-4", runtime, ["hermes-telegram"], "", user_id="491827364"
)
assert sig_a != sig_b
def test_signature_stable_with_same_user_id(self):
from gateway.run import GatewayRunner
runtime = {"provider": "anthropic", "api_key": "k", "base_url": "", "api_mode": "chat_completions"}
sig_1 = GatewayRunner._agent_config_signature(
"claude-sonnet-4", runtime, ["hermes-telegram"], "", user_id="7654321"
)
sig_2 = GatewayRunner._agent_config_signature(
"claude-sonnet-4", runtime, ["hermes-telegram"], "", user_id="7654321"
)
assert sig_1 == sig_2
def test_signature_changes_with_user_id_alt(self):
from gateway.run import GatewayRunner
runtime = {"provider": "anthropic", "api_key": "k", "base_url": "", "api_mode": "chat_completions"}
sig_a = GatewayRunner._agent_config_signature(
"claude-sonnet-4", runtime, ["hermes-telegram"], "",
user_id="7654321", user_id_alt="@igor_tg",
)
sig_b = GatewayRunner._agent_config_signature(
"claude-sonnet-4", runtime, ["hermes-telegram"], "",
user_id="7654321", user_id_alt="@erosika_tg",
)
assert sig_a != sig_b
def test_signature_omits_user_id_when_absent(self):
"""Default-None user_id must not change signatures vs unset call.
Callers that pass no user_id kwarg must produce a signature
byte-identical to ``user_id=None`` so in-flight caches survive
the rollout of this fix.
"""
from gateway.run import GatewayRunner
runtime = {"provider": "anthropic", "api_key": "k", "base_url": "", "api_mode": "chat_completions"}
sig_implicit = GatewayRunner._agent_config_signature(
"claude-sonnet-4", runtime, ["hermes-telegram"], "",
)
sig_explicit_none = GatewayRunner._agent_config_signature(
"claude-sonnet-4", runtime, ["hermes-telegram"], "",
user_id=None, user_id_alt=None,
)
assert sig_implicit == sig_explicit_none
class TestAgentCacheMessageCountRebaseline:
"""The cross-process coherence guard (#45966) must NOT invalidate the
cache on this process's OWN writes.
The guard snapshots ``message_count`` at agent-build time (before the
turn writes its own rows) and never refreshes it on reuse. Without a
post-turn re-baseline, the gateway's own turn grows the count and the
next turn sees a mismatch and rebuilds the agent — every turn, for every
conversation — silently destroying per-conversation prompt caching.
``_refresh_agent_cache_message_count`` re-baselines the stored count to
the now-current value after each turn, so the guard fires ONLY when a
different process changed the transcript. These tests pin both halves of
the invariant against the REAL SessionDB + the REAL guard condition.
"""
def _runner_with_db(self, db):
from hermes_state import AsyncSessionDB
runner = _make_runner()
# The gateway holds the async facade; the production refresh awaits it.
runner._session_db = AsyncSessionDB(db)
return runner
@staticmethod
def _guard_would_reuse(runner, session_key, session_id):
"""Mirror the production cache-hit guard's reuse decision exactly.
Reuse iff the live on-disk count equals the snapshot stored next to
the cached agent (or either side is None / it's a legacy 2-tuple).
"""
try:
row = runner._session_db._db.get_session(session_id)
live = row.get("message_count", 0) if row else None
except Exception:
live = None
with runner._agent_cache_lock:
cached = runner._agent_cache.get(session_key)
cached_mc = cached[2] if cached and len(cached) > 2 else None
invalidate = (
cached_mc is not None
and live is not None
and live != cached_mc
)
return not invalidate
@pytest.mark.asyncio
async def test_same_process_turns_preserve_cached_agent(self, tmp_path):
"""The regression guard: consecutive same-process turns must REUSE
the cached agent (prompt cache preserved), not rebuild every turn.
Drives the real lifecycle: snapshot at build (before this turn's
writes), turn appends its own rows, then the post-turn re-baseline
runs — so the NEXT turn's guard sees no external change and reuses.
"""
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "sessions.db")
db.create_session("s1", source="telegram")
runner = self._runner_with_db(db)
agent = object()
# Turn 1: cache miss -> build. Snapshot is the count BEFORE this
# turn's own writes (production stores _current_msg_count here).
_row = db.get_session("s1")
build_count = _row.get("message_count", 0) if _row else 0
with runner._agent_cache_lock:
runner._agent_cache["telegram:s1"] = (agent, "sig", build_count)
reuses = 0
for _turn in range(1, 6):
# This process's own turn flushes its user + assistant rows.
db.append_message("s1", role="user", content="u")
db.append_message("s1", role="assistant", content="a")
# Post-turn re-baseline (the fix).
await runner._refresh_agent_cache_message_count("telegram:s1", "s1")
# Next turn's guard decision.
if self._guard_would_reuse(runner, "telegram:s1", "s1"):
reuses += 1
# All 5 follow-on turns must reuse — WITHOUT the re-baseline this is 0.
assert reuses == 5
# The same agent instance is still cached (never rebuilt).
with runner._agent_cache_lock:
assert runner._agent_cache["telegram:s1"][0] is agent
@pytest.mark.asyncio
async def test_cross_process_write_still_invalidates(self, tmp_path):
"""After the re-baseline, a DIFFERENT process appending to the same
session must still flip the guard to rebuild (the #45966 fix holds).
"""
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "sessions.db")
db.create_session("s1", source="telegram")
runner = self._runner_with_db(db)
agent = object()
with runner._agent_cache_lock:
_row = db.get_session("s1")
runner._agent_cache["telegram:s1"] = (
agent, "sig", (_row.get("message_count", 0) if _row else 0),
)
# Our own turn + re-baseline -> reuse next turn.
db.append_message("s1", role="user", content="u")
db.append_message("s1", role="assistant", content="a")
await runner._refresh_agent_cache_message_count("telegram:s1", "s1")
assert self._guard_would_reuse(runner, "telegram:s1", "s1") is True
# ANOTHER process (e.g. the desktop dashboard backend) appends a turn
# to the SAME session in the shared DB — we have NOT re-baselined for it.
db.append_message("s1", role="user", content="external from dashboard")
# Guard must now reject reuse so the agent rebuilds from fresh disk.
assert self._guard_would_reuse(runner, "telegram:s1", "s1") is False
@pytest.mark.asyncio
async def test_rebaseline_is_fail_safe_and_skips_legacy_and_pending(self, tmp_path):
"""Re-baseline must never crash and must leave legacy 2-tuples and
pending-sentinel entries untouched."""
from hermes_state import AsyncSessionDB, SessionDB
from gateway.run import _AGENT_PENDING_SENTINEL
db = SessionDB(db_path=tmp_path / "sessions.db")
db.create_session("s1", source="telegram")
db.append_message("s1", role="user", content="hi")
runner = self._runner_with_db(db)
# No session_db -> no-op, no crash.
runner._session_db = None
await runner._refresh_agent_cache_message_count("telegram:s1", "s1")
runner._session_db = AsyncSessionDB(db)
# Falsy session_id -> no-op.
await runner._refresh_agent_cache_message_count("telegram:s1", "")
await runner._refresh_agent_cache_message_count("telegram:s1", None)
# Legacy 2-tuple is left untouched (it opts out of the guard).
with runner._agent_cache_lock:
runner._agent_cache["telegram:s1"] = (object(), "sig")
await runner._refresh_agent_cache_message_count("telegram:s1", "s1")
with runner._agent_cache_lock:
assert len(runner._agent_cache["telegram:s1"]) == 2
# Pending sentinel entry is left untouched.
with runner._agent_cache_lock:
runner._agent_cache["telegram:s1"] = (_AGENT_PENDING_SENTINEL, "sig", 0)
await runner._refresh_agent_cache_message_count("telegram:s1", "s1")
with runner._agent_cache_lock:
assert runner._agent_cache["telegram:s1"][0] is _AGENT_PENDING_SENTINEL
assert runner._agent_cache["telegram:s1"][2] == 0
# A probe that raises is swallowed (no crash, snapshot unchanged).
class _BoomDB:
def get_session(self, _sid):
raise RuntimeError("db locked")
runner._session_db = AsyncSessionDB(_BoomDB()) # type: ignore[assignment]
with runner._agent_cache_lock:
runner._agent_cache["telegram:s1"] = (object(), "sig", 5)
await runner._refresh_agent_cache_message_count("telegram:s1", "s1")
with runner._agent_cache_lock:
assert runner._agent_cache["telegram:s1"][2] == 5
@pytest.mark.asyncio
async def test_in_band_followup_reuses_cached_agent(self, tmp_path):
"""Behavioral regression for the in-band queued (/queue) follow-up.
#46237 re-baselines the snapshot only on the EXTERNAL-turn boundary
(in ``_handle_message_with_agent``, after the whole ``_run_agent``
chain unwinds). The recursive in-band follow-up re-enters the cache
guard MID-CHAIN — while the cache still holds the build-time snapshot
and the first turn has already flushed its own rows — so without a
re-baseline at the follow-up boundary the guard sees the grown count
and rebuilds the agent on THIS process's own writes, re-introducing
the every-turn rebuild #46237 set out to fix, on the follow-up path.
Pins both halves at that boundary: WITHOUT the re-baseline the in-band
follow-up would rebuild; WITH it the follow-up REUSES the warm agent.
The guard's reuse decision (``_guard_would_reuse``) mirrors the real
cache-hit guard, which reads ``get_session(session_id)`` with the same
``session_id`` the recursive ``_run_agent`` call is given.
"""
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "sessions.db")
db.create_session("s1", source="telegram")
runner = self._runner_with_db(db)
agent = object()
# First turn: cache miss -> build. Snapshot is the pre-turn count.
_row = db.get_session("s1")
build_count = _row.get("message_count", 0) if _row else 0
with runner._agent_cache_lock:
runner._agent_cache["telegram:s1"] = (agent, "sig", build_count)
# First turn flushes its own user + assistant rows.
db.append_message("s1", role="user", content="u")
db.append_message("s1", role="assistant", content="a")
# Bug reproduction: re-entering the guard at the in-band follow-up
# boundary WITHOUT the re-baseline sees the grown count and rebuilds.
assert self._guard_would_reuse(runner, "telegram:s1", "s1") is False
# The fix: re-baseline at the follow-up boundary.
await runner._refresh_agent_cache_message_count("telegram:s1", "s1")
# The in-band follow-up now REUSES the cached, warm-prefix agent.
assert self._guard_would_reuse(runner, "telegram:s1", "s1") is True
with runner._agent_cache_lock:
assert runner._agent_cache["telegram:s1"][0] is agent
def test_in_band_followup_rebaseline_precedes_recursion(self):
"""Pin the FIX PLACEMENT in the production source.
The behavioral test above proves the re-baseline makes the in-band
follow-up reuse the cached agent, but it calls the helper directly —
it would still pass if the production call were deleted. This guards
the actual call site: the queued (/queue) follow-up recurses via
``followup_result = await self._run_agent(...)`` inside
``_run_agent_inner`` and the re-baseline MUST run BEFORE that
recursion (running it only after, like the external-turn site, is too
late for the in-band path — the follow-up would already have rebuilt).
"""
import inspect
from gateway.run import GatewayRunner
# The recursion + pre-recursion re-baseline live in the extracted
# ``_run_agent_inner`` (older trees had them inline in ``_run_agent``).
src = inspect.getsource(GatewayRunner._run_agent_inner)
marker = "followup_result = await self._run_agent("
assert marker in src, "in-band queued follow-up recursion not found in _run_agent_inner"
before_recursion = src[: src.index(marker)]
assert "_refresh_agent_cache_message_count" in before_recursion, (
"the in-band queued follow-up recursion must be preceded by a "
"_refresh_agent_cache_message_count re-baseline, else the follow-up "
"rebuilds the agent on this process's own first-turn writes"
)
class TestCrossProcessInvalidationDefersCleanup:
"""#52197: cross-process cache invalidation must NOT run agent cleanup
while holding ``_agent_cache_lock``.
The #45966 guard popped the stale cached agent and then called the
blocking ``_cleanup_agent_resources`` (memory-provider shutdown, socket
teardown) *inside* the ``with _agent_cache_lock:`` block, on the gateway
event-loop thread. While that ran, ``_sweep_idle_cached_agents`` (driven
by the session-expiry watcher) blocked acquiring the same lock and the
asyncio loop stalled, tripping Discord heartbeat-blocked warnings.
The fix mirrors the cap-enforcer / idle-sweep paths: pop under the lock,
release it, then schedule the SOFT release (which preserves the session's
terminal sandbox / browser / bg processes for the immediately-rebuilt
agent) on a daemon thread.
These tests replicate the exact eviction sequence the production guard now
performs and pin the invariant: the lock is free while cleanup runs, and
the hard-teardown path is never used here.
"""
def _runner(self):
from collections import OrderedDict
from gateway.run import GatewayRunner
runner = GatewayRunner.__new__(GatewayRunner)
runner._agent_cache = OrderedDict()
runner._agent_cache_lock = threading.Lock()
return runner
@staticmethod
def _evict_like_production(runner, session_key):
"""Run the post-#52197 cross-process eviction sequence verbatim:
pop the stale entry under the lock, then schedule the soft release
on a daemon thread AFTER the lock is released."""
_xproc_evicted_agent = None
with runner._agent_cache_lock:
evicted = runner._agent_cache.pop(session_key, None)
_ev_agent = evicted[0] if isinstance(evicted, tuple) and evicted else None
if _ev_agent is not None:
_xproc_evicted_agent = _ev_agent
if _xproc_evicted_agent is not None:
threading.Thread(
target=runner._release_evicted_agent_soft,
args=(_xproc_evicted_agent,),
daemon=True,
name="agent-xproc-evict-test",
).start()
def test_cleanup_runs_with_lock_released(self):
"""The cache lock must be acquirable WHILE the evicted agent's
cleanup is running — proving cleanup is off the locked path."""
runner = self._runner()
cleanup_started = threading.Event()
release_lock = threading.Event()
def _soft(agent):
cleanup_started.set()
# Block here as if memory-provider shutdown / socket teardown is
# slow. If cleanup were still holding _agent_cache_lock, the
# assertion below could never acquire it.
release_lock.wait(timeout=2.0)
runner._release_evicted_agent_soft = _soft
runner._cleanup_agent_resources = MagicMock()
old_agent = MagicMock()
with runner._agent_cache_lock:
runner._agent_cache["telegram:s1"] = (old_agent, "sig", 3)
self._evict_like_production(runner, "telegram:s1")
# Wait until the (blocking) cleanup is mid-flight.
assert cleanup_started.wait(timeout=2.0)
# The lock MUST be free right now — this is the heart of #52197.
# A 0.5s acquire timeout would fire if cleanup held the lock.
acquired = runner._agent_cache_lock.acquire(timeout=0.5)
assert acquired, "cache lock blocked during cross-process cleanup (#52197)"
runner._agent_cache_lock.release()
# Let the cleanup thread finish.
release_lock.set()
# Stale entry was popped, hard-teardown path never used.
assert "telegram:s1" not in runner._agent_cache
runner._cleanup_agent_resources.assert_not_called()
def test_soft_release_scheduled_for_evicted_agent(self):
"""The evicted agent is handed to the soft-release path, not the
hard ``_cleanup_agent_resources`` teardown."""
runner = self._runner()
release_calls: list = []
runner._release_evicted_agent_soft = lambda agent: release_calls.append(agent)
runner._cleanup_agent_resources = MagicMock()
old_agent = MagicMock()
with runner._agent_cache_lock:
runner._agent_cache["telegram:s1"] = (old_agent, "sig", 3)
self._evict_like_production(runner, "telegram:s1")
import time as _t
deadline = _t.time() + 2.0
while _t.time() < deadline and not release_calls:
_t.sleep(0.02)
assert release_calls == [old_agent]
runner._cleanup_agent_resources.assert_not_called()
Reference in a new issue View git blame Copy permalink