Merge branch 'main' of github.com:NousResearch/hermes-agent into feat/ink-refactor

This commit is contained in:
Brooklyn Nicholson
2026-04-16 08:23:20 -05:00
128 changed files with 8357 additions and 842 deletions

View File

@@ -89,7 +89,8 @@ class TestReadCodexAccessToken:
hermes_home.mkdir(parents=True, exist_ok=True)
(hermes_home / "auth.json").write_text(json.dumps({"version": 1, "providers": {}}))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
result = _read_codex_access_token()
with patch("agent.auxiliary_client._select_pool_entry", return_value=(False, None)):
result = _read_codex_access_token()
assert result is None
def test_empty_token_returns_none(self, tmp_path, monkeypatch):
@@ -146,7 +147,8 @@ class TestReadCodexAccessToken:
},
}))
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
result = _read_codex_access_token()
with patch("agent.auxiliary_client._select_pool_entry", return_value=(False, None)):
result = _read_codex_access_token()
assert result is None, "Expired JWT should return None"
def test_valid_jwt_returns_token(self, tmp_path, monkeypatch):
@@ -585,7 +587,10 @@ class TestGetTextAuxiliaryClient:
assert call_kwargs.kwargs["base_url"] == "http://localhost:1234/v1"
def test_codex_fallback_when_nothing_else(self, codex_auth_dir):
with patch("agent.auxiliary_client._read_nous_auth", return_value=None), \
with patch("agent.auxiliary_client._try_openrouter", return_value=(None, None)), \
patch("agent.auxiliary_client._try_nous", return_value=(None, None)), \
patch("agent.auxiliary_client._try_custom_endpoint", return_value=(None, None)), \
patch("agent.auxiliary_client._read_main_provider", return_value="openrouter"), \
patch("agent.auxiliary_client.OpenAI") as mock_openai:
client, model = get_text_auxiliary_client()
assert model == "gpt-5.2-codex"
@@ -623,17 +628,21 @@ class TestGetTextAuxiliaryClient:
monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
with patch("agent.auxiliary_client._read_nous_auth", return_value=None), \
patch("agent.auxiliary_client._read_codex_access_token", return_value=None), \
patch("agent.auxiliary_client._resolve_api_key_provider", return_value=(None, None)):
with patch("agent.auxiliary_client._resolve_auto", return_value=(None, None)):
client, model = get_text_auxiliary_client()
assert client is None
assert model is None
def test_custom_endpoint_uses_codex_wrapper_when_runtime_requests_responses_api(self):
def test_custom_endpoint_uses_codex_wrapper_when_runtime_requests_responses_api(self, monkeypatch):
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
with patch("agent.auxiliary_client._resolve_custom_runtime",
return_value=("https://api.openai.com/v1", "sk-test", "codex_responses")), \
patch("agent.auxiliary_client._read_main_model", return_value="gpt-5.3-codex"), \
patch("agent.auxiliary_client._try_openrouter", return_value=(None, None)), \
patch("agent.auxiliary_client._try_nous", return_value=(None, None)), \
patch("agent.auxiliary_client._read_main_provider", return_value="openrouter"), \
patch("agent.auxiliary_client.OpenAI") as mock_openai:
client, model = get_text_auxiliary_client()

View File

@@ -232,7 +232,7 @@ class TestResolveVisionProviderClientModelNormalization:
assert provider == "zai"
assert client is not None
assert model == "glm-5.1"
assert model == "glm-5v-turbo" # zai has dedicated vision model in _PROVIDER_VISION_MODELS
class TestVisionPathApiMode:

View File

