"""Tests for gateway /compress user-facing messaging.""" from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch import pytest from gateway.config import GatewayConfig, Platform, PlatformConfig from gateway.platforms.base import MessageEvent from gateway.session import SessionEntry, SessionSource, build_session_key def _make_source() -> SessionSource: return SessionSource( platform=Platform.TELEGRAM, user_id="u1", chat_id="c1", user_name="tester", chat_type="dm", ) def _make_event(text: str = "/compress") -> MessageEvent: return MessageEvent(text=text, source=_make_source(), message_id="m1") def _make_history() -> list[dict[str, str]]: return [ {"role": "user", "content": "one"}, {"role": "assistant", "content": "two"}, {"role": "user", "content": "three"}, {"role": "assistant", "content": "four"}, ] def _make_runner(history: list[dict[str, str]]): from gateway.run import GatewayRunner runner = object.__new__(GatewayRunner) runner.config = GatewayConfig( platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")} ) 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.session_store = MagicMock() runner.session_store.get_or_create_session.return_value = session_entry runner.session_store.load_transcript.return_value = history runner.session_store.rewrite_transcript = MagicMock() runner.session_store.update_session = MagicMock() runner.session_store._save = MagicMock() runner._session_db = None return runner @pytest.mark.asyncio async def test_compress_command_works_when_auto_compaction_disabled(): """compression.enabled: false disables *automatic* compaction only. The gateway /compress handler has never gated on the flag — pin that contract (every manual-compress surface must allow manual compression regardless of the auto toggle, #64438) and the force=True cooldown bypass that manual compression relies on.""" history = _make_history() compressed = [ history[0], {"role": "assistant", "content": "compressed summary"}, history[-1], ] runner = _make_runner(history) agent_instance = MagicMock() agent_instance.shutdown_memory_provider = MagicMock() agent_instance.close = MagicMock() agent_instance._cached_system_prompt = "" agent_instance.tools = None agent_instance.compression_enabled = False agent_instance.context_compressor.has_content_to_compress.return_value = True agent_instance.session_id = "sess-1" agent_instance._compress_context.return_value = (compressed, "") # Explicit non-lock-skip: MagicMock getattr would return a truthy mock. agent_instance._compression_skipped_due_to_lock = False def _estimate(messages, **_kwargs): return 100 if messages == history else 60 with ( patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}), patch("gateway.run._resolve_gateway_model", return_value="test-model"), patch("run_agent.AIAgent", return_value=agent_instance), patch("agent.model_metadata.estimate_request_tokens_rough", side_effect=_estimate), ): result = await runner._handle_compress_command(_make_event()) assert "disabled" not in result.lower() assert "Compressed:" in result agent_instance._compress_context.assert_called_once() assert agent_instance._compress_context.call_args.kwargs.get("force") is True @pytest.mark.asyncio async def test_compress_command_surfaces_aux_model_failure_even_when_recovered(): """When the user's configured ``auxiliary.compression.model`` errors out but compression recovers by retrying on the main model, /compress must STILL inform the user. Silent recovery hides broken config the user needs to fix.""" history = _make_history() # Compressed transcript — normal successful compression, no placeholder. compressed = [ history[0], {"role": "assistant", "content": "summary via main model"}, history[-1], ] runner = _make_runner(history) agent_instance = MagicMock() agent_instance.shutdown_memory_provider = MagicMock() agent_instance.close = MagicMock() agent_instance._cached_system_prompt = "" agent_instance.tools = None agent_instance.context_compressor.has_content_to_compress.return_value = True # Fallback placeholder was NOT used — recovery succeeded. agent_instance.context_compressor._last_compress_aborted = False agent_instance.context_compressor._last_summary_fallback_used = False agent_instance.context_compressor._last_summary_dropped_count = 0 agent_instance.context_compressor._last_summary_error = None # But the configured aux model DID fail before the retry succeeded. agent_instance.context_compressor._last_aux_model_failure_model = ( "gemini-3-flash-preview" ) agent_instance.context_compressor._last_aux_model_failure_error = ( "404 model not found: gemini-3-flash-preview" ) agent_instance.session_id = "sess-1" agent_instance._compress_context.return_value = (compressed, "") agent_instance._compression_skipped_due_to_lock = False def _estimate(messages, **_kwargs): if messages == history: return 100 if messages == compressed: return 60 raise AssertionError(f"unexpected transcript: {messages!r}") with ( patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "***"}), patch("gateway.run._resolve_gateway_model", return_value="test-model"), patch("run_agent.AIAgent", return_value=agent_instance), patch("agent.model_metadata.estimate_request_tokens_rough", side_effect=_estimate), ): result = await runner._handle_compress_command(_make_event()) # Compression succeeded assert "Compressed:" in result # No ⚠️ warning (that's reserved for dropped-turns case) assert "⚠️" not in result # But there IS an info note about the broken aux model assert "ℹ️" in result assert "gemini-3-flash-preview" in result assert "404" in result assert "auxiliary.compression.model" in result # The user's context is explicitly called out as intact assert "intact" in result agent_instance.shutdown_memory_provider.assert_called_once() agent_instance.close.assert_called_once() @pytest.mark.asyncio async def test_compress_command_in_place_skips_destructive_rewrite(): """In-place compaction (compression.in_place / #38763) persists via archive_and_compact() inside _compress_context — the previous active rows are soft-archived and the compacted set inserted. Calling rewrite_transcript() afterwards would invoke replace_messages(active_only=False), DELETEing the just-archived rows (silent data loss, #61145). The handler must skip the rewrite and still report success.""" history = _make_history() compressed = [ history[0], {"role": "assistant", "content": "compacted summary"}, history[-1], ] runner = _make_runner(history) runner._session_db = object() session_entry = runner.session_store.get_or_create_session.return_value runner.session_store.rewrite_transcript = MagicMock() agent_instance = MagicMock() agent_instance.shutdown_memory_provider = MagicMock() agent_instance.close = MagicMock() agent_instance._cached_system_prompt = "" agent_instance.tools = None agent_instance.context_compressor.has_content_to_compress.return_value = True # In-place compaction: session_id is UNCHANGED but marked as a success. agent_instance._last_compaction_in_place = True agent_instance.session_id = "sess-1" agent_instance._compress_context.return_value = (compressed, "") agent_instance._compression_skipped_due_to_lock = False def _estimate(messages, **_kwargs): if messages == history: return 100 if messages == compressed: return 60 raise AssertionError(f"unexpected transcript: {messages!r}") with ( patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "***"}), patch("gateway.run._resolve_gateway_model", return_value="test-model"), patch("run_agent.AIAgent", return_value=agent_instance), patch("agent.model_metadata.estimate_request_tokens_rough", side_effect=_estimate), ): result = await runner._handle_compress_command(_make_event()) assert "Compressed:" in result # The destructive rewrite must NOT run — archive_and_compact() already # persisted, and rewrite_transcript would wipe the archived rows. runner.session_store.rewrite_transcript.assert_not_called() assert session_entry.session_id == "sess-1" agent_instance.shutdown_memory_provider.assert_called_once() agent_instance.close.assert_called_once() @pytest.mark.asyncio async def test_compress_command_preserves_platform_and_gateway_session_key(): """The temporary compression agent must carry the originating source's platform and stable gateway session key, matching a normal gateway turn. Without them ``_session_source_for_agent`` falls back to a default "cli" host source, so an external context engine misattributes the retained transcript tail and later duplicates it on resume (#50422).""" history = _make_history() runner = _make_runner(history) agent_instance = MagicMock() agent_instance.shutdown_memory_provider = MagicMock() agent_instance.close = MagicMock() agent_instance._cached_system_prompt = "" agent_instance.tools = None agent_instance.context_compressor.has_content_to_compress.return_value = True agent_instance.session_id = "sess-1" agent_instance._compress_context.return_value = (list(history), "") agent_instance._compression_skipped_due_to_lock = False with ( patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}), patch("gateway.run._resolve_gateway_model", return_value="test-model"), patch("run_agent.AIAgent", return_value=agent_instance) as mock_agent, patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100), ): await runner._handle_compress_command(_make_event()) assert mock_agent.call_count == 1 _, kwargs = mock_agent.call_args # Platform preserved as the live turn's config key (TELEGRAM -> "telegram"), # not the unbound "cli"/"local" fallback. assert kwargs.get("platform") == "telegram" # Stable gateway session key preserved, identical to a normal gateway turn. assert kwargs.get("gateway_session_key") == runner._session_key_for_source(_make_source()) assert kwargs["gateway_session_key"] @pytest.mark.asyncio async def test_compress_command_passes_tool_messages_to_compressor(): """Tool results must reach _compress_context (#3854). Filtering the transcript to user/assistant-only starved the compressor's tool-result pruning — tool messages are usually the bulk of the context. """ history = [ {"role": "user", "content": "run it"}, { "role": "assistant", "content": None, "tool_calls": [{"id": "t1", "type": "function", "function": {"name": "x", "arguments": "{}"}}], }, {"role": "tool", "content": "BIG RESULT " * 50, "tool_call_id": "t1"}, {"role": "assistant", "content": "done"}, {"role": "user", "content": "thanks"}, {"role": "assistant", "content": "np"}, ] runner = _make_runner(history) agent_instance = MagicMock() agent_instance.shutdown_memory_provider = MagicMock() agent_instance.close = MagicMock() agent_instance._cached_system_prompt = "" agent_instance.tools = None agent_instance.context_compressor.has_content_to_compress.return_value = True agent_instance.session_id = "sess-1" agent_instance._compress_context.return_value = (list(history), "") with ( patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}), patch("gateway.run._resolve_gateway_model", return_value="test-model"), patch("run_agent.AIAgent", return_value=agent_instance), patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100), ): await runner._handle_compress_command(_make_event()) args, _kwargs = agent_instance._compress_context.call_args passed = args[0] roles = [m.get("role") for m in passed] assert "tool" in roles, f"tool messages filtered out: {roles}" # Assistant tool_calls stubs (content=None) must survive too, or the # tool message would dangle without its call. assert any(m.get("tool_calls") for m in passed), "assistant tool_calls stub dropped" @pytest.mark.asyncio async def test_compress_command_multiplexed_runs_under_profile_secret_scope(tmp_path): """Manual /compress must install the source profile's secret scope. Multiplexed gateways resolve credentials fail-closed (Workstream A): ``get_secret`` raises ``UnscopedSecretError`` on any read outside a ``set_secret_scope`` block. The agent turn is scoped by ``_run_agent``'s wrapper, but slash-command dispatch is not — manual /compress reached the compressor's provider resolution unscoped and died with ``get_secret('OPENROUTER_BASE_URL') called with no profile secret scope active``. The credential read happens inside the executor hop, so this also pins that the handler uses the contextvar-preserving executor (``_run_in_executor_with_context``), not a bare ``run_in_executor``. """ from agent import secret_scope as ss history = _make_history() compressed = [ history[0], {"role": "assistant", "content": "compressed summary"}, history[-1], ] runner = _make_runner(history) runner.config = GatewayConfig( platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")}, multiplex_profiles=True, ) profile_home = tmp_path / "profiles" / "milo" profile_home.mkdir(parents=True) (profile_home / ".env").write_text( "OPENROUTER_BASE_URL=https://scoped.example/v1\n" ) runner._resolve_profile_home_for_source = MagicMock(return_value=profile_home) agent_instance = MagicMock() agent_instance.shutdown_memory_provider = MagicMock() agent_instance.close = MagicMock() agent_instance._cached_system_prompt = "" agent_instance.tools = None agent_instance.context_compressor.has_content_to_compress.return_value = True agent_instance.context_compressor._last_compress_aborted = False agent_instance.context_compressor._last_summary_fallback_used = False agent_instance.context_compressor._last_summary_dropped_count = 0 agent_instance.context_compressor._last_summary_error = None agent_instance.context_compressor._last_aux_model_failure_model = None agent_instance.context_compressor._last_aux_model_failure_error = None agent_instance.session_id = "sess-1" agent_instance._compression_skipped_due_to_lock = False seen: dict[str, str | None] = {} def _compress(*_args, **_kwargs): # Runs in the executor thread — exactly where the aux client # resolves provider credentials. Fail-closed get_secret raises # here unless the profile scope survived the thread hop. seen["base_url"] = ss.get_secret("OPENROUTER_BASE_URL") return (compressed, "") agent_instance._compress_context.side_effect = _compress ss.set_multiplex_active(True) try: with ( patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "***"}), patch("gateway.run._resolve_gateway_model", return_value="test-model"), patch("run_agent.AIAgent", return_value=agent_instance), patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100), ): result = await runner._handle_compress_command(_make_event()) finally: ss.set_multiplex_active(False) runner._shutdown_executor() assert "failed" not in result.lower(), result assert seen["base_url"] == "https://scoped.example/v1" runner._resolve_profile_home_for_source.assert_called_once() @pytest.mark.asyncio async def test_compress_command_single_profile_skips_profile_resolution(): """Multiplexing off → the scope wrapper is a transparent pass-through. Single-profile gateways must not pay the profile-resolution path (and ``_resolve_profile_home_for_source`` assumes multiplex config exists) — mirrors the gating contract of ``_run_agent``'s wrapper. """ history = _make_history() runner = _make_runner(history) runner._resolve_profile_home_for_source = MagicMock() agent_instance = MagicMock() agent_instance.shutdown_memory_provider = MagicMock() agent_instance.close = MagicMock() agent_instance._cached_system_prompt = "" agent_instance.tools = None agent_instance.context_compressor.has_content_to_compress.return_value = True agent_instance.session_id = "sess-1" agent_instance._compress_context.return_value = (list(history), "") agent_instance._compression_skipped_due_to_lock = False with ( patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "***"}), patch("gateway.run._resolve_gateway_model", return_value="test-model"), patch("run_agent.AIAgent", return_value=agent_instance), patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100), ): await runner._handle_compress_command(_make_event()) runner._resolve_profile_home_for_source.assert_not_called() runner._shutdown_executor()