@@ -252,6 +252,11 @@ def test_exhausted_402_entry_resets_after_one_hour(tmp_path, monkeypatch):
def test_explicit_reset_timestamp_overrides_default_429_ttl(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
# Prevent auto-seeding from Codex CLI tokens on the host
monkeypatch.setattr(
"hermes_cli.auth._import_codex_cli_tokens",
lambda: None,
)
_write_auth_store(
tmp_path,
{

View File

@@ -939,3 +939,74 @@ class TestOnMemoryWriteBridge:
mgr.on_memory_write("add", "user", "test")
# Good provider still received the call despite bad provider crashing
assert good.memory_writes == [("add", "user", "test")]
class TestHonchoCadenceTracking:
"""Verify Honcho provider cadence gating depends on on_turn_start().
Bug: _turn_count was never updated because on_turn_start() was not called
from run_conversation(). This meant cadence checks always passed (every
turn fired both context refresh and dialectic). Fixed by calling
on_turn_start(self._user_turn_count, msg) before prefetch_all().
"""
def test_turn_count_updates_on_turn_start(self):
"""on_turn_start sets _turn_count, enabling cadence math."""
from plugins.memory.honcho import HonchoMemoryProvider
p = HonchoMemoryProvider()
assert p._turn_count == 0
p.on_turn_start(1, "hello")
assert p._turn_count == 1
p.on_turn_start(5, "world")
assert p._turn_count == 5
def test_queue_prefetch_respects_dialectic_cadence(self):
"""With dialecticCadence=3, dialectic should skip turns 2 and 3."""
from plugins.memory.honcho import HonchoMemoryProvider
p = HonchoMemoryProvider()
p._dialectic_cadence = 3
p._recall_mode = "context"
p._session_key = "test-session"
# Simulate a manager that records prefetch calls
class FakeManager:
def prefetch_context(self, key, query=None):
pass
def prefetch_dialectic(self, key, query):
pass
p._manager = FakeManager()
# Simulate turn 1: last_dialectic_turn = -999, so (1 - (-999)) >= 3 -> fires
p.on_turn_start(1, "turn 1")
p._last_dialectic_turn = 1 # simulate it fired
p._last_context_turn = 1
# Simulate turn 2: (2 - 1) = 1 < 3 -> should NOT fire dialectic
p.on_turn_start(2, "turn 2")
assert (p._turn_count - p._last_dialectic_turn) < p._dialectic_cadence
# Simulate turn 3: (3 - 1) = 2 < 3 -> should NOT fire dialectic
p.on_turn_start(3, "turn 3")
assert (p._turn_count - p._last_dialectic_turn) < p._dialectic_cadence
# Simulate turn 4: (4 - 1) = 3 >= 3 -> should fire dialectic
p.on_turn_start(4, "turn 4")
assert (p._turn_count - p._last_dialectic_turn) >= p._dialectic_cadence
def test_injection_frequency_first_turn_with_1indexed(self):
"""injection_frequency='first-turn' must inject on turn 1 (1-indexed)."""
from plugins.memory.honcho import HonchoMemoryProvider
p = HonchoMemoryProvider()
p._injection_frequency = "first-turn"
# Turn 1 should inject (not skip)
p.on_turn_start(1, "first message")
assert p._turn_count == 1
# The guard is `_turn_count > 1`, so turn 1 passes through
should_skip = p._injection_frequency == "first-turn" and p._turn_count > 1
assert not should_skip, "First turn (turn 1) should NOT be skipped"
# Turn 2 should skip
p.on_turn_start(2, "second message")
should_skip = p._injection_frequency == "first-turn" and p._turn_count > 1
assert should_skip, "Second turn (turn 2) SHOULD be skipped"

View File

@@ -34,6 +34,7 @@ class _FakeAgent:
[{"id": "t1", "content": "unfinished task", "status": "in_progress"}]
)
self.flush_memories = MagicMock()
self.commit_memory_session = MagicMock()
self._invalidate_system_prompt = MagicMock()
# Token counters (non-zero to verify reset)

View File

@@ -1,5 +1,6 @@
"""Tests for CLI /status command behavior."""
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
@@ -83,3 +84,18 @@ def test_show_session_status_prints_gateway_style_summary():
_, kwargs = cli_obj.console.print.call_args
assert kwargs.get("highlight") is False
assert kwargs.get("markup") is False
def test_profile_command_reports_custom_root_profile(monkeypatch, tmp_path, capsys):
"""Profile detection works for custom-root deployments (not under ~/.hermes)."""
cli_obj = _make_cli()
profile_home = tmp_path / "profiles" / "coder"
monkeypatch.setenv("HERMES_HOME", str(profile_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path / "unrelated-home")
cli_obj._handle_profile_command()
out = capsys.readouterr().out
assert "Profile: coder" in out
assert f"Home: {profile_home}" in out

View File

@@ -144,6 +144,18 @@ class TestGatewayPersonalityNone:
assert "none" in result.lower()
@pytest.mark.asyncio
async def test_empty_personality_list_uses_profile_display_path(self, tmp_path):
runner = self._make_runner(personalities={})
(tmp_path / "config.yaml").write_text(yaml.dump({"agent": {"personalities": {}}}))
with patch("gateway.run._hermes_home", tmp_path), \
patch("hermes_constants.display_hermes_home", return_value="~/.hermes/profiles/coder"):
event = self._make_event("")
result = await runner._handle_personality_command(event)
assert result == "No personalities configured in `~/.hermes/profiles/coder/config.yaml`"
class TestPersonalityDictFormat:
"""Test dict-format custom personalities with description, tone, style."""

66
tests/gateway/conftest.py Normal file
View File

@@ -0,0 +1,66 @@
"""Shared fixtures for gateway tests.
The ``_ensure_telegram_mock`` helper guarantees that a minimal mock of
the ``telegram`` package is registered in :data:`sys.modules` **before**
any test file triggers ``from gateway.platforms.telegram import ...``.
Without this, ``pytest-xdist`` workers that happen to collect
``test_telegram_caption_merge.py`` (bare top-level import, no per-file
mock) first will cache ``ChatType = None`` from the production
ImportError fallback, causing 30+ downstream test failures wherever
``ChatType.GROUP`` / ``ChatType.SUPERGROUP`` is accessed.
Individual test files may still call their own ``_ensure_telegram_mock``
— it short-circuits when the mock is already present.
"""
import sys
from unittest.mock import MagicMock
def _ensure_telegram_mock() -> None:
"""Install a comprehensive telegram mock in sys.modules.
Idempotent — skips when the real library is already imported.
Uses ``sys.modules[name] = mod`` (overwrite) instead of
``setdefault`` so it wins even if a partial/broken import
already cached a module with ``ChatType = None``.
"""
if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
return # Real library is installed — nothing to mock
mod = MagicMock()
mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
mod.constants.ParseMode.MARKDOWN = "Markdown"
mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
mod.constants.ParseMode.HTML = "HTML"
mod.constants.ChatType.PRIVATE = "private"
mod.constants.ChatType.GROUP = "group"
mod.constants.ChatType.SUPERGROUP = "supergroup"
mod.constants.ChatType.CHANNEL = "channel"
# Real exception classes so ``except (NetworkError, ...)`` clauses
# in production code don't blow up with TypeError.
mod.error.NetworkError = type("NetworkError", (OSError,), {})
mod.error.TimedOut = type("TimedOut", (OSError,), {})
mod.error.BadRequest = type("BadRequest", (Exception,), {})
mod.error.Forbidden = type("Forbidden", (Exception,), {})
mod.error.InvalidToken = type("InvalidToken", (Exception,), {})
mod.error.RetryAfter = type("RetryAfter", (Exception,), {"retry_after": 1})
mod.error.Conflict = type("Conflict", (Exception,), {})
# Update.ALL_TYPES used in start_polling()
mod.Update.ALL_TYPES = []
for name in (
"telegram",
"telegram.ext",
"telegram.constants",
"telegram.request",
):
sys.modules[name] = mod
sys.modules["telegram.error"] = mod.error
# Run at collection time — before any test file's module-level imports.
_ensure_telegram_mock()

View File

@@ -284,6 +284,58 @@ class TestLoadGatewayConfig:
assert config.unauthorized_dm_behavior == "ignore"
assert config.platforms[Platform.WHATSAPP].extra["unauthorized_dm_behavior"] == "pair"
def test_bridges_telegram_disable_link_previews_from_config_yaml(self, tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
config_path = hermes_home / "config.yaml"
config_path.write_text(
"telegram:\n"
" disable_link_previews: true\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
config = load_gateway_config()
assert config.platforms[Platform.TELEGRAM].extra["disable_link_previews"] is True
def test_bridges_telegram_proxy_url_from_config_yaml(self, tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
config_path = hermes_home / "config.yaml"
config_path.write_text(
"telegram:\n"
" proxy_url: socks5://127.0.0.1:1080\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.delenv("TELEGRAM_PROXY", raising=False)
load_gateway_config()
import os
assert os.environ.get("TELEGRAM_PROXY") == "socks5://127.0.0.1:1080"
def test_telegram_proxy_env_takes_precedence_over_config(self, tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
config_path = hermes_home / "config.yaml"
config_path.write_text(
"telegram:\n"
" proxy_url: http://from-config:8080\n",
encoding="utf-8",
)
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("TELEGRAM_PROXY", "socks5://from-env:1080")
load_gateway_config()
import os
assert os.environ.get("TELEGRAM_PROXY") == "socks5://from-env:1080"
class TestHomeChannelEnvOverrides:
"""Home channel env vars should apply even when the platform was already

View File

@@ -1,12 +1,15 @@
"""Tests for duplicate reply suppression across the gateway stack.
Covers three fix paths:
Covers four fix paths:
1. base.py: stale response suppressed when interrupt_event is set and a
pending message exists (#8221 / #2483)
2. run.py return path: already_sent propagated from stream consumer's
already_sent flag without requiring response_previewed (#8375)
3. run.py queued-message path: first response correctly detected as
already-streamed when already_sent is True without response_previewed
2. run.py return path: only confirmed final streamed delivery suppresses
the fallback final send; partial streamed output must not
3. run.py queued-message path: first response is skipped only when the
final response was actually streamed, not merely when partial output existed
4. stream_consumer.py cancellation handler: only confirms final delivery
when the best-effort send actually succeeds, not merely because partial
content was sent earlier
"""
import asyncio
@@ -153,15 +156,16 @@ class TestBaseInterruptSuppression:
assert any(s["content"] == "Valid response" for s in adapter.sent)
# ===================================================================
# Test 2: run.py — already_sent without response_previewed (#8375)
# Test 2: run.py — partial streamed output must not suppress final send
# ===================================================================
class TestAlreadySentWithoutResponsePreviewed:
"""The already_sent flag on the response dict should be set when the
stream consumer's already_sent is True, even if response_previewed is
False. This prevents duplicate sends when streaming was interrupted
by flood control."""
class TestOnlyFinalStreamDeliverySuppressesFinalSend:
"""The gateway should suppress the fallback final send only when the
stream consumer confirmed the final assistant reply was delivered.
Partial streamed output is not enough. If only already_sent=True,
the fallback final send must still happen so Telegram users don't lose
the real answer."""
def _make_mock_stream_consumer(self, already_sent=False, final_response_sent=False):
sc = SimpleNamespace(
@@ -170,21 +174,20 @@ class TestAlreadySentWithoutResponsePreviewed:
)
return sc
def test_already_sent_set_without_response_previewed(self):
"""Stream consumer already_sent=True should propagate to response
dict even when response_previewed is False."""
def test_partial_stream_output_does_not_set_already_sent(self):
"""already_sent=True alone must NOT suppress final delivery."""
sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=False)
response = {"final_response": "text", "response_previewed": False}
# Reproduce the logic from run.py return path (post-fix)
if sc and isinstance(response, dict) and not response.get("failed"):
if (
getattr(sc, "final_response_sent", False)
or getattr(sc, "already_sent", False)
):
_final = response.get("final_response") or ""
_is_empty_sentinel = not _final or _final == "(empty)"
_streamed = bool(sc and getattr(sc, "final_response_sent", False))
_previewed = bool(response.get("response_previewed"))
if not _is_empty_sentinel and (_streamed or _previewed):
response["already_sent"] = True
assert response.get("already_sent") is True
assert "already_sent" not in response
def test_already_sent_not_set_when_nothing_sent(self):
"""When stream consumer hasn't sent anything, already_sent should
@@ -193,24 +196,26 @@ class TestAlreadySentWithoutResponsePreviewed:
response = {"final_response": "text", "response_previewed": False}
if sc and isinstance(response, dict) and not response.get("failed"):
if (
getattr(sc, "final_response_sent", False)
or getattr(sc, "already_sent", False)
):
_final = response.get("final_response") or ""
_is_empty_sentinel = not _final or _final == "(empty)"
_streamed = bool(sc and getattr(sc, "final_response_sent", False))
_previewed = bool(response.get("response_previewed"))
if not _is_empty_sentinel and (_streamed or _previewed):
response["already_sent"] = True
assert "already_sent" not in response
def test_already_sent_set_on_final_response_sent(self):
"""final_response_sent=True should still work as before."""
"""final_response_sent=True should suppress duplicate final sends."""
sc = self._make_mock_stream_consumer(already_sent=False, final_response_sent=True)
response = {"final_response": "text"}
if sc and isinstance(response, dict) and not response.get("failed"):
if (
getattr(sc, "final_response_sent", False)
or getattr(sc, "already_sent", False)
):
_final = response.get("final_response") or ""
_is_empty_sentinel = not _final or _final == "(empty)"
_streamed = bool(sc and getattr(sc, "final_response_sent", False))
_previewed = bool(response.get("response_previewed"))
if not _is_empty_sentinel and (_streamed or _previewed):
response["already_sent"] = True
assert response.get("already_sent") is True
@@ -222,10 +227,11 @@ class TestAlreadySentWithoutResponsePreviewed:
response = {"final_response": "Error: something broke", "failed": True}
if sc and isinstance(response, dict) and not response.get("failed"):
if (
getattr(sc, "final_response_sent", False)
or getattr(sc, "already_sent", False)
):
_final = response.get("final_response") or ""
_is_empty_sentinel = not _final or _final == "(empty)"
_streamed = bool(sc and getattr(sc, "final_response_sent", False))
_previewed = bool(response.get("response_previewed"))
if not _is_empty_sentinel and (_streamed or _previewed):
response["already_sent"] = True
assert "already_sent" not in response
@@ -255,10 +261,9 @@ class TestEmptyResponseNotSuppressed:
if sc and isinstance(response, dict) and not response.get("failed"):
_final = response.get("final_response") or ""
_is_empty_sentinel = not _final or _final == "(empty)"
if not _is_empty_sentinel and (
getattr(sc, "final_response_sent", False)
or getattr(sc, "already_sent", False)
):
_streamed = bool(sc and getattr(sc, "final_response_sent", False))
_previewed = bool(response.get("response_previewed"))
if not _is_empty_sentinel and (_streamed or _previewed):
response["already_sent"] = True
def test_empty_sentinel_not_suppressed_with_already_sent(self):
@@ -283,10 +288,10 @@ class TestEmptyResponseNotSuppressed:
self._apply_suppression_logic(response, sc)
assert "already_sent" not in response
def test_real_response_still_suppressed_with_already_sent(self):
"""Normal non-empty response should still be suppressed when
streaming delivered content."""
sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=False)
def test_real_response_still_suppressed_only_when_final_delivery_confirmed(self):
"""Normal non-empty response should be suppressed only when the final
response was actually streamed."""
sc = self._make_mock_stream_consumer(already_sent=True, final_response_sent=True)
response = {"final_response": "Here are the search results..."}
self._apply_suppression_logic(response, sc)
assert response.get("already_sent") is True
@@ -299,8 +304,8 @@ class TestEmptyResponseNotSuppressed:
assert "already_sent" not in response
class TestQueuedMessageAlreadyStreamed:
"""The queued-message path should detect that the first response was
already streamed (already_sent=True) even without response_previewed."""
"""The queued-message path should skip the first response only when the
final response was actually streamed."""
def _make_mock_sc(self, already_sent=False, final_response_sent=False):
return SimpleNamespace(
@@ -308,18 +313,38 @@ class TestQueuedMessageAlreadyStreamed:
final_response_sent=final_response_sent,
)
def test_queued_path_detects_already_streamed(self):
"""already_sent=True on stream consumer means first response was
streamed — skip re-sending before processing queued message."""
_sc = self._make_mock_sc(already_sent=True)
def test_queued_path_only_skips_send_when_final_response_was_streamed(self):
"""Partial streamed output alone must not suppress the first response
before the queued follow-up is processed."""
_sc = self._make_mock_sc(already_sent=True, final_response_sent=False)
# Reproduce the queued-message logic from run.py (post-fix)
_already_streamed = bool(
_sc
and (
getattr(_sc, "final_response_sent", False)
or getattr(_sc, "already_sent", False)
)
_sc and getattr(_sc, "final_response_sent", False)
)
assert _already_streamed is False
def test_queued_path_detects_confirmed_final_stream_delivery(self):
"""Confirmed final streamed delivery should skip the resend."""
_sc = self._make_mock_sc(already_sent=True, final_response_sent=True)
response = {"response_previewed": False}
_already_streamed = bool(
(_sc and getattr(_sc, "final_response_sent", False))
or bool(response.get("response_previewed"))
)
assert _already_streamed is True
def test_queued_path_detects_previewed_response_delivery(self):
"""A response already previewed via the adapter should not be resent
before processing the queued follow-up."""
_sc = self._make_mock_sc(already_sent=False, final_response_sent=False)
response = {"response_previewed": True}
_already_streamed = bool(
(_sc and getattr(_sc, "final_response_sent", False))
or bool(response.get("response_previewed"))
)
assert _already_streamed is True
@@ -327,14 +352,10 @@ class TestQueuedMessageAlreadyStreamed:
def test_queued_path_sends_when_not_streamed(self):
"""Nothing was streamed — first response should be sent before
processing the queued message."""
_sc = self._make_mock_sc(already_sent=False)
_sc = self._make_mock_sc(already_sent=False, final_response_sent=False)
_already_streamed = bool(
_sc
and (
getattr(_sc, "final_response_sent", False)
or getattr(_sc, "already_sent", False)
)
_sc and getattr(_sc, "final_response_sent", False)
)
assert _already_streamed is False
@@ -344,11 +365,96 @@ class TestQueuedMessageAlreadyStreamed:
_sc = None
_already_streamed = bool(
_sc
and (
getattr(_sc, "final_response_sent", False)
or getattr(_sc, "already_sent", False)
)
_sc and getattr(_sc, "final_response_sent", False)
)
assert _already_streamed is False
# ===================================================================
# Test 4: stream_consumer.py — cancellation handler delivery confirmation
# ===================================================================
class TestCancellationHandlerDeliveryConfirmation:
"""The stream consumer's cancellation handler should only set
final_response_sent when the best-effort send actually succeeds.
Partial content (already_sent=True) alone must not promote to
final_response_sent — that would suppress the gateway's fallback
send even when the user never received the real answer."""
def test_partial_only_no_accumulated_stays_false(self):
"""Cancelled after sending intermediate text, nothing accumulated.
final_response_sent must stay False so the gateway fallback fires."""
already_sent = True
final_response_sent = False
accumulated = ""
message_id = None
_best_effort_ok = False
if accumulated and message_id:
_best_effort_ok = True # wouldn't enter
if _best_effort_ok and not final_response_sent:
final_response_sent = True
assert final_response_sent is False
def test_best_effort_succeeds_sets_true(self):
"""When accumulated content exists and best-effort send succeeds,
final_response_sent should become True."""
already_sent = True
final_response_sent = False
accumulated = "Here are the search results..."
message_id = "msg_123"
_best_effort_ok = False
if accumulated and message_id:
_best_effort_ok = True # simulating successful _send_or_edit
if _best_effort_ok and not final_response_sent:
final_response_sent = True
assert final_response_sent is True
def test_best_effort_fails_stays_false(self):
"""When best-effort send fails (flood control, network), the
gateway fallback must deliver the response."""
already_sent = True
final_response_sent = False
accumulated = "Here are the search results..."
message_id = "msg_123"
_best_effort_ok = False
if accumulated and message_id:
_best_effort_ok = False # simulating failed _send_or_edit
if _best_effort_ok and not final_response_sent:
final_response_sent = True
assert final_response_sent is False
def test_preserves_existing_true(self):
"""If final_response_sent was already True before cancellation,
it must remain True regardless."""
already_sent = True
final_response_sent = True
accumulated = ""
message_id = None
_best_effort_ok = False
if accumulated and message_id:
pass
if _best_effort_ok and not final_response_sent:
final_response_sent = True
assert final_response_sent is True
def test_old_behavior_would_have_promoted_partial(self):
"""Verify the old code would have incorrectly promoted
already_sent to final_response_sent even with no accumulated
content — proving the bug existed."""
already_sent = True
final_response_sent = False
# OLD cancellation handler logic:
if already_sent:
final_response_sent = True
assert final_response_sent is True # the bug: partial promoted to final

View File

@@ -0,0 +1,54 @@
"""Tests for Unicode dash normalization in /insights command flag parsing.
Telegram on iOS auto-converts -- to em/en dashes. The /insights handler
normalizes these before parsing --days and --source flags.
"""
import re
import pytest
# The regex from gateway/run.py insights handler
_UNICODE_DASH_RE = re.compile(r'[\u2012\u2013\u2014\u2015](days|source)')
def _normalize_insights_args(raw: str) -> str:
"""Apply the same normalization as the /insights handler."""
return _UNICODE_DASH_RE.sub(r'--\1', raw)
class TestInsightsUnicodeDashFlags:
"""--days and --source must survive iOS Unicode dash conversion."""
@pytest.mark.parametrize("input_str,expected", [
# Standard double hyphen (baseline)
("--days 7", "--days 7"),
("--source telegram", "--source telegram"),
# Em dash (U+2014)
("\u2014days 7", "--days 7"),
("\u2014source telegram", "--source telegram"),
# En dash (U+2013)
("\u2013days 7", "--days 7"),
("\u2013source telegram", "--source telegram"),
# Figure dash (U+2012)
("\u2012days 7", "--days 7"),
# Horizontal bar (U+2015)
("\u2015days 7", "--days 7"),
# Combined flags with em dashes
("\u2014days 30 \u2014source cli", "--days 30 --source cli"),
])
def test_unicode_dash_normalized(self, input_str, expected):
result = _normalize_insights_args(input_str)
assert result == expected
def test_regular_hyphens_unaffected(self):
"""Normal --days/--source must pass through unchanged."""
assert _normalize_insights_args("--days 7 --source discord") == "--days 7 --source discord"
def test_bare_number_still_works(self):
"""Shorthand /insights 7 (no flag) must not be mangled."""
assert _normalize_insights_args("7") == "7"
def test_no_flags_unchanged(self):
"""Input with no flags passes through as-is."""
assert _normalize_insights_args("") == ""
assert _normalize_insights_args("30") == "30"

View File

@@ -1,5 +1,6 @@
"""Tests for topic-aware gateway progress updates."""
import asyncio
import importlib
import sys
import time
@@ -415,6 +416,21 @@ class QueuedCommentaryAgent:
}
class BackgroundReviewAgent:
def __init__(self, **kwargs):
self.background_review_callback = kwargs.get("background_review_callback")
self.tools = []
def run_conversation(self, message, conversation_history=None, task_id=None):
if self.background_review_callback:
self.background_review_callback("💾 Skill 'prospect-scanner' created.")
return {
"final_response": "done",
"messages": [],
"api_calls": 1,
}
class VerboseAgent:
"""Agent that emits a tool call with args whose JSON exceeds 200 chars."""
LONG_CODE = "x" * 300
@@ -668,6 +684,66 @@ async def test_run_agent_queued_message_does_not_treat_commentary_as_final(monke
assert "final response 1" in sent_texts
@pytest.mark.asyncio
async def test_run_agent_defers_background_review_notification_until_release(monkeypatch, tmp_path):
adapter, result = await _run_with_agent(
monkeypatch,
tmp_path,
BackgroundReviewAgent,
session_id="sess-bg-review-order",
config_data={"display": {"interim_assistant_messages": True}},
)
assert result["final_response"] == "done"
assert adapter.sent == []
@pytest.mark.asyncio
async def test_base_processing_releases_post_delivery_callback_after_main_send():
"""Post-delivery callbacks on the adapter fire after the main response."""
adapter = ProgressCaptureAdapter()
async def _handler(event):
return "done"
adapter.set_message_handler(_handler)
released = []
def _post_delivery_cb():
released.append(True)
adapter.sent.append(
{
"chat_id": "bg-review",
"content": "💾 Skill 'prospect-scanner' created.",
"reply_to": None,
"metadata": None,
}
)
source = SessionSource(
platform=Platform.TELEGRAM,
chat_id="-1001",
chat_type="group",
thread_id="17585",
)
event = MessageEvent(
text="hello",
message_type=MessageType.TEXT,
source=source,
message_id="msg-1",
)
session_key = "agent:main:telegram:group:-1001:17585"
adapter._active_sessions[session_key] = asyncio.Event()
adapter._post_delivery_callbacks[session_key] = _post_delivery_cb
await adapter._process_message_background(event, session_key)
sent_texts = [call["content"] for call in adapter.sent]
assert sent_texts == ["done", "💾 Skill 'prospect-scanner' created."]
assert released == [True]
@pytest.mark.asyncio
async def test_verbose_mode_does_not_truncate_args_by_default(monkeypatch, tmp_path):
"""Verbose mode with default tool_preview_length (0) should NOT truncate args.

View File

@@ -283,6 +283,19 @@ class TestBuildSessionContextPrompt:
assert "Local" in prompt
assert "machine running this agent" in prompt
def test_local_delivery_path_uses_display_hermes_home(self):
config = GatewayConfig()
source = SessionSource(
platform=Platform.LOCAL, chat_id="cli",
chat_name="CLI terminal", chat_type="dm",
)
ctx = build_session_context(source, config)
with patch("hermes_constants.display_hermes_home", return_value="~/.hermes/profiles/coder"):
prompt = build_session_context_prompt(ctx)
assert "~/.hermes/profiles/coder/cron/output/" in prompt
def test_whatsapp_prompt(self):
config = GatewayConfig(
platforms={

View File

@@ -209,11 +209,13 @@ def test_set_session_env_includes_session_key():
# Capture baseline value before setting (may be non-empty from another
# test in the same pytest-xdist worker sharing the context).
baseline = get_session_env("HERMES_SESSION_KEY")
tokens = runner._set_session_env(context)
assert get_session_env("HERMES_SESSION_KEY") == "tg:-1001:17585"
runner._clear_session_env(tokens)
assert get_session_env("HERMES_SESSION_KEY") == baseline
# After clearing, the session key must not retain the value we just set.
# The exact post-clear value depends on context propagation from other
# tests, so only check that our value was removed, not what replaced it.
assert get_session_env("HERMES_SESSION_KEY") != "tg:-1001:17585"
def test_session_key_no_race_condition_with_contextvars(monkeypatch):
@@ -251,3 +253,72 @@ def test_session_key_no_race_condition_with_contextvars(monkeypatch):
assert results["session-B"] == "session-B", (
f"Session B got '{results['session-B']}' instead of 'session-B' — race condition!"
)
@pytest.mark.asyncio
async def test_run_in_executor_with_context_preserves_session_env(monkeypatch):
"""Gateway executor work should inherit session contextvars for tool routing."""
runner = object.__new__(GatewayRunner)
monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False)
monkeypatch.delenv("HERMES_SESSION_CHAT_ID", raising=False)
monkeypatch.delenv("HERMES_SESSION_THREAD_ID", raising=False)
monkeypatch.delenv("HERMES_SESSION_USER_ID", raising=False)
source = SessionSource(
platform=Platform.TELEGRAM,
chat_id="2144471399",
chat_type="dm",
user_id="123456",
user_name="alice",
thread_id=None,
)
context = SessionContext(
source=source,
connected_platforms=[],
home_channels={},
session_key="agent:main:telegram:dm:2144471399",
)
tokens = runner._set_session_env(context)
try:
result = await runner._run_in_executor_with_context(
lambda: {
"platform": get_session_env("HERMES_SESSION_PLATFORM"),
"chat_id": get_session_env("HERMES_SESSION_CHAT_ID"),
"user_id": get_session_env("HERMES_SESSION_USER_ID"),
"session_key": get_session_env("HERMES_SESSION_KEY"),
}
)
finally:
runner._clear_session_env(tokens)
assert result == {
"platform": "telegram",
"chat_id": "2144471399",
"user_id": "123456",
"session_key": "agent:main:telegram:dm:2144471399",
}
@pytest.mark.asyncio
async def test_run_in_executor_with_context_forwards_args():
"""_run_in_executor_with_context should forward *args to the callable."""
runner = object.__new__(GatewayRunner)
def add(a, b):
return a + b
result = await runner._run_in_executor_with_context(add, 3, 7)
assert result == 10
@pytest.mark.asyncio
async def test_run_in_executor_with_context_propagates_exceptions():
"""Exceptions inside the executor should propagate to the caller."""
runner = object.__new__(GatewayRunner)
def blow_up():
raise ValueError("boom")
with pytest.raises(ValueError, match="boom"):
await runner._run_in_executor_with_context(blow_up)

View File

@@ -14,7 +14,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from gateway.config import GatewayConfig, Platform, PlatformConfig
from gateway.platforms.base import MessageEvent, MessageType
from gateway.platforms.base import MessageEvent, MessageType, merge_pending_message_event
from gateway.run import GatewayRunner, _AGENT_PENDING_SENTINEL
from gateway.session import SessionSource, build_session_key
@@ -184,6 +184,80 @@ async def test_second_message_during_sentinel_queued_not_duplicate():
await task1
def test_merge_pending_message_event_merges_text_and_photo_followups():
pending = {}
source = SessionSource(
platform=Platform.TELEGRAM,
chat_id="12345",
chat_type="dm",
user_id="u1",
)
session_key = build_session_key(source)
text_event = MessageEvent(
text="first follow-up",
message_type=MessageType.TEXT,
source=source,
)
photo_event = MessageEvent(
text="see screenshot",
message_type=MessageType.PHOTO,
source=source,
media_urls=["/tmp/test.png"],
media_types=["image/png"],
)
merge_pending_message_event(pending, session_key, text_event, merge_text=True)
merge_pending_message_event(pending, session_key, photo_event, merge_text=True)
merged = pending[session_key]
assert merged.message_type == MessageType.PHOTO
assert merged.text == "first follow-up\n\nsee screenshot"
assert merged.media_urls == ["/tmp/test.png"]
assert merged.media_types == ["image/png"]
@pytest.mark.asyncio
async def test_recent_telegram_text_followup_is_queued_without_interrupt():
runner = _make_runner()
event = _make_event(text="follow-up")
session_key = build_session_key(event.source)
fake_agent = MagicMock()
fake_agent.get_activity_summary.return_value = {"seconds_since_activity": 0}
runner._running_agents[session_key] = fake_agent
import time as _time
runner._running_agents_ts[session_key] = _time.time()
result = await runner._handle_message(event)
assert result is None
fake_agent.interrupt.assert_not_called()
adapter = runner.adapters[Platform.TELEGRAM]
assert adapter._pending_messages[session_key].text == "follow-up"
@pytest.mark.asyncio
async def test_recent_telegram_followups_append_in_pending_queue():
runner = _make_runner()
first = _make_event(text="part one")
second = _make_event(text="part two")
session_key = build_session_key(first.source)
fake_agent = MagicMock()
fake_agent.get_activity_summary.return_value = {"seconds_since_activity": 0}
runner._running_agents[session_key] = fake_agent
import time as _time
runner._running_agents_ts[session_key] = _time.time()
await runner._handle_message(first)
await runner._handle_message(second)
fake_agent.interrupt.assert_not_called()
adapter = runner.adapters[Platform.TELEGRAM]
assert adapter._pending_messages[session_key].text == "part one\npart two"
# ------------------------------------------------------------------
# Test 5: Sentinel not placed for command messages
# ------------------------------------------------------------------
@@ -273,6 +347,7 @@ async def test_stop_hard_kills_running_agent():
# Simulate a running (possibly hung) agent
fake_agent = MagicMock()
fake_agent.get_activity_summary.return_value = {"seconds_since_activity": 0}
runner._running_agents[session_key] = fake_agent
# Send /stop
@@ -305,6 +380,7 @@ async def test_stop_clears_pending_messages():
)
fake_agent = MagicMock()
fake_agent.get_activity_summary.return_value = {"seconds_since_activity": 0}
runner._running_agents[session_key] = fake_agent
runner._pending_messages[session_key] = "some queued text"

View File

@@ -1678,11 +1678,11 @@ class TestProgressMessageThread:
msg_event = captured_events[0]
source = msg_event.source
# For a top-level DM: source.thread_id should remain None
# (session keying must not be affected)
assert source.thread_id is None, (
"source.thread_id must stay None for top-level DMs "
"so they share one continuous session"
# With default dm_top_level_threads_as_sessions=True, source.thread_id
# should equal the message ts so each DM thread gets its own session.
assert source.thread_id == "1234567890.000001", (
"source.thread_id must equal the message ts for top-level DMs "
"so each reply thread gets its own session"
)
# The message_id should be the event's ts — this is what the gateway
@@ -1707,6 +1707,34 @@ class TestProgressMessageThread:
"ensuring progress messages land in the thread"
)
@pytest.mark.asyncio
async def test_dm_toplevel_shares_session_when_disabled(self, adapter):
"""Opting out restores legacy single-session-per-DM-channel behavior."""
adapter.config.extra["dm_top_level_threads_as_sessions"] = False
event = {
"channel": "D_DM",
"channel_type": "im",
"user": "U_USER",
"text": "Hello bot",
"ts": "1234567890.000001",
}
captured_events = []
adapter.handle_message = AsyncMock(side_effect=lambda e: captured_events.append(e))
with patch.object(adapter, "_resolve_user_name", new=AsyncMock(return_value="testuser")):
await adapter._handle_slack_message(event)
assert len(captured_events) == 1
msg_event = captured_events[0]
source = msg_event.source
assert source.thread_id is None, (
"source.thread_id must stay None when "
"dm_top_level_threads_as_sessions is disabled"
)
@pytest.mark.asyncio
async def test_channel_mention_progress_uses_thread_ts(self, adapter):
"""Progress messages for a channel @mention should go into the reply thread."""

View File

@@ -279,3 +279,28 @@ async def test_status_command_bypasses_active_session_guard():
assert "Agent Running" in sent[0]
assert not interrupt_event.is_set(), "/status incorrectly triggered an agent interrupt"
assert session_key not in adapter._pending_messages, "/status was incorrectly queued"
@pytest.mark.asyncio
async def test_profile_command_reports_custom_root_profile(monkeypatch, tmp_path):
"""Gateway /profile detects custom-root profiles (not under ~/.hermes)."""
from pathlib import Path
session_entry = SessionEntry(
session_key=build_session_key(_make_source()),
session_id="sess-1",
created_at=datetime.now(),
updated_at=datetime.now(),
platform=Platform.TELEGRAM,
chat_type="dm",
)
runner = _make_runner(session_entry)
profile_home = tmp_path / "profiles" / "coder"
monkeypatch.setenv("HERMES_HOME", str(profile_home))
monkeypatch.setattr(Path, "home", lambda: tmp_path / "unrelated-home")
result = await runner._handle_profile_command(_make_event("/profile"))
assert "**Profile:** `coder`" in result
assert f"**Home:** `{profile_home}`" in result

View File

@@ -50,9 +50,9 @@ from gateway.platforms.telegram import TelegramAdapter
from gateway.config import Platform, PlatformConfig
def _make_adapter():
def _make_adapter(extra=None):
"""Create a TelegramAdapter with mocked internals."""
config = PlatformConfig(enabled=True, token="test-token")
config = PlatformConfig(enabled=True, token="test-token", extra=extra or {})
adapter = TelegramAdapter(config)
adapter._bot = AsyncMock()
adapter._app = MagicMock()
@@ -134,6 +134,23 @@ class TestTelegramExecApproval:
)
assert result.success is False
@pytest.mark.asyncio
async def test_disable_link_previews_sets_preview_kwargs(self):
adapter = _make_adapter(extra={"disable_link_previews": True})
mock_msg = MagicMock()
mock_msg.message_id = 42
adapter._bot.send_message = AsyncMock(return_value=mock_msg)
await adapter.send_exec_approval(
chat_id="12345", command="ls", session_key="s"
)
kwargs = adapter._bot.send_message.call_args[1]
assert (
kwargs.get("disable_web_page_preview") is True
or kwargs.get("link_preview_options") is not None
)
@pytest.mark.asyncio
async def test_truncates_long_command(self):
adapter = _make_adapter()

View File

@@ -45,6 +45,11 @@ class FakeRetryAfter(Exception):
# Build a fake telegram module tree so the adapter's internal imports work
_fake_telegram = types.ModuleType("telegram")
_fake_telegram.Update = object
_fake_telegram.Bot = object
_fake_telegram.Message = object
_fake_telegram.InlineKeyboardButton = object
_fake_telegram.InlineKeyboardMarkup = object
_fake_telegram_error = types.ModuleType("telegram.error")
_fake_telegram_error.NetworkError = FakeNetworkError
_fake_telegram_error.BadRequest = FakeBadRequest
@@ -52,7 +57,21 @@ _fake_telegram_error.TimedOut = FakeTimedOut
_fake_telegram.error = _fake_telegram_error
_fake_telegram_constants = types.ModuleType("telegram.constants")
_fake_telegram_constants.ParseMode = SimpleNamespace(MARKDOWN_V2="MarkdownV2")
_fake_telegram_constants.ChatType = SimpleNamespace(
GROUP="group",
SUPERGROUP="supergroup",
CHANNEL="channel",
)
_fake_telegram.constants = _fake_telegram_constants
_fake_telegram_ext = types.ModuleType("telegram.ext")
_fake_telegram_ext.Application = object
_fake_telegram_ext.CommandHandler = object
_fake_telegram_ext.CallbackQueryHandler = object
_fake_telegram_ext.MessageHandler = object
_fake_telegram_ext.ContextTypes = SimpleNamespace(DEFAULT_TYPE=object)
_fake_telegram_ext.filters = object
_fake_telegram_request = types.ModuleType("telegram.request")
_fake_telegram_request.HTTPXRequest = object
@pytest.fixture(autouse=True)
@@ -61,6 +80,8 @@ def _inject_fake_telegram(monkeypatch):
monkeypatch.setitem(sys.modules, "telegram", _fake_telegram)
monkeypatch.setitem(sys.modules, "telegram.error", _fake_telegram_error)
monkeypatch.setitem(sys.modules, "telegram.constants", _fake_telegram_constants)
monkeypatch.setitem(sys.modules, "telegram.ext", _fake_telegram_ext)
monkeypatch.setitem(sys.modules, "telegram.request", _fake_telegram_request)
def _make_adapter():
@@ -68,6 +89,7 @@ def _make_adapter():
config = PlatformConfig(enabled=True, token="fake-token")
adapter = object.__new__(TelegramAdapter)
adapter.config = config
adapter._config = config
adapter._platform = Platform.TELEGRAM
adapter._connected = True
@@ -82,6 +104,81 @@ def _make_adapter():
return adapter
def test_forum_general_topic_without_message_thread_id_keeps_thread_context():
"""Forum General-topic messages should keep synthetic thread context."""
from gateway.platforms import telegram as telegram_mod
adapter = _make_adapter()
message = SimpleNamespace(
text="hello from General",
caption=None,
chat=SimpleNamespace(
id=-100123,
type=telegram_mod.ChatType.SUPERGROUP,
is_forum=True,
title="Forum group",
),
from_user=SimpleNamespace(id=456, full_name="Alice"),
message_thread_id=None,
reply_to_message=None,
message_id=10,
date=None,
)
event = adapter._build_message_event(message, msg_type=SimpleNamespace(value="text"))
assert event.source.chat_id == "-100123"
assert event.source.chat_type == "group"
assert event.source.thread_id == "1"
@pytest.mark.asyncio
async def test_send_omits_general_topic_thread_id():
"""Telegram sends to forum General should omit message_thread_id=1."""
adapter = _make_adapter()
call_log = []
async def mock_send_message(**kwargs):
call_log.append(dict(kwargs))
return SimpleNamespace(message_id=42)
adapter._bot = SimpleNamespace(send_message=mock_send_message)
result = await adapter.send(
chat_id="-100123",
content="test message",
metadata={"thread_id": "1"},
)
assert result.success is True
assert len(call_log) == 1
assert call_log[0]["chat_id"] == -100123
assert call_log[0]["text"] == "test message"
assert call_log[0]["reply_to_message_id"] is None
assert call_log[0]["message_thread_id"] is None
@pytest.mark.asyncio
async def test_send_typing_retries_without_general_thread_when_not_found():
"""Typing for forum General should fall back if Telegram rejects thread 1."""
adapter = _make_adapter()
call_log = []
async def mock_send_chat_action(**kwargs):
call_log.append(dict(kwargs))
if kwargs.get("message_thread_id") == 1:
raise FakeBadRequest("Message thread not found")
adapter._bot = SimpleNamespace(send_chat_action=mock_send_chat_action)
await adapter.send_typing("-100123", metadata={"thread_id": "1"})
assert call_log == [
{"chat_id": -100123, "action": "typing", "message_thread_id": 1},
{"chat_id": -100123, "action": "typing", "message_thread_id": None},
]
@pytest.mark.asyncio
async def test_send_retries_without_thread_on_thread_not_found():
"""When message_thread_id causes 'thread not found', retry without it."""

View File

@@ -613,6 +613,7 @@ class TestDetectVenvDir:
# Not inside a virtualenv
monkeypatch.setattr("sys.prefix", "/usr")
monkeypatch.setattr("sys.base_prefix", "/usr")
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
dot_venv = tmp_path / ".venv"
@@ -624,6 +625,7 @@ class TestDetectVenvDir:
def test_falls_back_to_venv_directory(self, tmp_path, monkeypatch):
monkeypatch.setattr("sys.prefix", "/usr")
monkeypatch.setattr("sys.base_prefix", "/usr")
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
venv = tmp_path / "venv"
@@ -635,6 +637,7 @@ class TestDetectVenvDir:
def test_prefers_dot_venv_over_venv(self, tmp_path, monkeypatch):
monkeypatch.setattr("sys.prefix", "/usr")
monkeypatch.setattr("sys.base_prefix", "/usr")
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
(tmp_path / ".venv").mkdir()
@@ -646,6 +649,7 @@ class TestDetectVenvDir:
def test_returns_none_when_no_virtualenv(self, tmp_path, monkeypatch):
monkeypatch.setattr("sys.prefix", "/usr")
monkeypatch.setattr("sys.base_prefix", "/usr")
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
monkeypatch.setattr(gateway_cli, "PROJECT_ROOT", tmp_path)
result = gateway_cli._detect_venv_dir()

View File

@@ -0,0 +1,101 @@
"""Regression tests for Copilot api_mode recomputation during /model switch.
When switching models within the Copilot provider (e.g. GPT-5 → Claude),
the stale api_mode from resolve_runtime_provider must be overridden with
a fresh value computed from the *new* model. Without the fix, Claude
requests went through the Responses API and failed with
``unsupported_api_for_model``.
"""
from unittest.mock import patch
from hermes_cli.model_switch import switch_model
_MOCK_VALIDATION = {
"accepted": True,
"persist": True,
"recognized": True,
"message": None,
}
def _run_copilot_switch(
raw_input: str,
current_provider: str = "copilot",
current_model: str = "gpt-5.4",
explicit_provider: str = "",
runtime_api_mode: str = "codex_responses",
):
"""Run switch_model with Copilot mocks and return the result."""
with (
patch("hermes_cli.model_switch.resolve_alias", return_value=None),
patch("hermes_cli.model_switch.list_provider_models", return_value=[]),
patch(
"hermes_cli.runtime_provider.resolve_runtime_provider",
return_value={
"api_key": "ghu_test_token",
"base_url": "https://api.githubcopilot.com",
"api_mode": runtime_api_mode,
},
),
patch(
"hermes_cli.models.validate_requested_model",
return_value=_MOCK_VALIDATION,
),
patch("hermes_cli.model_switch.get_model_info", return_value=None),
patch("hermes_cli.model_switch.get_model_capabilities", return_value=None),
patch("hermes_cli.models.detect_provider_for_model", return_value=None),
):
return switch_model(
raw_input=raw_input,
current_provider=current_provider,
current_model=current_model,
explicit_provider=explicit_provider,
)
def test_same_provider_copilot_switch_recomputes_api_mode():
"""GPT-5 → Claude on copilot: api_mode must flip to chat_completions."""
result = _run_copilot_switch(
raw_input="claude-opus-4.6",
current_provider="copilot",
current_model="gpt-5.4",
)
assert result.success, f"switch_model failed: {result.error_message}"
assert result.new_model == "claude-opus-4.6"
assert result.target_provider == "copilot"
assert result.api_mode == "chat_completions"
def test_explicit_copilot_switch_uses_selected_model_api_mode():
"""Cross-provider switch to copilot: api_mode from new model, not stale runtime."""
result = _run_copilot_switch(
raw_input="claude-opus-4.6",
current_provider="openrouter",
current_model="anthropic/claude-sonnet-4.6",
explicit_provider="copilot",
)
assert result.success, f"switch_model failed: {result.error_message}"
assert result.new_model == "claude-opus-4.6"
assert result.target_provider == "github-copilot"
assert result.api_mode == "chat_completions"
def test_copilot_gpt5_keeps_codex_responses():
"""GPT-5 → GPT-5 on copilot: api_mode must stay codex_responses."""
result = _run_copilot_switch(
raw_input="gpt-5.4-mini",
current_provider="copilot",
current_model="gpt-5.4",
runtime_api_mode="codex_responses",
)
assert result.success, f"switch_model failed: {result.error_message}"
assert result.new_model == "gpt-5.4-mini"
assert result.target_provider == "copilot"
# gpt-5.4-mini is a GPT-5 variant — should use codex_responses
# (gpt-5-mini is the special case that uses chat_completions)
assert result.api_mode == "codex_responses"

View File

@@ -163,7 +163,7 @@ class TestNormalizeProvider:
class TestProviderLabel:
def test_known_labels_and_auto(self):
assert provider_label("anthropic") == "Anthropic"
assert provider_label("kimi") == "Kimi / Moonshot"
assert provider_label("kimi") == "Kimi / Kimi Coding Plan"
assert provider_label("copilot") == "GitHub Copilot"
assert provider_label("copilot-acp") == "GitHub Copilot ACP"
assert provider_label("auto") == "Auto"

View File

@@ -0,0 +1,351 @@
"""Tests for Ollama Cloud provider integration."""
import os
import pytest
from unittest.mock import patch, MagicMock
from hermes_cli.auth import PROVIDER_REGISTRY, resolve_provider, resolve_api_key_provider_credentials
from hermes_cli.models import _PROVIDER_MODELS, _PROVIDER_LABELS, _PROVIDER_ALIASES, normalize_provider
from hermes_cli.model_normalize import normalize_model_for_provider
from agent.model_metadata import _URL_TO_PROVIDER, _PROVIDER_PREFIXES
from agent.models_dev import PROVIDER_TO_MODELS_DEV, list_agentic_models
# ── Provider Registry ──
class TestOllamaCloudProviderRegistry:
def test_ollama_cloud_in_registry(self):
assert "ollama-cloud" in PROVIDER_REGISTRY
def test_ollama_cloud_config(self):
pconfig = PROVIDER_REGISTRY["ollama-cloud"]
assert pconfig.id == "ollama-cloud"
assert pconfig.name == "Ollama Cloud"
assert pconfig.auth_type == "api_key"
assert pconfig.inference_base_url == "https://ollama.com/v1"
def test_ollama_cloud_env_vars(self):
pconfig = PROVIDER_REGISTRY["ollama-cloud"]
assert pconfig.api_key_env_vars == ("OLLAMA_API_KEY",)
assert pconfig.base_url_env_var == "OLLAMA_BASE_URL"
def test_ollama_cloud_base_url(self):
assert "ollama.com" in PROVIDER_REGISTRY["ollama-cloud"].inference_base_url
# ── Provider Aliases ──
PROVIDER_ENV_VARS = (
"OPENROUTER_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY",
"GOOGLE_API_KEY", "GEMINI_API_KEY", "OLLAMA_API_KEY",
"GLM_API_KEY", "ZAI_API_KEY", "KIMI_API_KEY",
"MINIMAX_API_KEY", "DEEPSEEK_API_KEY",
)
@pytest.fixture(autouse=True)
def _clean_provider_env(monkeypatch):
for var in PROVIDER_ENV_VARS:
monkeypatch.delenv(var, raising=False)
class TestOllamaCloudAliases:
def test_explicit_ollama_cloud(self):
assert resolve_provider("ollama-cloud") == "ollama-cloud"
def test_alias_ollama_underscore(self):
"""ollama_cloud (underscore) is the unambiguous cloud alias."""
assert resolve_provider("ollama_cloud") == "ollama-cloud"
def test_bare_ollama_stays_local(self):
"""Bare 'ollama' alias routes to 'custom' (local) — not cloud."""
assert resolve_provider("ollama") == "custom"
def test_models_py_aliases(self):
assert _PROVIDER_ALIASES.get("ollama_cloud") == "ollama-cloud"
# bare "ollama" stays local
assert _PROVIDER_ALIASES.get("ollama") == "custom"
def test_normalize_provider(self):
assert normalize_provider("ollama-cloud") == "ollama-cloud"
# ── Auto-detection ──
class TestOllamaCloudAutoDetection:
def test_auto_detects_ollama_api_key(self, monkeypatch):
monkeypatch.setenv("OLLAMA_API_KEY", "test-ollama-key")
assert resolve_provider("auto") == "ollama-cloud"
# ── Credential Resolution ──
class TestOllamaCloudCredentials:
def test_resolve_with_ollama_api_key(self, monkeypatch):
monkeypatch.setenv("OLLAMA_API_KEY", "ollama-secret")
creds = resolve_api_key_provider_credentials("ollama-cloud")
assert creds["provider"] == "ollama-cloud"
assert creds["api_key"] == "ollama-secret"
assert creds["base_url"] == "https://ollama.com/v1"
def test_resolve_with_custom_base_url(self, monkeypatch):
monkeypatch.setenv("OLLAMA_API_KEY", "key")
monkeypatch.setenv("OLLAMA_BASE_URL", "https://custom.ollama/v1")
creds = resolve_api_key_provider_credentials("ollama-cloud")
assert creds["base_url"] == "https://custom.ollama/v1"
def test_runtime_ollama_cloud(self, monkeypatch):
monkeypatch.setenv("OLLAMA_API_KEY", "ollama-key")
from hermes_cli.runtime_provider import resolve_runtime_provider
result = resolve_runtime_provider(requested="ollama-cloud")
assert result["provider"] == "ollama-cloud"
assert result["api_mode"] == "chat_completions"
assert result["api_key"] == "ollama-key"
assert result["base_url"] == "https://ollama.com/v1"
# ── Model Catalog (dynamic — no static list) ──
class TestOllamaCloudModelCatalog:
def test_no_static_model_list(self):
"""Ollama Cloud models are fetched dynamically — no static list to maintain."""
assert "ollama-cloud" not in _PROVIDER_MODELS
def test_provider_label(self):
assert "ollama-cloud" in _PROVIDER_LABELS
assert _PROVIDER_LABELS["ollama-cloud"] == "Ollama Cloud"
# ── Merged Model Discovery ──
class TestOllamaCloudMergedDiscovery:
def test_merges_live_and_models_dev(self, tmp_path, monkeypatch):
"""Live API models appear first, models.dev additions fill gaps."""
from hermes_cli.models import fetch_ollama_cloud_models
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("OLLAMA_API_KEY", "test-key")
mock_mdev = {
"ollama-cloud": {
"models": {
"glm-5": {"tool_call": True},
"kimi-k2.5": {"tool_call": True},
"nemotron-3-super": {"tool_call": True},
}
}
}
with patch("hermes_cli.models.fetch_api_models", return_value=["qwen3.5:397b", "glm-5"]), \
patch("agent.models_dev.fetch_models_dev", return_value=mock_mdev):
result = fetch_ollama_cloud_models(force_refresh=True)
# Live models first, then models.dev additions (deduped)
assert result[0] == "qwen3.5:397b" # from live API
assert result[1] == "glm-5" # from live API (also in models.dev)
assert "kimi-k2.5" in result # from models.dev only
assert "nemotron-3-super" in result # from models.dev only
assert result.count("glm-5") == 1 # no duplicates
def test_falls_back_to_models_dev_without_api_key(self, tmp_path, monkeypatch):
"""Without API key, only models.dev results are returned."""
from hermes_cli.models import fetch_ollama_cloud_models
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
mock_mdev = {
"ollama-cloud": {
"models": {
"glm-5": {"tool_call": True},
}
}
}
with patch("agent.models_dev.fetch_models_dev", return_value=mock_mdev):
result = fetch_ollama_cloud_models(force_refresh=True)
assert result == ["glm-5"]
def test_uses_disk_cache(self, tmp_path, monkeypatch):
"""Second call returns cached results without hitting APIs."""
from hermes_cli.models import fetch_ollama_cloud_models
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("OLLAMA_API_KEY", "test-key")
with patch("hermes_cli.models.fetch_api_models", return_value=["model-a"]) as mock_api, \
patch("agent.models_dev.fetch_models_dev", return_value={}):
first = fetch_ollama_cloud_models(force_refresh=True)
assert first == ["model-a"]
assert mock_api.call_count == 1
# Second call — should use disk cache, not call API
second = fetch_ollama_cloud_models()
assert second == ["model-a"]
assert mock_api.call_count == 1 # no extra API call
def test_force_refresh_bypasses_cache(self, tmp_path, monkeypatch):
"""force_refresh=True always hits the API even with fresh cache."""
from hermes_cli.models import fetch_ollama_cloud_models
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("OLLAMA_API_KEY", "test-key")
with patch("hermes_cli.models.fetch_api_models", return_value=["model-a"]) as mock_api, \
patch("agent.models_dev.fetch_models_dev", return_value={}):
fetch_ollama_cloud_models(force_refresh=True)
fetch_ollama_cloud_models(force_refresh=True)
assert mock_api.call_count == 2
def test_stale_cache_used_on_total_failure(self, tmp_path, monkeypatch):
"""If both API and models.dev fail, stale cache is returned."""
from hermes_cli.models import fetch_ollama_cloud_models, _save_ollama_cloud_cache
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("OLLAMA_API_KEY", "test-key")
# Pre-populate a stale cache
_save_ollama_cloud_cache(["stale-model"])
# Make the cache appear stale by backdating it
import json
cache_path = tmp_path / "ollama_cloud_models_cache.json"
with open(cache_path) as f:
data = json.load(f)
data["cached_at"] = 0 # epoch = very stale
with open(cache_path, "w") as f:
json.dump(data, f)
with patch("hermes_cli.models.fetch_api_models", return_value=None), \
patch("agent.models_dev.fetch_models_dev", return_value={}):
result = fetch_ollama_cloud_models(force_refresh=True)
assert result == ["stale-model"]
def test_empty_on_total_failure_no_cache(self, tmp_path, monkeypatch):
"""Returns empty list when everything fails and no cache exists."""
from hermes_cli.models import fetch_ollama_cloud_models
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.delenv("OLLAMA_API_KEY", raising=False)
with patch("agent.models_dev.fetch_models_dev", return_value={}):
result = fetch_ollama_cloud_models(force_refresh=True)
assert result == []
# ── Model Normalization ──
class TestOllamaCloudModelNormalization:
def test_passthrough_bare_name(self):
"""Ollama Cloud is a passthrough provider — model names used as-is."""
assert normalize_model_for_provider("qwen3.5:397b", "ollama-cloud") == "qwen3.5:397b"
def test_passthrough_with_tag(self):
assert normalize_model_for_provider("cogito-2.1:671b", "ollama-cloud") == "cogito-2.1:671b"
def test_passthrough_no_tag(self):
assert normalize_model_for_provider("glm-5", "ollama-cloud") == "glm-5"
# ── URL-to-Provider Mapping ──
class TestOllamaCloudUrlMapping:
def test_url_to_provider(self):
assert _URL_TO_PROVIDER.get("ollama.com") == "ollama-cloud"
def test_provider_prefix_canonical(self):
assert "ollama-cloud" in _PROVIDER_PREFIXES
def test_provider_prefix_alias(self):
assert "ollama" in _PROVIDER_PREFIXES
# ── models.dev Integration ──
class TestOllamaCloudModelsDev:
def test_ollama_cloud_mapped(self):
assert PROVIDER_TO_MODELS_DEV.get("ollama-cloud") == "ollama-cloud"
def test_list_agentic_models_with_mock_data(self):
"""list_agentic_models filters correctly from mock models.dev data."""
mock_data = {
"ollama-cloud": {
"models": {
"qwen3.5:397b": {"tool_call": True},
"glm-5": {"tool_call": True},
"nemotron-3-nano:30b": {"tool_call": True},
"some-embedding:latest": {"tool_call": False},
}
}
}
with patch("agent.models_dev.fetch_models_dev", return_value=mock_data):
result = list_agentic_models("ollama-cloud")
assert "qwen3.5:397b" in result
assert "glm-5" in result
assert "nemotron-3-nano:30b" in result
assert "some-embedding:latest" not in result # no tool_call
# ── Agent Init (no SyntaxError) ──
class TestOllamaCloudAgentInit:
def test_agent_imports_without_error(self):
"""Verify run_agent.py has no SyntaxError."""
import importlib
import run_agent
importlib.reload(run_agent)
def test_ollama_cloud_agent_uses_chat_completions(self, monkeypatch):
"""Ollama Cloud falls through to chat_completions — no special elif needed."""
monkeypatch.setenv("OLLAMA_API_KEY", "test-key")
with patch("run_agent.OpenAI") as mock_openai:
mock_openai.return_value = MagicMock()
from run_agent import AIAgent
agent = AIAgent(
model="qwen3.5:397b",
provider="ollama-cloud",
api_key="test-key",
base_url="https://ollama.com/v1",
)
assert agent.api_mode == "chat_completions"
assert agent.provider == "ollama-cloud"
# ── providers.py New System ──
class TestOllamaCloudProvidersNew:
def test_overlay_exists(self):
from hermes_cli.providers import HERMES_OVERLAYS
assert "ollama-cloud" in HERMES_OVERLAYS
overlay = HERMES_OVERLAYS["ollama-cloud"]
assert overlay.transport == "openai_chat"
assert overlay.base_url_env_var == "OLLAMA_BASE_URL"
def test_alias_resolves(self):
from hermes_cli.providers import normalize_provider as np
assert np("ollama") == "custom" # bare "ollama" = local
assert np("ollama-cloud") == "ollama-cloud"
def test_label_override(self):
from hermes_cli.providers import _LABEL_OVERRIDES
assert _LABEL_OVERRIDES.get("ollama-cloud") == "Ollama Cloud"
def test_get_label(self):
from hermes_cli.providers import get_label
assert get_label("ollama-cloud") == "Ollama Cloud"
def test_get_provider(self):
from hermes_cli.providers import get_provider
pdef = get_provider("ollama-cloud")
assert pdef is not None
assert pdef.id == "ollama-cloud"
assert pdef.transport == "openai_chat"
# ── Auxiliary Model ──
class TestOllamaCloudAuxiliary:
def test_aux_model_defined(self):
from agent.auxiliary_client import _API_KEY_PROVIDER_AUX_MODELS
assert "ollama-cloud" in _API_KEY_PROVIDER_AUX_MODELS
assert _API_KEY_PROVIDER_AUX_MODELS["ollama-cloud"] == "nemotron-3-nano:30b"

View File

@@ -18,6 +18,8 @@ from hermes_cli.plugins import (
PluginManager,
PluginManifest,
get_plugin_manager,
get_plugin_command_handler,
get_plugin_commands,
get_pre_tool_call_block_message,
discover_plugins,
invoke_hook,
@@ -605,7 +607,292 @@ class TestPreLlmCallTargetRouting:
assert "plain text C" in _plugin_user_context
# NOTE: TestPluginCommands removed register_command() was never implemented
# in PluginContext (hermes_cli/plugins.py). The tests referenced _plugin_commands,
# commands_registered, get_plugin_command_handler, and GATEWAY_KNOWN_COMMANDS
# integration — all of which are unimplemented features.
# ── TestPluginCommands ────────────────────────────────────────────────────
class TestPluginCommands:
"""Tests for plugin slash command registration via register_command()."""
def test_register_command_basic(self):
"""register_command() stores handler, description, and plugin name."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
handler = lambda args: f"echo {args}"
ctx.register_command("mycmd", handler, description="My custom command")
assert "mycmd" in mgr._plugin_commands
entry = mgr._plugin_commands["mycmd"]
assert entry["handler"] is handler
assert entry["description"] == "My custom command"
assert entry["plugin"] == "test-plugin"
def test_register_command_normalizes_name(self):
"""Names are lowercased, stripped, and leading slashes removed."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
ctx.register_command("/MyCmd ", lambda a: a, description="test")
assert "mycmd" in mgr._plugin_commands
assert "/MyCmd " not in mgr._plugin_commands
def test_register_command_empty_name_rejected(self, caplog):
"""Empty name after normalization is rejected with a warning."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
with caplog.at_level(logging.WARNING):
ctx.register_command("", lambda a: a)
assert len(mgr._plugin_commands) == 0
assert "empty name" in caplog.text
def test_register_command_builtin_conflict_rejected(self, caplog):
"""Commands that conflict with built-in names are rejected."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
with caplog.at_level(logging.WARNING):
ctx.register_command("help", lambda a: a)
assert "help" not in mgr._plugin_commands
assert "conflicts" in caplog.text.lower()
def test_register_command_default_description(self):
"""Missing description defaults to 'Plugin command'."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
ctx.register_command("status-cmd", lambda a: a)
assert mgr._plugin_commands["status-cmd"]["description"] == "Plugin command"
def test_get_plugin_command_handler_found(self):
"""get_plugin_command_handler() returns the handler for a registered command."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
handler = lambda args: f"result: {args}"
ctx.register_command("mycmd", handler, description="test")
with patch("hermes_cli.plugins._plugin_manager", mgr):
result = get_plugin_command_handler("mycmd")
assert result is handler
def test_get_plugin_command_handler_not_found(self):
"""get_plugin_command_handler() returns None for unregistered commands."""
mgr = PluginManager()
with patch("hermes_cli.plugins._plugin_manager", mgr):
assert get_plugin_command_handler("nonexistent") is None
def test_get_plugin_commands_returns_dict(self):
"""get_plugin_commands() returns the full commands dict."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
ctx.register_command("cmd-a", lambda a: a, description="A")
ctx.register_command("cmd-b", lambda a: a, description="B")
with patch("hermes_cli.plugins._plugin_manager", mgr):
cmds = get_plugin_commands()
assert "cmd-a" in cmds
assert "cmd-b" in cmds
assert cmds["cmd-a"]["description"] == "A"
def test_commands_tracked_on_loaded_plugin(self, tmp_path, monkeypatch):
"""Commands registered during discover_and_load() are tracked on LoadedPlugin."""
plugins_dir = tmp_path / "hermes_test" / "plugins"
_make_plugin_dir(
plugins_dir, "cmd-plugin",
register_body=(
'ctx.register_command("mycmd", lambda a: "ok", description="Test")'
),
)
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
mgr = PluginManager()
mgr.discover_and_load()
loaded = mgr._plugins["cmd-plugin"]
assert loaded.enabled
assert "mycmd" in loaded.commands_registered
def test_commands_in_list_plugins_output(self, tmp_path, monkeypatch):
"""list_plugins() includes command count."""
plugins_dir = tmp_path / "hermes_test" / "plugins"
_make_plugin_dir(
plugins_dir, "cmd-plugin",
register_body=(
'ctx.register_command("mycmd", lambda a: "ok", description="Test")'
),
)
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
mgr = PluginManager()
mgr.discover_and_load()
info = mgr.list_plugins()
assert len(info) == 1
assert info[0]["commands"] == 1
def test_handler_receives_raw_args(self):
"""The handler is called with the raw argument string."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
received = []
ctx.register_command("echo", lambda args: received.append(args) or "ok")
handler = mgr._plugin_commands["echo"]["handler"]
handler("hello world")
assert received == ["hello world"]
def test_multiple_plugins_register_different_commands(self):
"""Multiple plugins can each register their own commands."""
mgr = PluginManager()
for plugin_name, cmd_name in [("plugin-a", "cmd-a"), ("plugin-b", "cmd-b")]:
manifest = PluginManifest(name=plugin_name, source="user")
ctx = PluginContext(manifest, mgr)
ctx.register_command(cmd_name, lambda a: a, description=f"From {plugin_name}")
assert "cmd-a" in mgr._plugin_commands
assert "cmd-b" in mgr._plugin_commands
assert mgr._plugin_commands["cmd-a"]["plugin"] == "plugin-a"
assert mgr._plugin_commands["cmd-b"]["plugin"] == "plugin-b"
# ── TestPluginDispatchTool ────────────────────────────────────────────────
class TestPluginDispatchTool:
"""Tests for PluginContext.dispatch_tool() — tool dispatch with agent context."""
def test_dispatch_tool_calls_registry(self):
"""dispatch_tool() delegates to registry.dispatch()."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
mock_registry = MagicMock()
mock_registry.dispatch.return_value = '{"result": "ok"}'
with patch("hermes_cli.plugins.PluginContext.dispatch_tool.__module__", "hermes_cli.plugins"):
with patch.dict("sys.modules", {}):
with patch("tools.registry.registry", mock_registry):
result = ctx.dispatch_tool("web_search", {"query": "test"})
assert result == '{"result": "ok"}'
def test_dispatch_tool_injects_parent_agent_from_cli_ref(self):
"""When _cli_ref has an agent, it's passed as parent_agent."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
mock_agent = MagicMock()
mock_cli = MagicMock()
mock_cli.agent = mock_agent
mgr._cli_ref = mock_cli
mock_registry = MagicMock()
mock_registry.dispatch.return_value = '{"ok": true}'
with patch("tools.registry.registry", mock_registry):
ctx.dispatch_tool("delegate_task", {"goal": "test"})
mock_registry.dispatch.assert_called_once()
call_kwargs = mock_registry.dispatch.call_args
assert call_kwargs[1].get("parent_agent") is mock_agent
def test_dispatch_tool_no_parent_agent_when_no_cli_ref(self):
"""When _cli_ref is None (gateway mode), no parent_agent is injected."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
mgr._cli_ref = None
mock_registry = MagicMock()
mock_registry.dispatch.return_value = '{"ok": true}'
with patch("tools.registry.registry", mock_registry):
ctx.dispatch_tool("delegate_task", {"goal": "test"})
call_kwargs = mock_registry.dispatch.call_args
assert "parent_agent" not in call_kwargs[1]
def test_dispatch_tool_no_parent_agent_when_agent_is_none(self):
"""When cli_ref exists but agent is None (not yet initialized), skip parent_agent."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
mock_cli = MagicMock()
mock_cli.agent = None
mgr._cli_ref = mock_cli
mock_registry = MagicMock()
mock_registry.dispatch.return_value = '{"ok": true}'
with patch("tools.registry.registry", mock_registry):
ctx.dispatch_tool("delegate_task", {"goal": "test"})
call_kwargs = mock_registry.dispatch.call_args
assert "parent_agent" not in call_kwargs[1]
def test_dispatch_tool_respects_explicit_parent_agent(self):
"""Explicit parent_agent kwarg is not overwritten by _cli_ref.agent."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
cli_agent = MagicMock(name="cli_agent")
mock_cli = MagicMock()
mock_cli.agent = cli_agent
mgr._cli_ref = mock_cli
explicit_agent = MagicMock(name="explicit_agent")
mock_registry = MagicMock()
mock_registry.dispatch.return_value = '{"ok": true}'
with patch("tools.registry.registry", mock_registry):
ctx.dispatch_tool("delegate_task", {"goal": "test"}, parent_agent=explicit_agent)
call_kwargs = mock_registry.dispatch.call_args
assert call_kwargs[1]["parent_agent"] is explicit_agent
def test_dispatch_tool_forwards_extra_kwargs(self):
"""Extra kwargs are forwarded to registry.dispatch()."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
mgr._cli_ref = None
mock_registry = MagicMock()
mock_registry.dispatch.return_value = '{"ok": true}'
with patch("tools.registry.registry", mock_registry):
ctx.dispatch_tool("some_tool", {"x": 1}, task_id="test-123")
call_kwargs = mock_registry.dispatch.call_args
assert call_kwargs[1]["task_id"] == "test-123"
def test_dispatch_tool_returns_json_string(self):
"""dispatch_tool() returns the raw JSON string from the registry."""
mgr = PluginManager()
manifest = PluginManifest(name="test-plugin", source="user")
ctx = PluginContext(manifest, mgr)
mgr._cli_ref = None
mock_registry = MagicMock()
mock_registry.dispatch.return_value = '{"error": "Unknown tool: fake"}'
with patch("tools.registry.registry", mock_registry):
result = ctx.dispatch_tool("fake", {})
assert '"error"' in result

View File

@@ -0,0 +1,56 @@
"""Tests for plugins/memory/honcho/cli.py."""
from types import SimpleNamespace
class TestCmdStatus:
def test_reports_connection_failure_when_session_setup_fails(self, monkeypatch, capsys, tmp_path):
import plugins.memory.honcho.cli as honcho_cli
cfg_path = tmp_path / "honcho.json"
cfg_path.write_text("{}")
class FakeConfig:
enabled = True
api_key = "root-key"
workspace_id = "hermes"
host = "hermes"
base_url = None
ai_peer = "hermes"
peer_name = "eri"
recall_mode = "hybrid"
user_observe_me = True
user_observe_others = False
ai_observe_me = False
ai_observe_others = True
write_frequency = "async"
session_strategy = "per-session"
context_tokens = 800
def resolve_session_name(self):
return "hermes"
monkeypatch.setattr(honcho_cli, "_read_config", lambda: {"apiKey": "***"})
monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
monkeypatch.setattr(honcho_cli, "_active_profile_name", lambda: "default")
monkeypatch.setattr(
"plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
lambda host=None: FakeConfig(),
)
monkeypatch.setattr(
"plugins.memory.honcho.client.get_honcho_client",
lambda cfg: object(),
)
def _boom(hcfg, client):
raise RuntimeError("Invalid API key")
monkeypatch.setattr(honcho_cli, "_show_peer_cards", _boom)
monkeypatch.setitem(__import__("sys").modules, "honcho", SimpleNamespace())
honcho_cli.cmd_status(SimpleNamespace(all=False))
out = capsys.readouterr().out
assert "FAILED (Invalid API key)" in out
assert "Connection... OK" not in out

View File

@@ -1,5 +1,6 @@
"""Tests for plugins/memory/honcho/client.py — Honcho client configuration."""
import importlib.util
import json
import os
from pathlib import Path
@@ -25,6 +26,7 @@ class TestHonchoClientConfigDefaults:
assert config.workspace_id == "hermes"
assert config.api_key is None
assert config.environment == "production"
assert config.timeout is None
assert config.enabled is False
assert config.save_messages is True
assert config.session_strategy == "per-directory"
@@ -76,6 +78,11 @@ class TestFromEnv:
assert config.base_url == "http://localhost:8000"
assert config.enabled is True
def test_reads_timeout_from_env(self):
with patch.dict(os.environ, {"HONCHO_TIMEOUT": "90"}, clear=True):
config = HonchoClientConfig.from_env()
assert config.timeout == 90.0
class TestFromGlobalConfig:
def test_missing_config_falls_back_to_env(self, tmp_path):
@@ -87,10 +94,10 @@ class TestFromGlobalConfig:
assert config.enabled is False
assert config.api_key is None
def test_reads_full_config(self, tmp_path):
def test_reads_full_config(self, tmp_path, monkeypatch):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({
"apiKey": "my-honcho-key",
"apiKey": "***",
"workspace": "my-workspace",
"environment": "staging",
"peerName": "alice",
@@ -108,9 +115,11 @@ class TestFromGlobalConfig:
}
}
}))
# Isolate from real ~/.hermes/honcho.json
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "isolated"))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.api_key == "my-honcho-key"
assert config.api_key == "***"
# Host block workspace overrides root workspace
assert config.workspace_id == "override-ws"
assert config.ai_peer == "override-ai"
@@ -154,10 +163,31 @@ class TestFromGlobalConfig:
def test_session_strategy_default_from_global_config(self, tmp_path):
"""from_global_config with no sessionStrategy should match dataclass default."""
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"apiKey": "key"}))
config_file.write_text(json.dumps({"apiKey": "***"}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.session_strategy == "per-directory"
def test_context_tokens_default_is_none(self, tmp_path):
"""Default context_tokens should be None (uncapped) unless explicitly set."""
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"apiKey": "***"}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.context_tokens is None
def test_context_tokens_explicit_sets_cap(self, tmp_path):
"""Explicit contextTokens in config sets the cap."""
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"apiKey": "***", "contextTokens": 1200}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.context_tokens == 1200
def test_context_tokens_explicit_overrides_default(self, tmp_path):
"""Explicit contextTokens in config should override the default."""
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"apiKey": "***", "contextTokens": 2000}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.context_tokens == 2000
def test_context_tokens_host_block_wins(self, tmp_path):
"""Host block contextTokens should override root."""
config_file = tmp_path / "config.json"
@@ -232,6 +262,20 @@ class TestFromGlobalConfig:
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.base_url == "http://root:9000"
def test_timeout_from_config_root(self, tmp_path):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"timeout": 75}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.timeout == 75.0
def test_request_timeout_alias_from_config_root(self, tmp_path):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"requestTimeout": "82.5"}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.timeout == 82.5
class TestResolveSessionName:
def test_manual_override(self):
@@ -333,13 +377,14 @@ class TestResolveConfigPath:
hermes_home.mkdir()
local_cfg = hermes_home / "honcho.json"
local_cfg.write_text(json.dumps({
"apiKey": "local-key",
"apiKey": "***",
"workspace": "local-ws",
}))
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}):
with patch.dict(os.environ, {"HERMES_HOME": str(hermes_home)}), \
patch.object(Path, "home", return_value=tmp_path):
config = HonchoClientConfig.from_global_config()
assert config.api_key == "local-key"
assert config.api_key == "***"
assert config.workspace_id == "local-ws"
@@ -500,46 +545,115 @@ class TestObservationModeMigration:
assert cfg.ai_observe_others is True
class TestInitOnSessionStart:
"""Tests for the initOnSessionStart config field."""
class TestGetHonchoClient:
def teardown_method(self):
reset_honcho_client()
def test_default_is_false(self):
@pytest.mark.skipif(
not importlib.util.find_spec("honcho"),
reason="honcho SDK not installed"
)
def test_passes_timeout_from_config(self):
fake_honcho = MagicMock(name="Honcho")
cfg = HonchoClientConfig(
api_key="test-key",
timeout=91.0,
workspace_id="hermes",
environment="production",
)
with patch("honcho.Honcho", return_value=fake_honcho) as mock_honcho:
client = get_honcho_client(cfg)
assert client is fake_honcho
mock_honcho.assert_called_once()
assert mock_honcho.call_args.kwargs["timeout"] == 91.0
@pytest.mark.skipif(
not importlib.util.find_spec("honcho"),
reason="honcho SDK not installed"
)
def test_hermes_config_timeout_override_used_when_config_timeout_missing(self):
fake_honcho = MagicMock(name="Honcho")
cfg = HonchoClientConfig(
api_key="test-key",
workspace_id="hermes",
environment="production",
)
with patch("honcho.Honcho", return_value=fake_honcho) as mock_honcho, \
patch("hermes_cli.config.load_config", return_value={"honcho": {"timeout": 88}}):
client = get_honcho_client(cfg)
assert client is fake_honcho
mock_honcho.assert_called_once()
assert mock_honcho.call_args.kwargs["timeout"] == 88.0
@pytest.mark.skipif(
not importlib.util.find_spec("honcho"),
reason="honcho SDK not installed"
)
def test_hermes_request_timeout_alias_used(self):
fake_honcho = MagicMock(name="Honcho")
cfg = HonchoClientConfig(
api_key="test-key",
workspace_id="hermes",
environment="production",
)
with patch("honcho.Honcho", return_value=fake_honcho) as mock_honcho, \
patch("hermes_cli.config.load_config", return_value={"honcho": {"request_timeout": "77.5"}}):
client = get_honcho_client(cfg)
assert client is fake_honcho
mock_honcho.assert_called_once()
assert mock_honcho.call_args.kwargs["timeout"] == 77.5
class TestResolveSessionNameGatewayKey:
"""Regression tests for gateway_session_key priority in resolve_session_name.
Ensures gateway platforms get stable per-chat Honcho sessions even when
sessionStrategy=per-session would otherwise create ephemeral sessions.
Regression: plugin refactor 924bc67e dropped gateway key plumbing.
"""
def test_gateway_key_overrides_per_session_strategy(self):
"""gateway_session_key must win over per-session session_id."""
config = HonchoClientConfig(session_strategy="per-session")
result = config.resolve_session_name(
session_id="20260412_171002_69bb38",
gateway_session_key="agent:main:telegram:dm:8439114563",
)
assert result == "agent-main-telegram-dm-8439114563"
def test_session_title_still_wins_over_gateway_key(self):
"""Explicit /title remap takes priority over gateway_session_key."""
config = HonchoClientConfig(session_strategy="per-session")
result = config.resolve_session_name(
session_title="my-custom-title",
session_id="20260412_171002_69bb38",
gateway_session_key="agent:main:telegram:dm:8439114563",
)
assert result == "my-custom-title"
def test_per_session_fallback_without_gateway_key(self):
"""Without gateway_session_key, per-session returns session_id (CLI path)."""
config = HonchoClientConfig(session_strategy="per-session")
result = config.resolve_session_name(
session_id="20260412_171002_69bb38",
gateway_session_key=None,
)
assert result == "20260412_171002_69bb38"
def test_gateway_key_sanitizes_special_chars(self):
"""Colons and other non-alphanumeric chars are replaced with hyphens."""
config = HonchoClientConfig()
assert config.init_on_session_start is False
def test_root_level_true(self, tmp_path):
cfg_file = tmp_path / "config.json"
cfg_file.write_text(json.dumps({
"apiKey": "k",
"initOnSessionStart": True,
}))
cfg = HonchoClientConfig.from_global_config(config_path=cfg_file)
assert cfg.init_on_session_start is True
def test_host_block_overrides_root(self, tmp_path):
cfg_file = tmp_path / "config.json"
cfg_file.write_text(json.dumps({
"apiKey": "k",
"initOnSessionStart": True,
"hosts": {"hermes": {"initOnSessionStart": False}},
}))
cfg = HonchoClientConfig.from_global_config(config_path=cfg_file)
assert cfg.init_on_session_start is False
def test_host_block_true_overrides_root_absent(self, tmp_path):
cfg_file = tmp_path / "config.json"
cfg_file.write_text(json.dumps({
"apiKey": "k",
"hosts": {"hermes": {"initOnSessionStart": True}},
}))
cfg = HonchoClientConfig.from_global_config(config_path=cfg_file)
assert cfg.init_on_session_start is True
def test_absent_everywhere_defaults_false(self, tmp_path):
cfg_file = tmp_path / "config.json"
cfg_file.write_text(json.dumps({"apiKey": "k"}))
cfg = HonchoClientConfig.from_global_config(config_path=cfg_file)
assert cfg.init_on_session_start is False
result = config.resolve_session_name(
gateway_session_key="agent:main:telegram:dm:8439114563",
)
assert result == "agent-main-telegram-dm-8439114563"
assert ":" not in result
class TestResetHonchoClient:
@@ -549,3 +663,91 @@ class TestResetHonchoClient:
assert mod._honcho_client is not None
reset_honcho_client()
assert mod._honcho_client is None
class TestDialecticDepthParsing:
"""Tests for _parse_dialectic_depth and _parse_dialectic_depth_levels."""
def test_default_depth_is_1(self, tmp_path):
"""Default dialecticDepth should be 1."""
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"apiKey": "***"}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth == 1
def test_depth_from_root(self, tmp_path):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"apiKey": "***", "dialecticDepth": 2}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth == 2
def test_depth_host_block_wins(self, tmp_path):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({
"apiKey": "***",
"dialecticDepth": 1,
"hosts": {"hermes": {"dialecticDepth": 3}},
}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth == 3
def test_depth_clamped_high(self, tmp_path):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"apiKey": "***", "dialecticDepth": 10}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth == 3
def test_depth_clamped_low(self, tmp_path):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"apiKey": "***", "dialecticDepth": -1}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth == 1
def test_depth_levels_default_none(self, tmp_path):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"apiKey": "***"}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth_levels is None
def test_depth_levels_from_config(self, tmp_path):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({
"apiKey": "***",
"dialecticDepth": 2,
"dialecticDepthLevels": ["minimal", "high"],
}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth_levels == ["minimal", "high"]
def test_depth_levels_padded_if_short(self, tmp_path):
"""Array shorter than depth gets padded with 'low'."""
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({
"apiKey": "***",
"dialecticDepth": 3,
"dialecticDepthLevels": ["high"],
}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth_levels == ["high", "low", "low"]
def test_depth_levels_truncated_if_long(self, tmp_path):
"""Array longer than depth gets truncated."""
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({
"apiKey": "***",
"dialecticDepth": 1,
"dialecticDepthLevels": ["high", "max", "medium"],
}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth_levels == ["high"]
def test_depth_levels_invalid_values_default_to_low(self, tmp_path):
"""Invalid reasoning levels in the array fall back to 'low'."""
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({
"apiKey": "***",
"dialecticDepth": 2,
"dialecticDepthLevels": ["invalid", "high"],
}))
config = HonchoClientConfig.from_global_config(config_path=config_file)
assert config.dialectic_depth_levels == ["low", "high"]

View File

@@ -205,27 +205,62 @@ class TestPeerLookupHelpers:
def test_get_peer_card_uses_direct_peer_lookup(self):
mgr, session = self._make_cached_manager()
user_peer = MagicMock()
user_peer.get_card.return_value = ["Name: Robert"]
mgr._get_or_create_peer = MagicMock(return_value=user_peer)
assistant_peer = MagicMock()
assistant_peer.get_card.return_value = ["Name: Robert"]
mgr._get_or_create_peer = MagicMock(return_value=assistant_peer)
assert mgr.get_peer_card(session.key) == ["Name: Robert"]
user_peer.get_card.assert_called_once_with()
assistant_peer.get_card.assert_called_once_with(target=session.user_peer_id)
def test_search_context_uses_peer_context_response(self):
def test_search_context_uses_assistant_perspective_with_target(self):
mgr, session = self._make_cached_manager()
user_peer = MagicMock()
user_peer.context.return_value = SimpleNamespace(
assistant_peer = MagicMock()
assistant_peer.context.return_value = SimpleNamespace(
representation="Robert runs neuralancer",
peer_card=["Location: Melbourne"],
)
mgr._get_or_create_peer = MagicMock(return_value=user_peer)
mgr._get_or_create_peer = MagicMock(return_value=assistant_peer)
result = mgr.search_context(session.key, "neuralancer")
assert "Robert runs neuralancer" in result
assert "- Location: Melbourne" in result
user_peer.context.assert_called_once_with(search_query="neuralancer")
assistant_peer.context.assert_called_once_with(
target=session.user_peer_id,
search_query="neuralancer",
)
def test_search_context_unified_mode_uses_user_self_context(self):
mgr, session = self._make_cached_manager()
mgr._ai_observe_others = False
user_peer = MagicMock()
user_peer.context.return_value = SimpleNamespace(
representation="Unified self context",
peer_card=["Name: Robert"],
)
mgr._get_or_create_peer = MagicMock(return_value=user_peer)
result = mgr.search_context(session.key, "self")
assert "Unified self context" in result
user_peer.context.assert_called_once_with(search_query="self")
def test_search_context_accepts_explicit_ai_peer_id(self):
mgr, session = self._make_cached_manager()
ai_peer = MagicMock()
ai_peer.context.return_value = SimpleNamespace(
representation="Assistant self context",
peer_card=["Role: Assistant"],
)
mgr._get_or_create_peer = MagicMock(return_value=ai_peer)
result = mgr.search_context(session.key, "assistant", peer=session.assistant_peer_id)
assert "Assistant self context" in result
ai_peer.context.assert_called_once_with(
target=session.assistant_peer_id,
search_query="assistant",
)
def test_get_prefetch_context_fetches_user_and_ai_from_peer_api(self):
mgr, session = self._make_cached_manager()
@@ -235,9 +270,15 @@ class TestPeerLookupHelpers:
peer_card=["Name: Robert"],
)
ai_peer = MagicMock()
ai_peer.context.return_value = SimpleNamespace(
representation="AI representation",
peer_card=["Owner: Robert"],
ai_peer.context.side_effect = lambda **kwargs: SimpleNamespace(
representation=(
"AI representation" if kwargs.get("target") == session.assistant_peer_id
else "Mixed representation"
),
peer_card=(
["Role: Assistant"] if kwargs.get("target") == session.assistant_peer_id
else ["Name: Robert"]
),
)
mgr._get_or_create_peer = MagicMock(side_effect=[user_peer, ai_peer])
@@ -247,17 +288,23 @@ class TestPeerLookupHelpers:
"representation": "User representation",
"card": "Name: Robert",
"ai_representation": "AI representation",
"ai_card": "Owner: Robert",
"ai_card": "Role: Assistant",
}
user_peer.context.assert_called_once_with()
ai_peer.context.assert_called_once_with()
user_peer.context.assert_called_once_with(target=session.user_peer_id)
ai_peer.context.assert_called_once_with(target=session.assistant_peer_id)
def test_get_ai_representation_uses_peer_api(self):
mgr, session = self._make_cached_manager()
ai_peer = MagicMock()
ai_peer.context.return_value = SimpleNamespace(
representation="AI representation",
peer_card=["Owner: Robert"],
ai_peer.context.side_effect = lambda **kwargs: SimpleNamespace(
representation=(
"AI representation" if kwargs.get("target") == session.assistant_peer_id
else "Mixed representation"
),
peer_card=(
["Role: Assistant"] if kwargs.get("target") == session.assistant_peer_id
else ["Name: Robert"]
),
)
mgr._get_or_create_peer = MagicMock(return_value=ai_peer)
@@ -265,9 +312,167 @@ class TestPeerLookupHelpers:
assert result == {
"representation": "AI representation",
"card": "Owner: Robert",
"card": "Role: Assistant",
}
ai_peer.context.assert_called_once_with()
ai_peer.context.assert_called_once_with(target=session.assistant_peer_id)
def test_create_conclusion_defaults_to_user_target(self):
mgr, session = self._make_cached_manager()
assistant_peer = MagicMock()
scope = MagicMock()
assistant_peer.conclusions_of.return_value = scope
mgr._get_or_create_peer = MagicMock(return_value=assistant_peer)
ok = mgr.create_conclusion(session.key, "User prefers dark mode")
assert ok is True
assistant_peer.conclusions_of.assert_called_once_with(session.user_peer_id)
scope.create.assert_called_once_with([{
"content": "User prefers dark mode",
"session_id": session.honcho_session_id,
}])
def test_create_conclusion_can_target_ai_peer(self):
mgr, session = self._make_cached_manager()
assistant_peer = MagicMock()
scope = MagicMock()
assistant_peer.conclusions_of.return_value = scope
mgr._get_or_create_peer = MagicMock(return_value=assistant_peer)
ok = mgr.create_conclusion(session.key, "Assistant prefers terse summaries", peer="ai")
assert ok is True
assistant_peer.conclusions_of.assert_called_once_with(session.assistant_peer_id)
scope.create.assert_called_once_with([{
"content": "Assistant prefers terse summaries",
"session_id": session.honcho_session_id,
}])
def test_create_conclusion_accepts_explicit_user_peer_id(self):
mgr, session = self._make_cached_manager()
assistant_peer = MagicMock()
scope = MagicMock()
assistant_peer.conclusions_of.return_value = scope
mgr._get_or_create_peer = MagicMock(return_value=assistant_peer)
ok = mgr.create_conclusion(session.key, "Robert prefers vinyl", peer=session.user_peer_id)
assert ok is True
assistant_peer.conclusions_of.assert_called_once_with(session.user_peer_id)
scope.create.assert_called_once_with([{
"content": "Robert prefers vinyl",
"session_id": session.honcho_session_id,
}])
class TestConcludeToolDispatch:
def test_honcho_conclude_defaults_to_user_peer(self):
provider = HonchoMemoryProvider()
provider._session_initialized = True
provider._session_key = "telegram:123"
provider._manager = MagicMock()
provider._manager.create_conclusion.return_value = True
result = provider.handle_tool_call(
"honcho_conclude",
{"conclusion": "User prefers dark mode"},
)
assert "Conclusion saved for user" in result
provider._manager.create_conclusion.assert_called_once_with(
"telegram:123",
"User prefers dark mode",
peer="user",
)
def test_honcho_conclude_can_target_ai_peer(self):
provider = HonchoMemoryProvider()
provider._session_initialized = True
provider._session_key = "telegram:123"
provider._manager = MagicMock()
provider._manager.create_conclusion.return_value = True
result = provider.handle_tool_call(
"honcho_conclude",
{"conclusion": "Assistant likes terse replies", "peer": "ai"},
)
assert "Conclusion saved for ai" in result
provider._manager.create_conclusion.assert_called_once_with(
"telegram:123",
"Assistant likes terse replies",
peer="ai",
)
def test_honcho_profile_can_target_explicit_peer_id(self):
provider = HonchoMemoryProvider()
provider._session_initialized = True
provider._session_key = "telegram:123"
provider._manager = MagicMock()
provider._manager.get_peer_card.return_value = ["Role: Assistant"]
result = provider.handle_tool_call(
"honcho_profile",
{"peer": "hermes"},
)
assert "Role: Assistant" in result
provider._manager.get_peer_card.assert_called_once_with("telegram:123", peer="hermes")
def test_honcho_search_can_target_explicit_peer_id(self):
provider = HonchoMemoryProvider()
provider._session_initialized = True
provider._session_key = "telegram:123"
provider._manager = MagicMock()
provider._manager.search_context.return_value = "Assistant self context"
result = provider.handle_tool_call(
"honcho_search",
{"query": "assistant", "peer": "hermes"},
)
assert "Assistant self context" in result
provider._manager.search_context.assert_called_once_with(
"telegram:123",
"assistant",
max_tokens=800,
peer="hermes",
)
def test_honcho_reasoning_can_target_explicit_peer_id(self):
provider = HonchoMemoryProvider()
provider._session_initialized = True
provider._session_key = "telegram:123"
provider._manager = MagicMock()
provider._manager.dialectic_query.return_value = "Assistant answer"
result = provider.handle_tool_call(
"honcho_reasoning",
{"query": "who are you", "peer": "hermes"},
)
assert "Assistant answer" in result
provider._manager.dialectic_query.assert_called_once_with(
"telegram:123",
"who are you",
reasoning_level=None,
peer="hermes",
)
def test_honcho_conclude_missing_both_params_returns_error(self):
"""Calling honcho_conclude with neither conclusion nor delete_id returns a tool error."""
import json
provider = HonchoMemoryProvider()
provider._session_initialized = True
provider._session_key = "telegram:123"
provider._manager = MagicMock()
result = provider.handle_tool_call("honcho_conclude", {})
parsed = json.loads(result)
assert "error" in parsed or "Missing required" in parsed.get("result", "")
provider._manager.create_conclusion.assert_not_called()
provider._manager.delete_conclusion.assert_not_called()
# ---------------------------------------------------------------------------
@@ -366,6 +571,54 @@ class TestToolsModeInitBehavior:
assert cfg.peer_name == "8439114563"
class TestPerSessionMigrateGuard:
"""Verify migrate_memory_files is skipped under per-session strategy.
per-session creates a fresh Honcho session every Hermes run. Uploading
MEMORY.md/USER.md/SOUL.md to each short-lived session floods the backend
with duplicate content. The guard was added to prevent orphan sessions
containing only <prior_memory_file> wrappers.
"""
def _make_provider_with_strategy(self, strategy, init_on_session_start=True):
"""Create a HonchoMemoryProvider and track migrate_memory_files calls."""
from plugins.memory.honcho.client import HonchoClientConfig
from unittest.mock import patch, MagicMock
cfg = HonchoClientConfig(
api_key="test-key",
enabled=True,
recall_mode="tools",
init_on_session_start=init_on_session_start,
session_strategy=strategy,
)
provider = HonchoMemoryProvider()
mock_manager = MagicMock()
mock_session = MagicMock()
mock_session.messages = [] # empty = new session → triggers migration path
mock_manager.get_or_create.return_value = mock_session
with patch("plugins.memory.honcho.client.HonchoClientConfig.from_global_config", return_value=cfg), \
patch("plugins.memory.honcho.client.get_honcho_client", return_value=MagicMock()), \
patch("plugins.memory.honcho.session.HonchoSessionManager", return_value=mock_manager), \
patch("hermes_constants.get_hermes_home", return_value=MagicMock()):
provider.initialize(session_id="test-session-001")
return provider, mock_manager
def test_migrate_skipped_for_per_session(self):
"""per-session strategy must NOT call migrate_memory_files."""
_, mock_manager = self._make_provider_with_strategy("per-session")
mock_manager.migrate_memory_files.assert_not_called()
def test_migrate_runs_for_per_directory(self):
"""per-directory strategy with empty session SHOULD call migrate_memory_files."""
_, mock_manager = self._make_provider_with_strategy("per-directory")
mock_manager.migrate_memory_files.assert_called_once()
class TestChunkMessage:
def test_short_message_single_chunk(self):
result = HonchoMemoryProvider._chunk_message("hello world", 100)
@@ -420,6 +673,60 @@ class TestChunkMessage:
assert len(chunk) <= 25000
# ---------------------------------------------------------------------------
# Context token budget enforcement
# ---------------------------------------------------------------------------
class TestTruncateToBudget:
def test_truncates_oversized_context(self):
"""Text exceeding context_tokens budget is truncated at a word boundary."""
from plugins.memory.honcho.client import HonchoClientConfig
provider = HonchoMemoryProvider()
provider._config = HonchoClientConfig(context_tokens=10)
long_text = "word " * 200 # ~1000 chars, well over 10*4=40 char budget
result = provider._truncate_to_budget(long_text)
assert len(result) <= 50 # budget_chars + ellipsis + word boundary slack
assert result.endswith("")
def test_no_truncation_within_budget(self):
"""Text within budget passes through unchanged."""
from plugins.memory.honcho.client import HonchoClientConfig
provider = HonchoMemoryProvider()
provider._config = HonchoClientConfig(context_tokens=1000)
short_text = "Name: Robert, Location: Melbourne"
assert provider._truncate_to_budget(short_text) == short_text
def test_no_truncation_when_context_tokens_none(self):
"""When context_tokens is None (explicit opt-out), no truncation."""
from plugins.memory.honcho.client import HonchoClientConfig
provider = HonchoMemoryProvider()
provider._config = HonchoClientConfig(context_tokens=None)
long_text = "word " * 500
assert provider._truncate_to_budget(long_text) == long_text
def test_context_tokens_cap_bounds_prefetch(self):
"""With an explicit token budget, oversized prefetch is bounded."""
from plugins.memory.honcho.client import HonchoClientConfig
provider = HonchoMemoryProvider()
provider._config = HonchoClientConfig(context_tokens=1200)
# Simulate a massive representation (10k chars)
huge_text = "x" * 10000
result = provider._truncate_to_budget(huge_text)
# 1200 tokens * 4 chars = 4800 chars + " …"
assert len(result) <= 4805
# ---------------------------------------------------------------------------
# Dialectic input guard
# ---------------------------------------------------------------------------
@@ -452,3 +759,387 @@ class TestDialecticInputGuard:
# The query passed to chat() should be truncated
actual_query = mock_peer.chat.call_args[0][0]
assert len(actual_query) <= 100
# ---------------------------------------------------------------------------
class TestDialecticCadenceDefaults:
"""Regression tests for dialectic_cadence default value."""
@staticmethod
def _make_provider(cfg_extra=None):
"""Create a HonchoMemoryProvider with mocked dependencies."""
from unittest.mock import patch, MagicMock
from plugins.memory.honcho.client import HonchoClientConfig
defaults = dict(api_key="test-key", enabled=True, recall_mode="hybrid")
if cfg_extra:
defaults.update(cfg_extra)
cfg = HonchoClientConfig(**defaults)
provider = HonchoMemoryProvider()
mock_manager = MagicMock()
mock_session = MagicMock()
mock_session.messages = []
mock_manager.get_or_create.return_value = mock_session
with patch("plugins.memory.honcho.client.HonchoClientConfig.from_global_config", return_value=cfg), \
patch("plugins.memory.honcho.client.get_honcho_client", return_value=MagicMock()), \
patch("plugins.memory.honcho.session.HonchoSessionManager", return_value=mock_manager), \
patch("hermes_constants.get_hermes_home", return_value=MagicMock()):
provider.initialize(session_id="test-session-001")
return provider
def test_default_is_3(self):
"""Default dialectic_cadence should be 3 to avoid per-turn LLM calls."""
provider = self._make_provider()
assert provider._dialectic_cadence == 3
def test_config_override(self):
"""dialecticCadence from config overrides the default."""
provider = self._make_provider(cfg_extra={"raw": {"dialecticCadence": 5}})
assert provider._dialectic_cadence == 5
class TestBaseContextSummary:
"""Base context injection should include session summary when available."""
def test_format_includes_summary(self):
"""Session summary should appear first in the formatted context."""
provider = HonchoMemoryProvider()
ctx = {
"summary": "Testing Honcho tools and dialectic depth.",
"representation": "Eri is a developer.",
"card": "Name: Eri Barrett",
}
formatted = provider._format_first_turn_context(ctx)
assert "## Session Summary" in formatted
assert formatted.index("Session Summary") < formatted.index("User Representation")
def test_format_without_summary(self):
"""No summary key means no summary section."""
provider = HonchoMemoryProvider()
ctx = {"representation": "Eri is a developer.", "card": "Name: Eri"}
formatted = provider._format_first_turn_context(ctx)
assert "Session Summary" not in formatted
assert "User Representation" in formatted
def test_format_empty_summary_skipped(self):
"""Empty summary string should not produce a section."""
provider = HonchoMemoryProvider()
ctx = {"summary": "", "representation": "rep", "card": "card"}
formatted = provider._format_first_turn_context(ctx)
assert "Session Summary" not in formatted
class TestDialecticDepth:
"""Tests for the dialecticDepth multi-pass system."""
@staticmethod
def _make_provider(cfg_extra=None):
from unittest.mock import patch, MagicMock
from plugins.memory.honcho.client import HonchoClientConfig
defaults = dict(api_key="test-key", enabled=True, recall_mode="hybrid")
if cfg_extra:
defaults.update(cfg_extra)
cfg = HonchoClientConfig(**defaults)
provider = HonchoMemoryProvider()
mock_manager = MagicMock()
mock_session = MagicMock()
mock_session.messages = []
mock_manager.get_or_create.return_value = mock_session
with patch("plugins.memory.honcho.client.HonchoClientConfig.from_global_config", return_value=cfg), \
patch("plugins.memory.honcho.client.get_honcho_client", return_value=MagicMock()), \
patch("plugins.memory.honcho.session.HonchoSessionManager", return_value=mock_manager), \
patch("hermes_constants.get_hermes_home", return_value=MagicMock()):
provider.initialize(session_id="test-session-001")
return provider
def test_default_depth_is_1(self):
"""Default dialecticDepth should be 1 — single .chat() call."""
provider = self._make_provider()
assert provider._dialectic_depth == 1
def test_depth_from_config(self):
"""dialecticDepth from config sets the depth."""
provider = self._make_provider(cfg_extra={"dialectic_depth": 2})
assert provider._dialectic_depth == 2
def test_depth_clamped_to_3(self):
"""dialecticDepth > 3 gets clamped to 3."""
provider = self._make_provider(cfg_extra={"dialectic_depth": 7})
assert provider._dialectic_depth == 3
def test_depth_clamped_to_1(self):
"""dialecticDepth < 1 gets clamped to 1."""
provider = self._make_provider(cfg_extra={"dialectic_depth": 0})
assert provider._dialectic_depth == 1
def test_depth_levels_from_config(self):
"""dialecticDepthLevels array is read from config."""
provider = self._make_provider(cfg_extra={
"dialectic_depth": 2,
"dialectic_depth_levels": ["minimal", "high"],
})
assert provider._dialectic_depth_levels == ["minimal", "high"]
def test_depth_levels_none_by_default(self):
"""When dialecticDepthLevels is not configured, it's None."""
provider = self._make_provider()
assert provider._dialectic_depth_levels is None
def test_resolve_pass_level_uses_depth_levels(self):
"""Per-pass levels from dialecticDepthLevels override proportional."""
provider = self._make_provider(cfg_extra={
"dialectic_depth": 2,
"dialectic_depth_levels": ["minimal", "high"],
})
assert provider._resolve_pass_level(0) == "minimal"
assert provider._resolve_pass_level(1) == "high"
def test_resolve_pass_level_proportional_depth_1(self):
"""Depth 1 pass 0 uses the base reasoning level."""
provider = self._make_provider(cfg_extra={
"dialectic_depth": 1,
"dialectic_reasoning_level": "medium",
})
assert provider._resolve_pass_level(0) == "medium"
def test_resolve_pass_level_proportional_depth_2(self):
"""Depth 2: pass 0 is minimal, pass 1 is base level."""
provider = self._make_provider(cfg_extra={
"dialectic_depth": 2,
"dialectic_reasoning_level": "high",
})
assert provider._resolve_pass_level(0) == "minimal"
assert provider._resolve_pass_level(1) == "high"
def test_cold_start_prompt(self):
"""Cold start (no base context) uses general user query."""
provider = self._make_provider()
prompt = provider._build_dialectic_prompt(0, [], is_cold=True)
assert "preferences" in prompt.lower()
assert "session" not in prompt.lower()
def test_warm_session_prompt(self):
"""Warm session (has context) uses session-scoped query."""
provider = self._make_provider()
prompt = provider._build_dialectic_prompt(0, [], is_cold=False)
assert "session" in prompt.lower()
assert "current conversation" in prompt.lower()
def test_signal_sufficient_short_response(self):
"""Short responses are not sufficient signal."""
assert not HonchoMemoryProvider._signal_sufficient("ok")
assert not HonchoMemoryProvider._signal_sufficient("")
assert not HonchoMemoryProvider._signal_sufficient(None)
def test_signal_sufficient_structured_response(self):
"""Structured responses with bullets/headers are sufficient."""
result = "## Current State\n- Working on Honcho PR\n- Testing dialectic depth\n" + "x" * 50
assert HonchoMemoryProvider._signal_sufficient(result)
def test_signal_sufficient_long_unstructured(self):
"""Long responses are sufficient even without structure."""
assert HonchoMemoryProvider._signal_sufficient("a" * 301)
def test_run_dialectic_depth_single_pass(self):
"""Depth 1 makes exactly one .chat() call."""
from unittest.mock import MagicMock
provider = self._make_provider(cfg_extra={"dialectic_depth": 1})
provider._manager = MagicMock()
provider._manager.dialectic_query.return_value = "user prefers zero-fluff"
provider._session_key = "test"
provider._base_context_cache = None # cold start
result = provider._run_dialectic_depth("hello")
assert result == "user prefers zero-fluff"
assert provider._manager.dialectic_query.call_count == 1
def test_run_dialectic_depth_two_passes(self):
"""Depth 2 makes two .chat() calls when pass 1 signal is weak."""
from unittest.mock import MagicMock
provider = self._make_provider(cfg_extra={"dialectic_depth": 2})
provider._manager = MagicMock()
provider._manager.dialectic_query.side_effect = [
"thin response", # pass 0: weak signal
"## Synthesis\n- Grounded in evidence\n- Current PR work\n" + "x" * 100, # pass 1: strong
]
provider._session_key = "test"
provider._base_context_cache = "existing context"
result = provider._run_dialectic_depth("test query")
assert provider._manager.dialectic_query.call_count == 2
assert "Synthesis" in result
def test_first_turn_runs_dialectic_synchronously(self):
"""First turn should fire the dialectic synchronously (cold start)."""
from unittest.mock import MagicMock, patch
provider = self._make_provider(cfg_extra={"dialectic_depth": 1})
provider._manager = MagicMock()
provider._manager.dialectic_query.return_value = "cold start synthesis"
provider._manager.get_prefetch_context.return_value = None
provider._manager.pop_context_result.return_value = None
provider._session_key = "test"
provider._base_context_cache = "" # cold start
provider._last_dialectic_turn = -999 # never fired
result = provider.prefetch("hello world")
assert "cold start synthesis" in result
assert provider._manager.dialectic_query.call_count == 1
# After first-turn sync, _last_dialectic_turn should be updated
assert provider._last_dialectic_turn != -999
def test_first_turn_dialectic_does_not_double_fire(self):
"""After first-turn sync dialectic, queue_prefetch should skip (cadence)."""
from unittest.mock import MagicMock
provider = self._make_provider(cfg_extra={"dialectic_depth": 1})
provider._manager = MagicMock()
provider._manager.dialectic_query.return_value = "cold start synthesis"
provider._manager.get_prefetch_context.return_value = None
provider._manager.pop_context_result.return_value = None
provider._session_key = "test"
provider._base_context_cache = ""
provider._last_dialectic_turn = -999
provider._turn_count = 0
# First turn fires sync dialectic
provider.prefetch("hello")
assert provider._manager.dialectic_query.call_count == 1
# Now queue_prefetch on same turn should skip (cadence: 0 - 0 < 3)
provider._manager.dialectic_query.reset_mock()
provider.queue_prefetch("hello")
assert provider._manager.dialectic_query.call_count == 0
def test_run_dialectic_depth_bails_early_on_strong_signal(self):
"""Depth 2 skips pass 1 when pass 0 returns strong signal."""
from unittest.mock import MagicMock
provider = self._make_provider(cfg_extra={"dialectic_depth": 2})
provider._manager = MagicMock()
provider._manager.dialectic_query.return_value = (
"## Full Assessment\n- Strong structured response\n- With evidence\n" + "x" * 200
)
provider._session_key = "test"
provider._base_context_cache = "existing context"
result = provider._run_dialectic_depth("test query")
# Only 1 call because pass 0 had sufficient signal
assert provider._manager.dialectic_query.call_count == 1
# ---------------------------------------------------------------------------
# set_peer_card None guard
# ---------------------------------------------------------------------------
class TestSetPeerCardNoneGuard:
"""set_peer_card must return None (not raise) when peer ID cannot be resolved."""
def _make_manager(self):
from plugins.memory.honcho.client import HonchoClientConfig
from plugins.memory.honcho.session import HonchoSessionManager
cfg = HonchoClientConfig(api_key="test-key", enabled=True)
mgr = HonchoSessionManager.__new__(HonchoSessionManager)
mgr._cache = {}
mgr._sessions_cache = {}
mgr._config = cfg
return mgr
def test_returns_none_when_peer_resolves_to_none(self):
"""set_peer_card returns None when _resolve_peer_id returns None."""
from unittest.mock import patch
mgr = self._make_manager()
session = HonchoSession(
key="test",
honcho_session_id="sid",
user_peer_id="user-peer",
assistant_peer_id="ai-peer",
)
mgr._cache["test"] = session
with patch.object(mgr, "_resolve_peer_id", return_value=None):
result = mgr.set_peer_card("test", ["fact 1", "fact 2"], peer="ghost")
assert result is None
def test_returns_none_when_session_missing(self):
"""set_peer_card returns None when session key is not in cache."""
mgr = self._make_manager()
result = mgr.set_peer_card("nonexistent", ["fact"], peer="user")
assert result is None
# ---------------------------------------------------------------------------
# get_session_context cache-miss fallback respects peer param
# ---------------------------------------------------------------------------
class TestGetSessionContextFallback:
"""get_session_context fallback must honour the peer param when honcho_session is absent."""
def _make_manager_with_session(self, user_peer_id="user-peer", assistant_peer_id="ai-peer"):
from plugins.memory.honcho.client import HonchoClientConfig
from plugins.memory.honcho.session import HonchoSessionManager
cfg = HonchoClientConfig(api_key="test-key", enabled=True)
mgr = HonchoSessionManager.__new__(HonchoSessionManager)
mgr._cache = {}
mgr._sessions_cache = {}
mgr._config = cfg
mgr._dialectic_dynamic = True
mgr._dialectic_reasoning_level = "low"
mgr._dialectic_max_input_chars = 10000
mgr._ai_observe_others = True
session = HonchoSession(
key="test",
honcho_session_id="sid-missing-from-sessions-cache",
user_peer_id=user_peer_id,
assistant_peer_id=assistant_peer_id,
)
mgr._cache["test"] = session
# Deliberately NOT adding to _sessions_cache to trigger fallback path
return mgr
def test_fallback_uses_user_peer_for_user(self):
"""On cache miss, peer='user' fetches user peer context."""
mgr = self._make_manager_with_session()
fetch_calls = []
def _fake_fetch(peer_id, search_query=None, *, target=None):
fetch_calls.append((peer_id, target))
return {"representation": "user rep", "card": []}
mgr._fetch_peer_context = _fake_fetch
mgr.get_session_context("test", peer="user")
assert len(fetch_calls) == 1
peer_id, target = fetch_calls[0]
assert peer_id == "user-peer"
assert target == "user-peer"
def test_fallback_uses_ai_peer_for_ai(self):
"""On cache miss, peer='ai' fetches assistant peer context, not user."""
mgr = self._make_manager_with_session()
fetch_calls = []
def _fake_fetch(peer_id, search_query=None, *, target=None):
fetch_calls.append((peer_id, target))
return {"representation": "ai rep", "card": []}
mgr._fetch_peer_context = _fake_fetch
mgr.get_session_context("test", peer="ai")
assert len(fetch_calls) == 1
peer_id, target = fetch_calls[0]
assert peer_id == "ai-peer", f"expected ai-peer, got {peer_id}"
assert target == "ai-peer"

View File

@@ -0,0 +1,139 @@
"""Tests for interrupt handling in concurrent tool execution."""
import concurrent.futures
import threading
import time
from unittest.mock import MagicMock, patch
import pytest
@pytest.fixture(autouse=True)
def _isolate_hermes(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
(tmp_path / ".hermes").mkdir(exist_ok=True)
def _make_agent(monkeypatch):
"""Create a minimal AIAgent-like object with just the methods under test."""
monkeypatch.setenv("OPENROUTER_API_KEY", "")
monkeypatch.setenv("HERMES_INFERENCE_PROVIDER", "")
# Avoid full AIAgent init — just import the class and build a stub
import run_agent as _ra
class _Stub:
_interrupt_requested = False
log_prefix = ""
quiet_mode = True
verbose_logging = False
log_prefix_chars = 200
_checkpoint_mgr = MagicMock(enabled=False)
_subdirectory_hints = MagicMock()
tool_progress_callback = None
tool_start_callback = None
tool_complete_callback = None
_todo_store = MagicMock()
_session_db = None
valid_tool_names = set()
_turns_since_memory = 0
_iters_since_skill = 0
_current_tool = None
_last_activity = 0
_print_fn = print
def _touch_activity(self, desc):
self._last_activity = time.time()
def _vprint(self, msg, force=False):
pass
def _safe_print(self, msg):
pass
def _should_emit_quiet_tool_messages(self):
return False
def _should_start_quiet_spinner(self):
return False
def _has_stream_consumers(self):
return False
stub = _Stub()
# Bind the real methods
stub._execute_tool_calls_concurrent = _ra.AIAgent._execute_tool_calls_concurrent.__get__(stub)
stub._invoke_tool = MagicMock(side_effect=lambda *a, **kw: '{"ok": true}')
return stub
class _FakeToolCall:
def __init__(self, name, args="{}", call_id="tc_1"):
self.function = MagicMock(name=name, arguments=args)
self.function.name = name
self.id = call_id
class _FakeAssistantMsg:
def __init__(self, tool_calls):
self.tool_calls = tool_calls
def test_concurrent_interrupt_cancels_pending(monkeypatch):
"""When _interrupt_requested is set during concurrent execution,
the wait loop should exit early and cancelled tools get interrupt messages."""
agent = _make_agent(monkeypatch)
# Create a tool that blocks until interrupted
barrier = threading.Event()
original_invoke = agent._invoke_tool
def slow_tool(name, args, task_id, call_id=None):
if name == "slow_one":
# Block until the test sets the interrupt
barrier.wait(timeout=10)
return '{"slow": true}'
return '{"fast": true}'
agent._invoke_tool = MagicMock(side_effect=slow_tool)
tc1 = _FakeToolCall("fast_one", call_id="tc_fast")
tc2 = _FakeToolCall("slow_one", call_id="tc_slow")
msg = _FakeAssistantMsg([tc1, tc2])
messages = []
def _set_interrupt_after_delay():
time.sleep(0.3)
agent._interrupt_requested = True
barrier.set() # unblock the slow tool
t = threading.Thread(target=_set_interrupt_after_delay)
t.start()
agent._execute_tool_calls_concurrent(msg, messages, "test_task")
t.join()
# Both tools should have results in messages
assert len(messages) == 2
# The interrupt was detected
assert agent._interrupt_requested is True
def test_concurrent_preflight_interrupt_skips_all(monkeypatch):
"""When _interrupt_requested is already set before concurrent execution,
all tools are skipped with cancellation messages."""
agent = _make_agent(monkeypatch)
agent._interrupt_requested = True
tc1 = _FakeToolCall("tool_a", call_id="tc_a")
tc2 = _FakeToolCall("tool_b", call_id="tc_b")
msg = _FakeAssistantMsg([tc1, tc2])
messages = []
agent._execute_tool_calls_concurrent(msg, messages, "test_task")
assert len(messages) == 2
assert "skipped due to user interrupt" in messages[0]["content"]
assert "skipped due to user interrupt" in messages[1]["content"]
# _invoke_tool should never have been called
agent._invoke_tool.assert_not_called()

View File

@@ -9,6 +9,8 @@ def _build_agent(model_cfg, custom_providers=None, model="anthropic/claude-opus-
if custom_providers is not None:
cfg["custom_providers"] = custom_providers
base_url = model_cfg.get("base_url", "")
with (
patch("hermes_cli.config.load_config", return_value=cfg),
patch("agent.model_metadata.get_model_context_length", return_value=128_000),
@@ -21,6 +23,7 @@ def _build_agent(model_cfg, custom_providers=None, model="anthropic/claude-opus-
agent = AIAgent(
model=model,
api_key="test-key-1234567890",
base_url=base_url,
quiet_mode=True,
skip_context_files=True,
skip_memory=True,

View File

@@ -805,7 +805,10 @@ class TestCodexReasoningPreflight:
reasoning_items = [i for i in normalized if i.get("type") == "reasoning"]
assert len(reasoning_items) == 1
assert reasoning_items[0]["encrypted_content"] == "abc123encrypted"
assert reasoning_items[0]["id"] == "r_001"
# Note: "id" is intentionally excluded from normalized output —
# with store=False the API returns 404 on server-side id resolution.
# The id is only used for local deduplication via seen_ids.
assert "id" not in reasoning_items[0]
assert reasoning_items[0]["summary"] == [{"type": "summary_text", "text": "Thinking about it"}]
def test_reasoning_item_without_id(self, monkeypatch):

View File

@@ -928,6 +928,7 @@ class TestBuildApiKwargs:
kwargs = agent._build_api_kwargs(messages)
assert kwargs["max_tokens"] == 4096
def test_qwen_portal_formats_messages_and_metadata(self, agent):
agent.base_url = "https://portal.qwen.ai/v1"
agent._base_url_lower = agent.base_url.lower()
@@ -984,6 +985,46 @@ class TestBuildApiKwargs:
kwargs = agent._build_api_kwargs(messages)
assert kwargs["max_tokens"] == 65536
def test_ollama_think_false_on_effort_none(self, agent):
"""Custom (Ollama) provider with effort=none should inject think=false."""
agent.provider = "custom"
agent.base_url = "http://localhost:11434/v1"
agent._base_url_lower = agent.base_url.lower()
agent.reasoning_config = {"effort": "none"}
messages = [{"role": "user", "content": "hi"}]
kwargs = agent._build_api_kwargs(messages)
assert kwargs.get("extra_body", {}).get("think") is False
def test_ollama_think_false_on_enabled_false(self, agent):
"""Custom (Ollama) provider with enabled=false should inject think=false."""
agent.provider = "custom"
agent.base_url = "http://localhost:11434/v1"
agent._base_url_lower = agent.base_url.lower()
agent.reasoning_config = {"enabled": False}
messages = [{"role": "user", "content": "hi"}]
kwargs = agent._build_api_kwargs(messages)
assert kwargs.get("extra_body", {}).get("think") is False
def test_ollama_no_think_param_when_reasoning_enabled(self, agent):
"""Custom provider with reasoning enabled should NOT inject think=false."""
agent.provider = "custom"
agent.base_url = "http://localhost:11434/v1"
agent._base_url_lower = agent.base_url.lower()
agent.reasoning_config = {"enabled": True, "effort": "medium"}
messages = [{"role": "user", "content": "hi"}]
kwargs = agent._build_api_kwargs(messages)
assert kwargs.get("extra_body", {}).get("think") is None
def test_non_custom_provider_unaffected(self, agent):
"""OpenRouter provider with effort=none should NOT inject think=false."""
agent.provider = "openrouter"
agent.model = "qwen/qwen3.5-plus-02-15"
agent.reasoning_config = {"effort": "none"}
messages = [{"role": "user", "content": "hi"}]
kwargs = agent._build_api_kwargs(messages)
assert kwargs.get("extra_body", {}).get("think") is None
class TestBuildAssistantMessage:
def test_basic_message(self, agent):
@@ -2202,6 +2243,114 @@ class TestRunConversation:
assert second_call_messages[-1]["role"] == "user"
assert "truncated by the output length limit" in second_call_messages[-1]["content"]
def test_ollama_glm_stop_after_tools_without_terminal_boundary_requests_continuation(self, agent):
"""Ollama-hosted GLM responses can misreport truncated output as stop."""
self._setup_agent(agent)
agent.base_url = "http://localhost:11434/v1"
agent._base_url_lower = agent.base_url.lower()
agent.model = "glm-5.1:cloud"
tool_turn = _mock_response(
content="",
finish_reason="tool_calls",
tool_calls=[_mock_tool_call(name="web_search", arguments="{}", call_id="c1")],
)
misreported_stop = _mock_response(
content="Based on the search results, the best next",
finish_reason="stop",
)
continued = _mock_response(
content=" step is to update the config.",
finish_reason="stop",
)
agent.client.chat.completions.create.side_effect = [
tool_turn,
misreported_stop,
continued,
]
with (
patch("run_agent.handle_function_call", return_value="search result"),
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello")
assert result["completed"] is True
assert result["api_calls"] == 3
assert (
result["final_response"]
== "Based on the search results, the best next step is to update the config."
)
third_call_messages = agent.client.chat.completions.create.call_args_list[2].kwargs["messages"]
assert third_call_messages[-1]["role"] == "user"
assert "truncated by the output length limit" in third_call_messages[-1]["content"]
def test_ollama_glm_stop_with_terminal_boundary_does_not_continue(self, agent):
"""Complete Ollama/GLM responses should not be reclassified as truncated."""
self._setup_agent(agent)
agent.base_url = "http://localhost:11434/v1"
agent._base_url_lower = agent.base_url.lower()
agent.model = "glm-5.1:cloud"
tool_turn = _mock_response(
content="",
finish_reason="tool_calls",
tool_calls=[_mock_tool_call(name="web_search", arguments="{}", call_id="c1")],
)
complete_stop = _mock_response(
content="Based on the search results, the best next step is to update the config.",
finish_reason="stop",
)
agent.client.chat.completions.create.side_effect = [tool_turn, complete_stop]
with (
patch("run_agent.handle_function_call", return_value="search result"),
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello")
assert result["completed"] is True
assert result["api_calls"] == 2
assert (
result["final_response"]
== "Based on the search results, the best next step is to update the config."
)
def test_non_ollama_stop_without_terminal_boundary_does_not_continue(self, agent):
"""The stop->length workaround should stay scoped to Ollama/GLM backends."""
self._setup_agent(agent)
agent.base_url = "https://api.openai.com/v1"
agent._base_url_lower = agent.base_url.lower()
agent.model = "gpt-4o-mini"
tool_turn = _mock_response(
content="",
finish_reason="tool_calls",
tool_calls=[_mock_tool_call(name="web_search", arguments="{}", call_id="c1")],
)
normal_stop = _mock_response(
content="Based on the search results, the best next",
finish_reason="stop",
)
agent.client.chat.completions.create.side_effect = [tool_turn, normal_stop]
with (
patch("run_agent.handle_function_call", return_value="search result"),
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello")
assert result["completed"] is True
assert result["api_calls"] == 2
assert result["final_response"] == "Based on the search results, the best next"
def test_length_thinking_exhausted_skips_continuation(self, agent):
"""When finish_reason='length' but content is only thinking, skip retries."""
self._setup_agent(agent)
@@ -3998,3 +4147,63 @@ class TestDeadRetryCode:
f"Expected 2 occurrences of 'if retry_count >= max_retries:' "
f"but found {occurrences}"
)
class TestMemoryContextSanitization:
"""run_conversation() must strip leaked <memory-context> blocks from user input."""
def test_memory_context_stripped_from_user_message(self):
"""Verify that <memory-context> blocks are removed before the message
enters the conversation loop — prevents stale Honcho injection from
leaking into user text."""
import inspect
src = inspect.getsource(AIAgent.run_conversation)
# The sanitize_context call must appear in run_conversation's preamble
assert "sanitize_context(user_message)" in src
assert "sanitize_context(persist_user_message)" in src
def test_sanitize_context_strips_full_block(self):
"""End-to-end: a user message with an embedded memory-context block
is cleaned to just the actual user text."""
from agent.memory_manager import sanitize_context
user_text = "how is the honcho working"
injected = (
user_text + "\n\n"
"<memory-context>\n"
"[System note: The following is recalled memory context, "
"NOT new user input. Treat as informational background data.]\n\n"
"## User Representation\n"
"[2026-01-13 02:13:00] stale observation about AstroMap\n"
"</memory-context>"
)
result = sanitize_context(injected)
assert "memory-context" not in result.lower()
assert "stale observation" not in result
assert "how is the honcho working" in result
class TestMemoryProviderTurnStart:
"""run_conversation() must call memory_manager.on_turn_start() before prefetch_all().
Without this call, providers like Honcho never update _turn_count, so cadence
checks (contextCadence, dialecticCadence) are always satisfied — every turn
fires both context refresh and dialectic, ignoring the configured cadence.
"""
def test_on_turn_start_called_before_prefetch(self):
"""Source-level check: on_turn_start appears before prefetch_all in run_conversation."""
import inspect
src = inspect.getsource(AIAgent.run_conversation)
# Find the actual method calls, not comments
idx_turn_start = src.index(".on_turn_start(")
idx_prefetch = src.index(".prefetch_all(")
assert idx_turn_start < idx_prefetch, (
"on_turn_start() must be called before prefetch_all() in run_conversation "
"so that memory providers have the correct turn count for cadence checks"
)
def test_on_turn_start_uses_user_turn_count(self):
"""Source-level check: on_turn_start receives self._user_turn_count."""
import inspect
src = inspect.getsource(AIAgent.run_conversation)
assert "on_turn_start(self._user_turn_count" in src

View File

@@ -160,7 +160,9 @@ class TestExchangeAuthCode:
assert flow.state == "saved-state"
assert flow.code_verifier == "saved-verifier"
assert flow.fetch_token_calls == [{"code": "4/test-auth-code"}]
assert json.loads(setup_module.TOKEN_PATH.read_text())["token"] == "access-token"
saved = json.loads(setup_module.TOKEN_PATH.read_text())
assert saved["token"] == "access-token"
assert saved["type"] == "authorized_user"
assert not setup_module.PENDING_AUTH_PATH.exists()
def test_extracts_code_from_redirect_url_and_checks_state(self, setup_module):

View File

@@ -46,6 +46,12 @@ def api_module(monkeypatch, tmp_path):
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
# Ensure the gws CLI code path is taken even when the binary isn't
# installed (CI). Without this, calendar_list() falls through to the
# Python SDK path which imports ``googleapiclient`` — not in deps.
module._gws_binary = lambda: "/usr/bin/gws"
# Bypass authentication check — no real token file in CI.
module._ensure_authenticated = lambda: None
return module
@@ -94,6 +100,7 @@ def test_bridge_refreshes_expired_token(bridge_module, tmp_path):
# Verify persisted
saved = json.loads(token_path.read_text())
assert saved["token"] == "ya29.refreshed"
assert saved["type"] == "authorized_user"
def test_bridge_exits_on_missing_token(bridge_module):
@@ -124,35 +131,41 @@ def test_bridge_main_injects_token_env(bridge_module, tmp_path):
assert captured["cmd"] == ["gws", "gmail", "+triage"]
def test_api_calendar_list_uses_agenda_by_default(api_module):
"""calendar list without dates uses +agenda helper."""
def test_api_calendar_list_uses_events_list(api_module):
"""calendar_list calls _run_gws with events list + params."""
captured = {}
def capture_run(cmd, **kwargs):
captured["cmd"] = cmd
return MagicMock(returncode=0)
return MagicMock(returncode=0, stdout="{}", stderr="")
args = api_module.argparse.Namespace(
start="", end="", max=25, calendar="primary", func=api_module.calendar_list,
)
with patch.object(subprocess, "run", side_effect=capture_run):
with pytest.raises(SystemExit):
api_module.calendar_list(args)
with patch.object(api_module.subprocess, "run", side_effect=capture_run):
api_module.calendar_list(args)
gws_args = captured["cmd"][2:] # skip python + bridge path
assert "calendar" in gws_args
assert "+agenda" in gws_args
assert "--days" in gws_args
cmd = captured["cmd"]
# _gws_binary() returns "/usr/bin/gws", so cmd[0] is that binary
assert cmd[0] == "/usr/bin/gws"
assert "calendar" in cmd
assert "events" in cmd
assert "list" in cmd
assert "--params" in cmd
params = json.loads(cmd[cmd.index("--params") + 1])
assert "timeMin" in params
assert "timeMax" in params
assert params["calendarId"] == "primary"
def test_api_calendar_list_respects_date_range(api_module):
"""calendar list with --start/--end uses raw events list API."""
"""calendar list with --start/--end passes correct time bounds."""
captured = {}
def capture_run(cmd, **kwargs):
captured["cmd"] = cmd
return MagicMock(returncode=0)
return MagicMock(returncode=0, stdout="{}", stderr="")
args = api_module.argparse.Namespace(
start="2026-04-01T00:00:00Z",
@@ -162,14 +175,62 @@ def test_api_calendar_list_respects_date_range(api_module):
func=api_module.calendar_list,
)
with patch.object(subprocess, "run", side_effect=capture_run):
with pytest.raises(SystemExit):
api_module.calendar_list(args)
with patch.object(api_module.subprocess, "run", side_effect=capture_run):
api_module.calendar_list(args)
gws_args = captured["cmd"][2:]
assert "events" in gws_args
assert "list" in gws_args
params_idx = gws_args.index("--params")
params = json.loads(gws_args[params_idx + 1])
cmd = captured["cmd"]
params_idx = cmd.index("--params")
params = json.loads(cmd[params_idx + 1])
assert params["timeMin"] == "2026-04-01T00:00:00Z"
assert params["timeMax"] == "2026-04-07T23:59:59Z"
def test_api_get_credentials_refresh_persists_authorized_user_type(api_module, monkeypatch):
token_path = api_module.TOKEN_PATH
_write_token(token_path, token="ya29.old")
class FakeCredentials:
def __init__(self):
self.expired = True
self.refresh_token = "1//refresh"
self.valid = True
def refresh(self, request):
self.expired = False
def to_json(self):
return json.dumps({
"token": "ya29.refreshed",
"refresh_token": "1//refresh",
"client_id": "123.apps.googleusercontent.com",
"client_secret": "secret",
"token_uri": "https://oauth2.googleapis.com/token",
})
class FakeCredentialsModule:
@staticmethod
def from_authorized_user_file(filename, scopes):
assert filename == str(token_path)
assert scopes == api_module.SCOPES
return FakeCredentials()
google_module = types.ModuleType("google")
oauth2_module = types.ModuleType("google.oauth2")
credentials_module = types.ModuleType("google.oauth2.credentials")
credentials_module.Credentials = FakeCredentialsModule
transport_module = types.ModuleType("google.auth.transport")
requests_module = types.ModuleType("google.auth.transport.requests")
requests_module.Request = lambda: object()
monkeypatch.setitem(sys.modules, "google", google_module)
monkeypatch.setitem(sys.modules, "google.oauth2", oauth2_module)
monkeypatch.setitem(sys.modules, "google.oauth2.credentials", credentials_module)
monkeypatch.setitem(sys.modules, "google.auth.transport", transport_module)
monkeypatch.setitem(sys.modules, "google.auth.transport.requests", requests_module)
creds = api_module.get_credentials()
saved = json.loads(token_path.read_text())
assert isinstance(creds, FakeCredentials)
assert saved["token"] == "ya29.refreshed"
assert saved["type"] == "authorized_user"

View File

@@ -1,8 +1,8 @@
"""Persistence tests for the Camofox browser backend.
Tests that managed persistence uses stable identity while default mode
uses random identity. The actual browser profile persistence is handled
by the Camofox server (when CAMOFOX_PROFILE_DIR is set).
uses random identity. Camofox automatically maps each userId to a
dedicated persistent Firefox profile on the server side.
"""
import json

View File

@@ -0,0 +1,166 @@
"""Tests for cloud browser provider runtime fallback to local Chromium.
Covers the fallback logic in _get_session_info() when a cloud provider
is configured but fails at runtime (issue #10883).
"""
import logging
from unittest.mock import Mock, patch
import pytest
import tools.browser_tool as browser_tool
def _reset_session_state(monkeypatch):
"""Clear caches so each test starts fresh."""
monkeypatch.setattr(browser_tool, "_active_sessions", {})
monkeypatch.setattr(browser_tool, "_cached_cloud_provider", None)
monkeypatch.setattr(browser_tool, "_cloud_provider_resolved", False)
monkeypatch.setattr(browser_tool, "_start_browser_cleanup_thread", lambda: None)
monkeypatch.setattr(browser_tool, "_update_session_activity", lambda t: None)
class TestCloudProviderRuntimeFallback:
"""Tests for _get_session_info cloud → local fallback."""
def test_cloud_failure_falls_back_to_local(self, monkeypatch):
"""When cloud provider.create_session raises, fall back to local."""
_reset_session_state(monkeypatch)
provider = Mock()
provider.create_session.side_effect = RuntimeError("401 Unauthorized")
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
session = browser_tool._get_session_info("task-1")
assert session["fallback_from_cloud"] is True
assert "401 Unauthorized" in session["fallback_reason"]
assert session["fallback_provider"] == "Mock"
assert session["features"]["local"] is True
assert session["cdp_url"] is None
def test_cloud_success_no_fallback(self, monkeypatch):
"""When cloud succeeds, no fallback markers are present."""
_reset_session_state(monkeypatch)
provider = Mock()
provider.create_session.return_value = {
"session_name": "cloud-sess",
"bb_session_id": "bb_123",
"cdp_url": None,
"features": {"browser_use": True},
}
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
session = browser_tool._get_session_info("task-2")
assert session["session_name"] == "cloud-sess"
assert "fallback_from_cloud" not in session
assert "fallback_reason" not in session
def test_cloud_and_local_both_fail(self, monkeypatch):
"""When both cloud and local fail, raise RuntimeError with both contexts."""
_reset_session_state(monkeypatch)
provider = Mock()
provider.create_session.side_effect = RuntimeError("cloud boom")
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
monkeypatch.setattr(
browser_tool, "_create_local_session",
Mock(side_effect=OSError("no chromium")),
)
with pytest.raises(RuntimeError, match="cloud boom.*local.*no chromium"):
browser_tool._get_session_info("task-3")
def test_no_provider_uses_local_directly(self, monkeypatch):
"""When no cloud provider is configured, local mode is used with no fallback markers."""
_reset_session_state(monkeypatch)
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: None)
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
session = browser_tool._get_session_info("task-4")
assert session["features"]["local"] is True
assert "fallback_from_cloud" not in session
def test_cdp_override_bypasses_provider(self, monkeypatch):
"""CDP override takes priority — cloud provider is never consulted."""
_reset_session_state(monkeypatch)
provider = Mock()
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: "ws://host:9222/devtools/browser/abc")
session = browser_tool._get_session_info("task-5")
provider.create_session.assert_not_called()
assert session["cdp_url"] == "ws://host:9222/devtools/browser/abc"
def test_fallback_logs_warning_with_provider_name(self, monkeypatch, caplog):
"""Fallback emits a warning log with the provider class name and error."""
_reset_session_state(monkeypatch)
BrowserUseProviderFake = type("BrowserUseProvider", (), {
"create_session": Mock(side_effect=ConnectionError("timeout")),
})
provider = BrowserUseProviderFake()
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
with caplog.at_level(logging.WARNING, logger="tools.browser_tool"):
session = browser_tool._get_session_info("task-6")
assert session["fallback_from_cloud"] is True
assert any("BrowserUseProvider" in r.message and "timeout" in r.message
for r in caplog.records)
def test_cloud_failure_does_not_poison_next_task(self, monkeypatch):
"""A fallback for one task_id doesn't affect a new task_id when cloud recovers."""
_reset_session_state(monkeypatch)
call_count = 0
def create_session_flaky(task_id):
nonlocal call_count
call_count += 1
if call_count == 1:
raise RuntimeError("transient failure")
return {
"session_name": "cloud-ok",
"bb_session_id": "bb_999",
"cdp_url": None,
"features": {"browser_use": True},
}
provider = Mock()
provider.create_session.side_effect = create_session_flaky
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
# First call fails → fallback
s1 = browser_tool._get_session_info("task-a")
assert s1["fallback_from_cloud"] is True
# Second call (different task) → cloud succeeds
s2 = browser_tool._get_session_info("task-b")
assert "fallback_from_cloud" not in s2
assert s2["session_name"] == "cloud-ok"
def test_cloud_returns_invalid_session_triggers_fallback(self, monkeypatch):
"""Cloud provider returning None or empty dict triggers fallback."""
_reset_session_state(monkeypatch)
provider = Mock()
provider.create_session.return_value = None
monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
session = browser_tool._get_session_info("task-7")
assert session["fallback_from_cloud"] is True
assert "invalid session" in session["fallback_reason"]

View File

@@ -123,7 +123,7 @@ class TestSendMatrix:
session.put.assert_called_once()
call_kwargs = session.put.call_args
url = call_kwargs[0][0]
assert url.startswith("https://matrix.example.com/_matrix/client/v3/rooms/!room:example.com/send/m.room.message/")
assert url.startswith("https://matrix.example.com/_matrix/client/v3/rooms/%21room%3Aexample.com/send/m.room.message/")
assert call_kwargs[1]["headers"]["Authorization"] == "Bearer syt_tok"
payload = call_kwargs[1]["json"]
assert payload["msgtype"] == "m.text"

View File

@@ -12,6 +12,7 @@ from gateway.config import Platform
from tools.send_message_tool import (
_parse_target_ref,
_send_discord,
_send_matrix_via_adapter,
_send_telegram,
_send_to_platform,
send_message_tool,
@@ -576,7 +577,7 @@ class TestSendToPlatformChunking:
sent_calls = []
async def fake_send(token, chat_id, message, media_files=None, thread_id=None):
async def fake_send(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False):
sent_calls.append(media_files or [])
return {"success": True, "platform": "telegram", "chat_id": chat_id, "message_id": str(len(sent_calls))}
@@ -594,6 +595,103 @@ class TestSendToPlatformChunking:
assert all(call == [] for call in sent_calls[:-1])
assert sent_calls[-1] == media
def test_matrix_media_uses_native_adapter_helper(self):
doc_path = Path("/tmp/test-send-message-matrix.pdf")
doc_path.write_bytes(b"%PDF-1.4 test")
try:
helper = AsyncMock(return_value={"success": True, "platform": "matrix", "chat_id": "!room:example.com", "message_id": "$evt"})
with patch("tools.send_message_tool._send_matrix_via_adapter", helper):
result = asyncio.run(
_send_to_platform(
Platform.MATRIX,
SimpleNamespace(enabled=True, token="tok", extra={"homeserver": "https://matrix.example.com"}),
"!room:example.com",
"here you go",
media_files=[(str(doc_path), False)],
)
)
assert result["success"] is True
helper.assert_awaited_once()
call = helper.await_args
assert call.args[1] == "!room:example.com"
assert call.args[2] == "here you go"
assert call.kwargs["media_files"] == [(str(doc_path), False)]
finally:
doc_path.unlink(missing_ok=True)
def test_matrix_text_only_uses_lightweight_path(self):
"""Text-only Matrix sends should NOT go through the heavy adapter path."""
helper = AsyncMock()
lightweight = AsyncMock(return_value={"success": True, "platform": "matrix", "chat_id": "!room:ex.com", "message_id": "$txt"})
with patch("tools.send_message_tool._send_matrix_via_adapter", helper), \
patch("tools.send_message_tool._send_matrix", lightweight):
result = asyncio.run(
_send_to_platform(
Platform.MATRIX,
SimpleNamespace(enabled=True, token="tok", extra={"homeserver": "https://matrix.example.com"}),
"!room:ex.com",
"just text, no files",
)
)
assert result["success"] is True
helper.assert_not_awaited()
lightweight.assert_awaited_once()
def test_send_matrix_via_adapter_sends_document(self, tmp_path):
file_path = tmp_path / "report.pdf"
file_path.write_bytes(b"%PDF-1.4 test")
calls = []
class FakeAdapter:
def __init__(self, _config):
self.connected = False
async def connect(self):
self.connected = True
calls.append(("connect",))
return True
async def send(self, chat_id, message, metadata=None):
calls.append(("send", chat_id, message, metadata))
return SimpleNamespace(success=True, message_id="$text")
async def send_document(self, chat_id, file_path, metadata=None):
calls.append(("send_document", chat_id, file_path, metadata))
return SimpleNamespace(success=True, message_id="$file")
async def disconnect(self):
calls.append(("disconnect",))
fake_module = SimpleNamespace(MatrixAdapter=FakeAdapter)
with patch.dict(sys.modules, {"gateway.platforms.matrix": fake_module}):
result = asyncio.run(
_send_matrix_via_adapter(
SimpleNamespace(enabled=True, token="tok", extra={"homeserver": "https://matrix.example.com"}),
"!room:example.com",
"report attached",
media_files=[(str(file_path), False)],
)
)
assert result == {
"success": True,
"platform": "matrix",
"chat_id": "!room:example.com",
"message_id": "$file",
}
assert calls == [
("connect",),
("send", "!room:example.com", "report attached", None),
("send_document", "!room:example.com", str(file_path), None),
("disconnect",),
]
# ---------------------------------------------------------------------------
# HTML auto-detection in Telegram send
@@ -658,6 +756,17 @@ class TestSendTelegramHtmlDetection:
kwargs = bot.send_message.await_args.kwargs
assert kwargs["parse_mode"] == "MarkdownV2"
def test_disable_link_previews_sets_disable_web_page_preview(self, monkeypatch):
bot = self._make_bot()
_install_telegram_mock(monkeypatch, bot)
asyncio.run(
_send_telegram("tok", "123", "https://example.com", disable_link_previews=True)
)
kwargs = bot.send_message.await_args.kwargs
assert kwargs["disable_web_page_preview"] is True
def test_html_with_code_and_pre_tags(self, monkeypatch):
bot = self._make_bot()
_install_telegram_mock(monkeypatch, bot)
@@ -707,6 +816,23 @@ class TestSendTelegramHtmlDetection:
second_call = bot.send_message.await_args_list[1].kwargs
assert second_call["parse_mode"] is None
def test_transient_bad_gateway_retries_text_send(self, monkeypatch):
bot = self._make_bot()
bot.send_message = AsyncMock(
side_effect=[
Exception("502 Bad Gateway"),
SimpleNamespace(message_id=2),
]
)
_install_telegram_mock(monkeypatch, bot)
with patch("asyncio.sleep", new=AsyncMock()) as sleep_mock:
result = asyncio.run(_send_telegram("tok", "123", "hello"))
assert result["success"] is True
assert bot.send_message.await_count == 2
sleep_mock.assert_awaited_once()
# ---------------------------------------------------------------------------
# Tests for Discord thread_id support