"""Tests for cron/scheduler.py — origin resolution, delivery routing, and error logging.""" import contextlib import itertools import json import logging import os from unittest.mock import AsyncMock, patch, MagicMock import pytest from cron.scheduler import _resolve_origin, _resolve_delivery_target, _deliver_result, _send_media_via_adapter, run_job, SILENT_MARKER, _build_job_prompt, _resolve_cron_enabled_toolsets, _merge_mcp_into_per_job_toolsets from tools.env_passthrough import clear_env_passthrough from tools.credential_files import clear_credential_files class TestPerJobToolsetMcpMerge: """A per-job enabled_toolsets allowlist must not silently drop MCP servers.""" CFG = { "mcp_servers": { "finnhub": {"enabled": True}, "playwright": {"enabled": True}, "disabled_one": {"enabled": False}, "string_enabled": {"enabled": "true"}, "not_a_dict": "ignored", } } def _enabled_names(self): return {"finnhub", "playwright", "string_enabled"} def test_native_only_list_gets_all_enabled_mcp_servers(self): result = _merge_mcp_into_per_job_toolsets(["web", "terminal"], self.CFG) assert result[:2] == ["web", "terminal"] assert set(result) == {"web", "terminal"} | self._enabled_names() def test_explicit_mcp_name_is_treated_as_allowlist(self): # User named one server -> add nothing further. result = _merge_mcp_into_per_job_toolsets(["web", "finnhub"], self.CFG) assert result == ["web", "finnhub"] assert "playwright" not in result def test_no_mcp_sentinel_opts_out_and_is_stripped(self): result = _merge_mcp_into_per_job_toolsets(["web", "no_mcp"], self.CFG) assert result == ["web"] assert not (set(result) & self._enabled_names()) def test_resolver_empty_per_job_falls_through_to_platform(self): # No per-job list -> must delegate to _get_platform_tools (the platform # fallback), NOT the per-job merge. Stub the platform resolver and assert # it is the path taken and its result is returned. job = {"enabled_toolsets": None} sentinel = ["web", "finnhub"] with patch("hermes_cli.tools_config._get_platform_tools", return_value=set(sentinel)) as m_platform: result = _resolve_cron_enabled_toolsets(job, self.CFG) m_platform.assert_called_once() # _get_platform_tools args: (cfg, "cron") assert m_platform.call_args[0][1] == "cron" assert set(result) == set(sentinel) class TestResolveOrigin: def test_full_origin(self): job = { "origin": { "platform": "telegram", "chat_id": "123456", "chat_name": "Test Chat", "thread_id": "42", } } result = _resolve_origin(job) assert isinstance(result, dict) assert result == job["origin"] assert result["platform"] == "telegram" assert result["chat_id"] == "123456" assert result["chat_name"] == "Test Chat" assert result["thread_id"] == "42" @pytest.mark.parametrize( "non_dict_origin", [ "combined-digest-replaces-x-and-y-20260503", 123, ["telegram", "12345"], ("platform", "chat_id"), 42.0, ], ) def test_non_dict_origin_returns_none_instead_of_crashing(self, non_dict_origin): """Non-dict origins (provenance strings from hand-edited or migrated jobs.json) must be treated as missing instead of crashing the scheduler tick on ``origin.get('platform')`` with ``'str' object has no attribute 'get'`` (#18722). Before this guard a job in this state crashed every fire attempt forever; ``mark_job_run`` recorded the error but the next tick re-loaded the poisoned origin and crashed identically. """ job = {"origin": non_dict_origin} assert _resolve_origin(job) is None class TestResolveDeliveryTarget: def test_origin_delivery_preserves_thread_id(self): job = { "deliver": "origin", "origin": { "platform": "telegram", "chat_id": "-1001", "thread_id": "17585", }, } assert _resolve_delivery_target(job) == { "platform": "telegram", "chat_id": "-1001", "thread_id": "17585", } def test_bare_platform_delivery_uses_home_root_instead_of_origin_thread(self, monkeypatch): monkeypatch.setenv("DISCORD_HOME_CHANNEL", "home-parent") monkeypatch.delenv("DISCORD_HOME_CHANNEL_THREAD_ID", raising=False) job = { "deliver": "discord", "origin": { "platform": "discord", "chat_id": "origin-parent", "thread_id": "origin-thread", }, } assert _resolve_delivery_target(job) == { "platform": "discord", "chat_id": "home-parent", "thread_id": None, } def test_telegram_cron_thread_id_overrides_home_thread_id(self, monkeypatch): """TELEGRAM_CRON_THREAD_ID wins over TELEGRAM_HOME_CHANNEL_THREAD_ID for cron (#24409).""" monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "-1001234567890") monkeypatch.setenv("TELEGRAM_HOME_CHANNEL_THREAD_ID", "5") monkeypatch.setenv("TELEGRAM_CRON_THREAD_ID", "42") assert _resolve_delivery_target({"deliver": "telegram"}) == { "platform": "telegram", "chat_id": "-1001234567890", "thread_id": "42", } def test_explicit_telegram_topic_target_overrides_cron_thread_id(self, monkeypatch): """Explicit ``telegram:chat:thread`` targets bypass TELEGRAM_CRON_THREAD_ID.""" monkeypatch.setenv("TELEGRAM_CRON_THREAD_ID", "999") job = {"deliver": "telegram:-1003724596514:17"} assert _resolve_delivery_target(job) == { "platform": "telegram", "chat_id": "-1003724596514", "thread_id": "17", } def test_human_friendly_label_resolved_via_channel_directory(self): """deliver: 'whatsapp:Alice (dm)' resolves to the real JID.""" job = {"deliver": "whatsapp:Alice (dm)"} with patch( "gateway.channel_directory.resolve_channel_name", return_value="12345678901234@lid", ) as resolve_mock: result = _resolve_delivery_target(job) resolve_mock.assert_called_once_with("whatsapp", "Alice (dm)") assert result == { "platform": "whatsapp", "chat_id": "12345678901234@lid", "thread_id": None, } def test_raw_id_not_mangled_when_directory_returns_none(self): """deliver: 'whatsapp:12345@lid' passes through when directory has no match.""" job = {"deliver": "whatsapp:12345@lid"} with patch( "gateway.channel_directory.resolve_channel_name", return_value=None, ): result = _resolve_delivery_target(job) assert result == { "platform": "whatsapp", "chat_id": "12345@lid", "thread_id": None, } def test_list_form_deliver_is_normalized(self, monkeypatch): """deliver=['telegram'] (Python list) should resolve like 'telegram' string. Regression test for #17139: MCP clients / scripts that pass the deliver field as an array-shaped value used to fail with "no delivery target resolved for deliver=['telegram']" because ``str(['telegram'])`` was passed through to ``split(',')`` verbatim. """ monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "-4004") job = { "deliver": ["telegram"], "origin": None, } assert _resolve_delivery_target(job) == { "platform": "telegram", "chat_id": "-4004", "thread_id": None, } class TestRoutingIntents: """``all`` routing intent expands at fire time.""" def test_all_expands_to_every_connected_home_channel(self, monkeypatch): """deliver='all' fans out to every platform with a configured home channel.""" from cron.scheduler import _resolve_delivery_targets monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "-111") monkeypatch.setenv("DISCORD_HOME_CHANNEL", "-222") monkeypatch.setenv("SLACK_HOME_CHANNEL", "C333") # Sanity: platforms without the env var must NOT appear in the expansion. monkeypatch.delenv("SIGNAL_HOME_CHANNEL", raising=False) monkeypatch.delenv("MATRIX_HOME_ROOM", raising=False) targets = _resolve_delivery_targets({"deliver": "all", "origin": None}) platforms = sorted(t["platform"] for t in targets) assert "telegram" in platforms assert "discord" in platforms assert "slack" in platforms assert "signal" not in platforms assert "matrix" not in platforms class TestDeliverResultWrapping: """Verify that cron deliveries are wrapped with header/footer and no longer mirrored.""" def _safe_media_path(self, tmp_path, monkeypatch, name, data=b"media"): root = tmp_path / "media-cache" media_file = root / name media_file.parent.mkdir(parents=True, exist_ok=True) media_file.write_bytes(data) monkeypatch.setattr( "gateway.platforms.base.MEDIA_DELIVERY_SAFE_ROOTS", (root,), ) return media_file.resolve() def test_delivery_wraps_content_with_header_and_footer(self): """Delivered content should include task name header and agent-invisible note.""" from gateway.config import Platform pconfig = MagicMock() pconfig.enabled = True mock_cfg = MagicMock() mock_cfg.platforms = {Platform.TELEGRAM: pconfig} with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})) as send_mock: job = { "id": "test-job", "name": "daily-report", "deliver": "origin", "origin": {"platform": "telegram", "chat_id": "123"}, } _deliver_result(job, "Here is today's summary.") send_mock.assert_called_once() sent_content = send_mock.call_args.kwargs.get("content") or send_mock.call_args[0][-1] assert "Cronjob Response: daily-report" in sent_content assert "(job_id: test-job)" in sent_content assert "-------------" in sent_content assert "Here is today's summary." in sent_content assert "To stop or manage this job" in sent_content def test_relay_fronted_home_uses_relay_config_and_live_adapter(self, monkeypatch, tmp_path): """Persisted Slack home survives restart without native Slack config.""" from concurrent.futures import Future from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig relay = MagicMock() relay.fronts_platform.side_effect = lambda platform: platform == Platform.SLACK relay.send_for_platform = AsyncMock(return_value=MagicMock(success=True)) relay.send_voice = AsyncMock(return_value=MagicMock(success=True)) relay.supports_inchannel_continuable = False config = GatewayConfig( platforms={ Platform.RELAY: PlatformConfig(enabled=True), Platform.SLACK: PlatformConfig( enabled=False, home_channel=HomeChannel( platform=Platform.SLACK, chat_id="D123", name="Owner DM", user_id="U123", ), ), }, ) loop = MagicMock() loop.is_running.return_value = True def fake_run_coro(coro, _loop): import asyncio as _asyncio future = Future() try: future.set_result(_asyncio.run(coro)) except BaseException as exc: # noqa: BLE001 future.set_exception(exc) return future standalone_send = AsyncMock(return_value={"success": True}) media_path = self._safe_media_path(tmp_path, monkeypatch, "relay-voice.mp3") monkeypatch.setenv("SLACK_HOME_CHANNEL", "D123") job = { "id": "relay-cron", "deliver": "slack", } with ( patch("gateway.config.load_gateway_config", return_value=config), patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro), patch("tools.send_message_tool._send_to_platform", new=standalone_send), ): result = _deliver_result( job, f"scheduled result\nMEDIA:{media_path}", adapters={Platform.RELAY: relay}, loop=loop, ) assert result is None relay.send_for_platform.assert_awaited_once() args = relay.send_for_platform.await_args.args assert args[:3] == (Platform.SLACK, "D123", "scheduled result") assert relay.send_for_platform.await_args.kwargs["metadata"]["user_id"] == "U123" relay.send_voice.assert_awaited_once() media_metadata = relay.send_voice.await_args.kwargs["metadata"] assert media_metadata["_relay_logical_platform"] == "slack" assert media_metadata["user_id"] == "U123" standalone_send.assert_not_awaited() def test_live_adapter_sends_media_as_attachments(self, tmp_path, monkeypatch): """When a live adapter is available, MEDIA files should be sent as native platform attachments (e.g., Discord voice, Telegram audio) rather than as literal 'MEDIA:/path' text.""" from gateway.config import Platform from concurrent.futures import Future media_path = self._safe_media_path(tmp_path, monkeypatch, "cron-voice.mp3") adapter = AsyncMock() adapter.send.return_value = MagicMock(success=True) adapter.send_voice.return_value = MagicMock(success=True) pconfig = MagicMock() pconfig.enabled = True mock_cfg = MagicMock() mock_cfg.platforms = {Platform.DISCORD: pconfig} loop = MagicMock() loop.is_running.return_value = True # run_coroutine_threadsafe returns concurrent.futures.Future (has timeout kwarg) def fake_run_coro(coro, _loop): # Actually run the routed coroutine (router._deliver_to_platform) # so the underlying adapter.send is invoked, then wrap the real # result in a completed Future (matching run_coroutine_threadsafe). import asyncio as _asyncio future = Future() try: future.set_result(_asyncio.run(coro)) except BaseException as _e: # noqa: BLE001 future.set_exception(_e) return future job = { "id": "tts-job", "deliver": "origin", "origin": {"platform": "discord", "chat_id": "9876"}, } with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro): _deliver_result( job, f"Here is TTS\nMEDIA:{media_path}", adapters={Platform.DISCORD: adapter}, loop=loop, ) # Text should be sent without the MEDIA tag adapter.send.assert_called_once() text_sent = adapter.send.call_args[0][1] assert "MEDIA:" not in text_sent assert "Here is TTS" in text_sent # Audio file should be sent as a voice attachment adapter.send_voice.assert_called_once() voice_call = adapter.send_voice.call_args assert voice_call[1]["audio_path"] == str(media_path) class TestDeliverResultErrorReturns: """Verify _deliver_result returns error strings on failure, None on success.""" def test_returns_error_when_platform_disabled(self): from gateway.config import Platform pconfig = MagicMock() pconfig.enabled = False mock_cfg = MagicMock() mock_cfg.platforms = {Platform.TELEGRAM: pconfig} with patch("gateway.config.load_gateway_config", return_value=mock_cfg): job = { "id": "disabled", "deliver": "origin", "origin": {"platform": "telegram", "chat_id": "123"}, } result = _deliver_result(job, "Output.") assert result is not None assert "not configured" in result class TestRunJobSessionPersistence: def test_run_job_passes_session_db_and_cron_platform(self, tmp_path): job = { "id": "test-job", "name": "test", "prompt": "hello", } fake_db = MagicMock() fake_db.get_compression_tip.side_effect = lambda session_id: session_id with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state.SessionDB", return_value=fake_db), \ patch( "hermes_cli.runtime_provider.resolve_runtime_provider", return_value={ "api_key": "test-key", "base_url": "https://example.invalid/v1", "provider": "openrouter", "api_mode": "chat_completions", }, ), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.return_value = {"final_response": "ok"} mock_agent_cls.return_value = mock_agent success, output, final_response, error = run_job(job) assert success is True assert error is None assert final_response == "ok" assert "ok" in output kwargs = mock_agent_cls.call_args.kwargs assert kwargs["session_db"] is fake_db assert kwargs["platform"] == "cron" assert kwargs["session_id"].startswith("cron_test-job_") original_session_id = kwargs["session_id"] fake_db.get_compression_tip.assert_called_once_with(original_session_id) fake_db.end_session.assert_called_once() call_args = fake_db.end_session.call_args assert call_args[0][0] == original_session_id assert call_args[0][1] == "cron_complete" fake_db.close.assert_called_once() mock_agent.close.assert_called_once() @contextlib.contextmanager def _run_job_patches(self, tmp_path, extra=()): """Apply every patch run_job tests need, as one bundle. Yields ``(fake_db, mock_agent_cls)``. Using an ExitStack that enters the whole list means a caller can never silently drop a patch by index — the previous positional-list form let a seam split shift ``resolve_runtime_provider`` off the end of the applied slice, so the real resolver ran and (only on a dev machine with ambient creds) hid an auth failure that CI then caught. Every test enters all patches. ``extra`` is an iterable of additional context managers (e.g. a per-test ``_get_platform_tools`` patch) entered alongside the base set. """ fake_db = MagicMock() mock_agent = MagicMock() mock_agent.run_conversation.return_value = {"final_response": "ok"} base = [ patch("cron.scheduler._hermes_home", tmp_path), patch("cron.scheduler._resolve_origin", return_value=None), patch("hermes_cli.env_loader.load_hermes_dotenv"), patch("hermes_cli.env_loader.reset_secret_source_cache"), patch("hermes_state.SessionDB", return_value=fake_db), patch( "hermes_cli.runtime_provider.resolve_runtime_provider", return_value={ "api_key": "test-key", "base_url": "https://example.invalid/v1", "provider": "openrouter", "api_mode": "chat_completions", }, ), patch("run_agent.AIAgent", return_value=mock_agent), ] with contextlib.ExitStack() as stack: entered = [stack.enter_context(cm) for cm in base] for cm in extra: stack.enter_context(cm) mock_agent_cls = entered[-1] # the AIAgent patch yield fake_db, mock_agent_cls def test_tick_skips_due_jobs_while_dispatch_is_paused(self, tmp_path): """The drain gate runs before advancing a due job's schedule.""" from cron.scheduler import tick job = { "id": "paused-due-job", "name": "paused due job", "schedule": {"kind": "interval", "seconds": 60}, "next_run_at": "2020-01-01T00:00:00+00:00", "enabled": True, } with patch("cron.scheduler.get_due_jobs", return_value=[job]), patch( "cron.scheduler.advance_next_runs" ) as advance, patch("cron.scheduler.run_one_job") as run_one: assert tick(verbose=False, sync=True, can_dispatch=lambda: False) == 0 advance.assert_not_called() run_one.assert_not_called() class TestRunJobConfigLogging: """Verify that config.yaml parse failures are logged, not silently swallowed.""" def test_bad_config_yaml_is_logged(self, caplog, tmp_path): """When config.yaml is malformed, a warning should be logged.""" bad_yaml = tmp_path / "config.yaml" bad_yaml.write_text("invalid: yaml: [[[bad") job = { "id": "test-job", "name": "test", "prompt": "hello", } # Mock heavy post-yaml work so the test only exercises the warning # path. Without these mocks, run_job continues into provider # resolution and MCP discovery, both of which can spawn subprocesses # / hit the network and have caused this test to time out on CI # (>30s wall clock) under load. See PR #33661 follow-up. with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value={"provider": "openrouter", "api_key": "x", "base_url": "https://example.invalid", "api_mode": "chat_completions"}), \ patch("tools.mcp_tool.discover_mcp_tools", return_value=[]), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.return_value = {"final_response": "ok"} mock_agent_cls.return_value = mock_agent with caplog.at_level(logging.WARNING, logger="cron.scheduler"): run_job(job) assert any("failed to load config.yaml" in r.message for r in caplog.records), \ f"Expected 'failed to load config.yaml' warning in logs, got: {[r.message for r in caplog.records]}" class TestRunJobConfigEnvVarExpansion: """Verify that ${VAR} references in config.yaml are expanded when running cron jobs.""" _RUNTIME = { "api_key": "test-key", "base_url": "https://example.invalid/v1", "provider": "openrouter", "api_mode": "chat_completions", } def test_model_env_ref_in_config_yaml_is_expanded(self, tmp_path, monkeypatch): """${VAR} in config.yaml model: is expanded using env after .env is loaded.""" (tmp_path / "config.yaml").write_text("model: ${_HERMES_TEST_CRON_MODEL}\n") monkeypatch.setenv("_HERMES_TEST_CRON_MODEL", "gpt-4o-mini-cron-test") job = {"id": "env-job", "name": "env test", "prompt": "hi"} fake_db = MagicMock() with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state.SessionDB", return_value=fake_db), \ patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value=self._RUNTIME), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.return_value = {"final_response": "ok"} mock_agent_cls.return_value = mock_agent success, _, _, error = run_job(job) assert success is True assert error is None kwargs = mock_agent_cls.call_args.kwargs assert kwargs["model"] == "gpt-4o-mini-cron-test", ( f"Expected model='gpt-4o-mini-cron-test', got {kwargs['model']!r}. " "config.yaml ${VAR} was not expanded in the cron execution path." ) def test_auth_fallback_switches_provider_and_model_together(self, tmp_path): """Codex auth failure must produce OpenRouter+GLM, never OpenRouter+GPT.""" from hermes_cli.auth import AuthError (tmp_path / "config.yaml").write_text( "model:\n" " default: gpt-5.6-sol\n" " provider: openai-codex\n" "fallback_providers:\n" " - provider: anthropic\n" " - provider: openrouter\n" " model: z-ai/glm-5.2\n", encoding="utf-8", ) job = { "id": "auth-fallback", "name": "auth fallback", "prompt": "hi", "provider_snapshot": "openai-codex", "model_snapshot": "gpt-5.6-sol", } fake_db = MagicMock() requested = [] def resolve_runtime(**kwargs): requested.append(kwargs.get("requested")) if kwargs.get("requested") in (None, "openai-codex"): # Cron must retain the configured primary provider for drift # comparison even when older/custom AuthError sites omit it. raise AuthError("No Codex credentials stored") assert kwargs["requested"] == "openrouter" assert kwargs["target_model"] == "z-ai/glm-5.2" return {**self._RUNTIME, "provider": "openrouter"} with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state.SessionDB", return_value=fake_db), \ patch("hermes_cli.runtime_provider.resolve_runtime_provider", side_effect=resolve_runtime), \ patch("tools.mcp_tool.discover_mcp_tools", return_value=[]), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.return_value = {"final_response": "ok"} mock_agent_cls.return_value = mock_agent success, _, _, error = run_job(job) assert success is True assert error is None assert requested == [None, "openrouter"] kwargs = mock_agent_cls.call_args.kwargs assert kwargs["provider"] == "openrouter" assert kwargs["model"] == "z-ai/glm-5.2" def test_unexpanded_ref_passthrough_when_var_unset(self, tmp_path, monkeypatch): """When the env var is not set, the literal ${VAR} is kept verbatim (not crashed).""" (tmp_path / "config.yaml").write_text("model: ${_HERMES_TEST_CRON_UNSET_VAR}\n") monkeypatch.delenv("_HERMES_TEST_CRON_UNSET_VAR", raising=False) job = {"id": "unset-job", "name": "unset var test", "prompt": "hi"} fake_db = MagicMock() with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state.SessionDB", return_value=fake_db), \ patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value=self._RUNTIME), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.return_value = {"final_response": "ok"} mock_agent_cls.return_value = mock_agent success, _, _, error = run_job(job) assert success is True kwargs = mock_agent_cls.call_args.kwargs # Unresolved refs are kept verbatim — _expand_env_vars contract assert kwargs["model"] == "${_HERMES_TEST_CRON_UNSET_VAR}" class TestRunJobModelResolution: """Verify defensive model resolution for jobs stored with ``model: null``. Issue #23979: a cron job created without an explicit model is stored as ``model: null``. At fire time the scheduler must: 1. fall back to ``HERMES_MODEL`` env if set, 2. else fall back to config.yaml ``model.default`` if set, 3. else fail fast with an actionable error — never let an empty string reach the provider where it surfaces as an opaque 400. """ _RUNTIME = { "api_key": "test-key", "base_url": "https://example.invalid/v1", "provider": "openrouter", "api_mode": "chat_completions", } def test_null_job_model_falls_back_to_env(self, tmp_path, monkeypatch): """``model: null`` on the job uses HERMES_MODEL when set.""" (tmp_path / "config.yaml").write_text("") monkeypatch.setenv("HERMES_MODEL", "env-model") job = {"id": "null-model-job", "name": "null model", "prompt": "hi", "model": None} fake_db = MagicMock() with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state.SessionDB", return_value=fake_db), \ patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value=self._RUNTIME), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.return_value = {"final_response": "ok"} mock_agent_cls.return_value = mock_agent success, _, _, error = run_job(job) assert success is True assert error is None assert mock_agent_cls.call_args.kwargs["model"] == "env-model" def test_no_model_anywhere_fails_with_actionable_error(self, tmp_path, monkeypatch): """All three sources empty → fail fast with a clear message, not an opaque 400.""" (tmp_path / "config.yaml").write_text("") monkeypatch.delenv("HERMES_MODEL", raising=False) job = {"id": "no-model-job", "name": "no model anywhere", "prompt": "hi", "model": None} fake_db = MagicMock() with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state.SessionDB", return_value=fake_db), \ patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value=self._RUNTIME), \ patch("run_agent.AIAgent") as mock_agent_cls: success, _, _, error = run_job(job) assert success is False assert error is not None assert "no model configured" in error # AIAgent must never be constructed with an empty model — that's # precisely the bug we're guarding against. mock_agent_cls.assert_not_called() def test_config_model_alias_key_resolves(self, tmp_path, monkeypatch): """A ``model: {model: ...}`` alias key resolves like the CLI sibling. ``hermes_cli/oneshot.py``, ``fallback_cmd.py`` and ``prompt_size.py`` all accept ``model.model`` as an alias for ``model.default``. The cron resolver mirrors that so a config that works in the CLI also works in cron. """ (tmp_path / "config.yaml").write_text("model:\n model: alias-key-model\n") monkeypatch.delenv("HERMES_MODEL", raising=False) job = {"id": "alias-job", "name": "alias", "prompt": "hi", "model": None} fake_db = MagicMock() with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state.SessionDB", return_value=fake_db), \ patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value=self._RUNTIME), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.return_value = {"final_response": "ok"} mock_agent_cls.return_value = mock_agent success, _, _, error = run_job(job) assert success is True assert error is None assert mock_agent_cls.call_args.kwargs["model"] == "alias-key-model" def test_corrupt_config_yaml_does_not_crash_with_job_model(self, tmp_path, monkeypatch): """A malformed config.yaml degrades gracefully when the job has a model.""" (tmp_path / "config.yaml").write_text("{{{invalid yaml!!!") monkeypatch.delenv("HERMES_MODEL", raising=False) job = {"id": "corrupt-job", "name": "corrupt", "prompt": "hi", "model": "explicit-model"} fake_db = MagicMock() with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state.SessionDB", return_value=fake_db), \ patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value=self._RUNTIME), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.return_value = {"final_response": "ok"} mock_agent_cls.return_value = mock_agent success, _, _, error = run_job(job) # Explicit job model survives the corrupt-config fall-through. assert success is True assert error is None assert mock_agent_cls.call_args.kwargs["model"] == "explicit-model" class TestRunJobSkillBacked: def test_run_job_preserves_skill_env_passthrough_into_worker_thread(self, tmp_path): job = { "id": "skill-env-job", "name": "skill env test", "prompt": "Use the skill.", "skill": "notion", } fake_db = MagicMock() def _skill_view(name): assert name == "notion" from tools.env_passthrough import register_env_passthrough register_env_passthrough(["NOTION_API_KEY"]) return json.dumps({"success": True, "content": "# notion\nUse Notion."}) def _run_conversation(prompt): from tools.env_passthrough import get_all_passthrough assert "NOTION_API_KEY" in get_all_passthrough() return {"final_response": "ok"} with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state.SessionDB", return_value=fake_db), \ patch( "hermes_cli.runtime_provider.resolve_runtime_provider", return_value={ "api_key": "***", "base_url": "https://example.invalid/v1", "provider": "openrouter", "api_mode": "chat_completions", }, ), \ patch("tools.skills_tool.skill_view", side_effect=_skill_view), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.side_effect = _run_conversation mock_agent_cls.return_value = mock_agent try: success, output, final_response, error = run_job(job) finally: clear_env_passthrough() assert success is True assert error is None assert final_response == "ok" class TestSilentDelivery: """Verify that [SILENT] responses suppress delivery while still saving output.""" def _make_job(self): return { "id": "monitor-job", "name": "monitor", "deliver": "origin", "origin": {"platform": "telegram", "chat_id": "123"}, } def test_silent_response_suppresses_delivery(self, caplog): with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \ patch("cron.scheduler.run_job", return_value=(True, "# output", "[SILENT]", None)), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result") as deliver_mock, \ patch("cron.scheduler.mark_job_run"): from cron.scheduler import tick with caplog.at_level(logging.INFO, logger="cron.scheduler"): tick(verbose=False) deliver_mock.assert_not_called() assert any(SILENT_MARKER in r.message for r in caplog.records) def test_report_quoting_marker_mid_sentence_still_delivers(self): """A genuine report that merely mentions the token mid-sentence must be delivered — the old substring check wrongly swallowed it.""" response = "I considered staying [SILENT] but here is the summary: 3 items merged." with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \ patch("cron.scheduler.run_job", return_value=(True, "# output", response, None)), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result") as deliver_mock, \ patch("cron.scheduler.mark_job_run"): from cron.scheduler import tick tick(verbose=False) deliver_mock.assert_called_once() def test_failed_job_always_delivers(self): """Failed jobs deliver regardless of [SILENT] in output.""" with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \ patch("cron.scheduler.run_job", return_value=(False, "# output", "", "some error")), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result") as deliver_mock, \ patch("cron.scheduler.mark_job_run"): from cron.scheduler import tick tick(verbose=False) deliver_mock.assert_called_once() def test_whitespace_only_response_is_marked_failed_not_delivered(self): """Whitespace-only final responses should behave like empty responses.""" with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \ patch("cron.scheduler.run_job", return_value=(True, "# output", " \n\t ", None)), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result") as deliver_mock, \ patch("cron.scheduler.mark_job_run") as mark_mock: from cron.scheduler import tick tick(verbose=False) deliver_mock.assert_not_called() mark_mock.assert_called_once_with( "monitor-job", False, "Agent completed but produced empty response (model error, timeout, or misconfiguration)", delivery_error=None, ) class TestOneShotDispatchClaim: """run_one_job must claim a finite one-shot's dispatch BEFORE run_job so a tick that dies mid-execution can't re-fire it forever (issue #38758).""" def _oneshot(self): return { "id": "monitor-job", "name": "monitor", "deliver": "origin", "origin": {"platform": "telegram", "chat_id": "123"}, "schedule": {"kind": "once", "run_at": "2026-01-01T00:00:00+00:00"}, "repeat": {"times": 1, "completed": 0}, } def test_claim_runs_before_run_job(self): order = [] with patch("cron.scheduler.get_due_jobs", return_value=[self._oneshot()]), \ patch("cron.scheduler.claim_dispatch", side_effect=lambda _id: order.append("claim") or True), \ patch("cron.scheduler.run_job", side_effect=lambda _j, **_kw: order.append("run") or (True, "# out", "ok", None)), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result"), \ patch("cron.scheduler.mark_job_run"): from cron.scheduler import tick tick(verbose=False) assert order == ["claim", "run"] # claim strictly before side effect class TestBuildJobPromptSilentHint: """Verify _build_job_prompt always injects [SILENT] guidance.""" def test_hint_always_present(self): job = {"prompt": "Check for updates"} result = _build_job_prompt(job) assert "[SILENT]" in result assert "Check for updates" in result class TestParseWakeGate: """Unit tests for _parse_wake_gate — pure function, no side effects.""" def test_empty_output_wakes(self): from cron.scheduler import _parse_wake_gate assert _parse_wake_gate("") is True assert _parse_wake_gate(None) is True def test_wake_gate_false_skips(self): from cron.scheduler import _parse_wake_gate assert _parse_wake_gate('{"wakeAgent": false}') is False class TestRunJobWakeGate: """Integration tests for run_job wake-gate short-circuit.""" @pytest.fixture(autouse=True) def _stub_runtime_provider(self): """Stub ``resolve_runtime_provider`` for wake-gate tests. ``run_job`` resolves the runtime provider BEFORE constructing ``AIAgent``, so these tests must mock ``resolve_runtime_provider`` in addition to ``AIAgent`` — otherwise in a hermetic CI env (no API keys), the resolver raises and the test fails before the patched AIAgent is ever reached. """ fake_runtime = { "provider": "openrouter", "api_mode": "chat_completions", "base_url": "https://openrouter.ai/api/v1", "api_key": "test-key", "source": "stub", "requested_provider": None, } with patch( "hermes_cli.runtime_provider.resolve_runtime_provider", return_value=fake_runtime, ): yield def _make_job(self, name="wake-gate-test", script="check.py"): """Minimal valid cron job dict for run_job.""" return { "id": f"job_{name}", "name": name, "prompt": "Do a thing", "schedule": "*/5 * * * *", "script": script, } def test_wake_false_skips_agent_and_returns_silent(self, caplog): """When _run_job_script output ends with {wakeAgent: false}, the agent is not invoked and run_job returns the SILENT marker so delivery is suppressed.""" from cron.scheduler import SILENT_MARKER import cron.scheduler as scheduler with patch.object(scheduler, "_run_job_script", return_value=(True, '{"wakeAgent": false}')), \ patch("run_agent.AIAgent") as agent_cls: success, doc, final, err = scheduler.run_job(self._make_job()) assert success is True assert err is None assert final == SILENT_MARKER assert "Script gate returned `wakeAgent=false`" in doc agent_cls.assert_not_called() def test_wake_true_runs_agent_with_injected_output(self): """When the script returns {wakeAgent: true, data: ...}, the agent is invoked and the data line still shows up in the prompt.""" import cron.scheduler as scheduler script_output = '{"wakeAgent": true, "data": {"new": 3}}' agent = MagicMock() agent.run_conversation = MagicMock(return_value={ "final_response": "ok", "messages": [] }) with patch.object(scheduler, "_run_job_script", return_value=(True, script_output)), \ patch("run_agent.AIAgent", return_value=agent) as agent_cls: success, doc, final, err = scheduler.run_job(self._make_job()) agent_cls.assert_called_once() # The script output should be visible in the prompt passed to # run_conversation. call_kwargs = agent.run_conversation.call_args prompt_arg = call_kwargs.args[0] if call_kwargs.args else call_kwargs.kwargs.get("user_message", "") assert script_output in prompt_arg assert success is True assert err is None class TestBuildJobPromptMissingSkill: """Verify that a missing skill logs a warning and does not crash the job.""" def _missing_skill_view(self, name: str) -> str: return json.dumps({"success": False, "error": f"Skill '{name}' not found."}) def test_missing_skill_injects_user_notice_into_prompt(self): """A system notice about the missing skill is injected into the prompt.""" with patch("tools.skills_tool.skill_view", side_effect=self._missing_skill_view): result = _build_job_prompt({"skills": ["ghost-skill"], "prompt": "do something"}) assert "ghost-skill" in result assert "not found" in result.lower() or "skipped" in result.lower() class TestBuildJobPromptAbsoluteSkillPath: """Cron jobs may store absolute skill paths; normalize before skill_view.""" def test_absolute_skill_path_normalized_before_skill_view(self, tmp_path): skills_dir = tmp_path / "skills" skill_dir = skills_dir / "alpha-skill" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text("# Alpha\nDo alpha.") absolute_path = str(skill_dir) seen_names: list[str] = [] def _skill_view(name: str) -> str: seen_names.append(name) if name == "alpha-skill": return json.dumps({"success": True, "content": "# Alpha\nDo alpha."}) return json.dumps({"success": False, "error": f"Skill '{name}' not found."}) with patch("tools.skills_tool.SKILLS_DIR", skills_dir), \ patch("tools.skills_tool.skill_view", side_effect=_skill_view): result = _build_job_prompt({"skills": [absolute_path], "prompt": "go"}) assert seen_names == ["alpha-skill"] assert "Do alpha." in result class TestBuildJobPromptBumpUse: """Verify that cron jobs bump skill usage counters so the curator sees them as active.""" def test_bump_use_called_for_loaded_skill(self): """bump_use is called for each successfully loaded skill.""" def _skill_view(name: str) -> str: return json.dumps({"success": True, "content": f"Content for {name}."}) with patch("tools.skills_tool.skill_view", side_effect=_skill_view), \ patch("tools.skill_usage.bump_use") as mock_bump: _build_job_prompt({"skills": ["alpha", "beta"], "prompt": "go"}) assert mock_bump.call_count == 2 calls = [c[0][0] for c in mock_bump.call_args_list] assert "alpha" in calls assert "beta" in calls class TestSendMediaViaAdapter: """Unit tests for _send_media_via_adapter — routes files to typed adapter methods.""" def _safe_media_path(self, tmp_path, monkeypatch, name, data=b"media"): root = tmp_path / "media-cache" media_file = root / name media_file.parent.mkdir(parents=True, exist_ok=True) media_file.write_bytes(data) monkeypatch.setattr( "gateway.platforms.base.MEDIA_DELIVERY_SAFE_ROOTS", (root,), ) return media_file.resolve() @staticmethod def _run_with_loop(adapter, chat_id, media_files, metadata, job): """Helper: run _send_media_via_adapter with immediate scheduling.""" from concurrent.futures import Future def fake_run_coro(coro, _loop): coro.close() completed = Future() completed.set_result(MagicMock(success=True)) return completed with patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro): _send_media_via_adapter(adapter, chat_id, media_files, metadata, MagicMock(), job) def test_multiple_media_files_all_delivered(self, tmp_path, monkeypatch): adapter = MagicMock() adapter.send_voice = AsyncMock() adapter.send_image_file = AsyncMock() voice_path = self._safe_media_path(tmp_path, monkeypatch, "voice.mp3") photo_path = self._safe_media_path(tmp_path, monkeypatch, "photo.jpg") media_files = [(str(voice_path), False), (str(photo_path), False)] self._run_with_loop(adapter, "123", media_files, None, {"id": "j3"}) adapter.send_voice.assert_called_once() adapter.send_image_file.assert_called_once() class TestParallelTick: """Verify that tick() runs due jobs concurrently and isolates ContextVars.""" @pytest.fixture(autouse=True) def _isolate_tick_lock(self, tmp_path): """Point the tick file lock at a per-test temp dir to avoid xdist contention.""" lock_dir = tmp_path / "cron" lock_dir.mkdir() lock_file = lock_dir / ".tick.lock" with patch("cron.scheduler._get_lock_paths", return_value=(lock_dir, lock_file)): yield def test_parallel_jobs_run_concurrently(self): """Two jobs launched in the same tick should overlap in time.""" import threading barrier = threading.Barrier(2, timeout=5) call_order = [] def mock_run_job(job, *, defer_agent_teardown=None): """Each job hits a barrier — both must be active simultaneously.""" call_order.append(("start", job["id"])) barrier.wait() # blocks until both threads reach here call_order.append(("end", job["id"])) return (True, "output", "response", None) jobs = [ {"id": "job-a", "name": "a", "deliver": "local"}, {"id": "job-b", "name": "b", "deliver": "local"}, ] with patch("cron.scheduler.get_due_jobs", return_value=jobs), \ patch("cron.scheduler.advance_next_runs"), \ patch("cron.scheduler.run_job", side_effect=mock_run_job), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result", return_value=None), \ patch("cron.scheduler.mark_job_run"): from cron.scheduler import tick result = tick(verbose=False) assert result == 2 # Both starts happened before both ends — proof of concurrency starts = [i for i, (action, _) in enumerate(call_order) if action == "start"] ends = [i for i, (action, _) in enumerate(call_order) if action == "end"] assert len(starts) == 2 assert len(ends) == 2 assert max(starts) < min(ends), f"Jobs not concurrent: {call_order}" def test_parallel_jobs_isolated_contextvars(self): """Each job's ContextVars must be isolated — no cross-contamination.""" from gateway.session_context import get_session_env seen = {} def mock_run_job(job, *, defer_agent_teardown=None): origin = job.get("origin", {}) # run_job sets ContextVars — verify each job sees its own from gateway.session_context import set_session_vars, clear_session_vars tokens = set_session_vars( platform=origin.get("platform", ""), chat_id=str(origin.get("chat_id", "")), ) import time time.sleep(0.05) # give other thread time to set its vars platform = get_session_env("HERMES_SESSION_PLATFORM") chat_id = get_session_env("HERMES_SESSION_CHAT_ID") seen[job["id"]] = {"platform": platform, "chat_id": chat_id} clear_session_vars(tokens) return (True, "output", "response", None) jobs = [ {"id": "tg-job", "name": "tg", "deliver": "local", "origin": {"platform": "telegram", "chat_id": "111"}}, {"id": "dc-job", "name": "dc", "deliver": "local", "origin": {"platform": "discord", "chat_id": "222"}}, ] with patch("cron.scheduler.get_due_jobs", return_value=jobs), \ patch("cron.scheduler.advance_next_runs"), \ patch("cron.scheduler.run_job", side_effect=mock_run_job), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result", return_value=None), \ patch("cron.scheduler.mark_job_run"): from cron.scheduler import tick tick(verbose=False) assert seen["tg-job"] == {"platform": "telegram", "chat_id": "111"} assert seen["dc-job"] == {"platform": "discord", "chat_id": "222"} class TestDeliverResultTimeoutCancelsFuture: """When future.result(timeout=60) raises TimeoutError in the live adapter delivery path, the outcome depends on whether the coroutine was already running. future.cancel() returning False means it is in flight on the wire (cannot be un-sent) → treat as DELIVERED and skip the standalone fallback to avoid a duplicate (#38922). future.cancel() returning True means it never started (wedged loop) → nothing was sent, so fall through to standalone or the message is silently dropped. Regression for #38922. """ def test_live_adapter_timeout_assumes_delivered_no_duplicate(self): """End-to-end: live adapter confirmation times out past the 60s budget. The fix (#38922) treats the send as already-dispatched/delivered and does NOT run the standalone fallback — otherwise the message is sent twice.""" from gateway.config import Platform from concurrent.futures import Future # Live adapter whose send() coroutine never resolves within the budget adapter = AsyncMock() adapter.send.return_value = MagicMock(success=True) pconfig = MagicMock() pconfig.enabled = True mock_cfg = MagicMock() mock_cfg.platforms = {Platform.TELEGRAM: pconfig} loop = MagicMock() loop.is_running.return_value = True # A real concurrent.futures.Future, but we override .result() to raise # TimeoutError exactly like the 60s wait firing in production. We make # .cancel() return False to simulate the coroutine being ALREADY RUNNING # on the gateway loop (in flight on the wire) — the case where the send # cannot be un-sent and a standalone resend would be a duplicate. captured_future = Future() cancel_calls = [] def in_flight_cancel(): cancel_calls.append(True) return False # already running — cannot be cancelled captured_future.cancel = in_flight_cancel captured_future.result = MagicMock(side_effect=TimeoutError("timed out")) def fake_run_coro(coro, _loop): coro.close() return captured_future job = { "id": "timeout-job", "deliver": "origin", "origin": {"platform": "telegram", "chat_id": "123"}, } standalone_send = AsyncMock(return_value={"success": True}) with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro), \ patch("tools.send_message_tool._send_to_platform", new=standalone_send): result = _deliver_result( job, "Hello world", adapters={Platform.TELEGRAM: adapter}, loop=loop, ) # 1. cancel() was attempted (returned False = in flight). assert cancel_calls == [True], "future.cancel() should be attempted on TimeoutError" # 2. Delivery is reported successful (no error string returned). assert result is None, f"expected successful delivery, got error: {result!r}" # 3. The standalone fallback must NOT run — that is the #38922 fix: # an in-flight confirmation timeout is assume-delivered, not a resend. standalone_send.assert_not_awaited() class TestDeliverResultLiveAdapterUnconfirmed: """Regression for #47056. When a live adapter's send() returns ``None`` (swallowed exception / busy platform) or a result object that lacks an explicit ``success`` attribute (bare dict / partial object), the scheduler must NOT log "delivered via live adapter" and silently drop the message. Every unconfirmed shape must fall through to the standalone delivery path so the message actually arrives. The pre-fix check ``send_result is None or not getattr(..., "success", True)`` let a ``.success``-less object default to True = silent success. """ def _run(self, send_value): from gateway.config import Platform from concurrent.futures import Future adapter = AsyncMock() adapter.send.return_value = send_value pconfig = MagicMock() pconfig.enabled = True mock_cfg = MagicMock() mock_cfg.platforms = {Platform.TELEGRAM: pconfig} loop = MagicMock() loop.is_running.return_value = True completed_future = Future() completed_future.set_result(send_value) def fake_run_coro(coro, _loop): coro.close() return completed_future job = { "id": "unconfirmed-job", "deliver": "origin", "origin": {"platform": "telegram", "chat_id": "123"}, } standalone_send = AsyncMock(return_value={"success": True}) with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro), \ patch("tools.send_message_tool._send_to_platform", new=standalone_send): result = _deliver_result( job, "Hello world", adapters={Platform.TELEGRAM: adapter}, loop=loop, ) return result, standalone_send def test_none_result_falls_through_to_standalone(self): """send() returning None must trigger the standalone fallback, not a silent "delivered" log.""" result, standalone_send = self._run(None) assert result is None, f"standalone should have delivered, got: {result!r}" standalone_send.assert_awaited_once() class TestDeliverOriginUnresolvableIsLocal: """Regression for #43014. A cron job created in a CLI session has no {platform, chat_id} origin. With ``deliver=origin`` (or auto-detect / deliver=None) and no configured platform home channel, delivery is unresolvable — but that is the EXPECTED state for CLI jobs, not an error. _deliver_result must return None (treat as local; output stays in last_output), not the "no delivery target resolved" error string that previously fired on every run. """ def _deliver(self, job, monkeypatch): import cron.scheduler as sched # No home channel for any platform → origin is unresolvable. monkeypatch.setattr(sched, "_get_home_target_chat_id", lambda *_: "") return _deliver_result(job, "CLI bulletin") def test_origin_with_no_home_channels_returns_none(self, monkeypatch): job = {"id": "cli-job", "deliver": "origin", "origin": "cli-session-provenance"} assert self._deliver(job, monkeypatch) is None class TestSendMediaTimeoutCancelsFuture: """Same orphan-coroutine guarantee for _send_media_via_adapter's future.result(timeout=30) call. If this times out mid-batch, the in-flight coroutine must be cancelled before the next file is tried. """ def test_media_send_timeout_cancels_future_and_continues(self, tmp_path, monkeypatch): """End-to-end: _send_media_via_adapter with a future whose .result() raises TimeoutError. Assert cancel() fires and the loop proceeds to the next file rather than hanging or crashing.""" from concurrent.futures import Future adapter = MagicMock() adapter.send_image_file = AsyncMock() adapter.send_video = AsyncMock() # First file: future that times out. Second file: future that resolves OK. timeout_future = Future() timeout_cancel_calls = [] original_cancel = timeout_future.cancel def tracking_cancel(): timeout_cancel_calls.append(True) return original_cancel() timeout_future.cancel = tracking_cancel timeout_future.result = MagicMock(side_effect=TimeoutError("timed out")) ok_future = Future() ok_future.set_result(MagicMock(success=True)) futures_iter = iter([timeout_future, ok_future]) def fake_run_coro(coro, _loop): coro.close() return next(futures_iter) root = tmp_path / "media-cache" slow = root / "slow.png" fast = root / "fast.mp4" slow.parent.mkdir(parents=True) slow.write_bytes(b"slow") fast.write_bytes(b"fast") monkeypatch.setattr( "gateway.platforms.base.MEDIA_DELIVERY_SAFE_ROOTS", (root,), ) media_files = [ (str(slow), False), # times out (str(fast), False), # succeeds ] loop = MagicMock() job = {"id": "media-timeout"} with patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro): # Should not raise — the except Exception clause swallows the timeout _send_media_via_adapter(adapter, "chat-1", media_files, None, loop, job) # 1. The timed-out future was cancelled (the bug fix) assert timeout_cancel_calls == [True], "future.cancel() must fire on TimeoutError" # 2. Second file still got dispatched — one timeout doesn't abort the batch adapter.send_video.assert_called_once() assert adapter.send_video.call_args[1]["video_path"] == str(fast.resolve()) class TestCronDeliveryTargets: """``cron_delivery_targets`` powers the dashboard delivery dropdown. It must list every configured + cron-deliverable platform (no hardcoded set), flag whether each has its home channel set, and never include platforms whose gateway isn't configured. """ def _patch_connected(self, monkeypatch, names): import gateway.config as gateway_config class _Platform: def __init__(self, value): self.value = value class _GatewayConfig: def get_connected_platforms(self_inner): return [_Platform(n) for n in names] monkeypatch.setattr( gateway_config, "load_gateway_config", lambda: _GatewayConfig() ) def test_lists_configured_platforms_flagging_missing_home_channel(self, monkeypatch): from cron.scheduler import cron_delivery_targets self._patch_connected(monkeypatch, ["matrix", "telegram"]) monkeypatch.delenv("MATRIX_HOME_ROOM", raising=False) monkeypatch.delenv("TELEGRAM_HOME_CHANNEL", raising=False) targets = {t["id"]: t for t in cron_delivery_targets()} assert set(targets) == {"matrix", "telegram"} # Configured but no home channel → surfaced, flagged for the UI. assert targets["matrix"]["home_target_set"] is False assert targets["matrix"]["home_env_var"] == "MATRIX_HOME_ROOM" assert targets["telegram"]["home_target_set"] is False class TestHomeTargetEnvVarRegistry: """Regression: ``_HOME_TARGET_ENV_VARS`` must include every gateway platform that supports cron-driven outbound delivery. Missing an entry means ``hermes cron create --deliver=`` silently fails to route through the platform's home channel.""" class TestCronDeliveryMirror: """cron.mirror_delivery / per-job attach_to_session: opt-in append of a cron delivery into the target chat's gateway session transcript. Default OFF preserves the historical isolation guarantee byte-for-byte. When enabled, delivery rides the existing gateway.mirror.mirror_to_session so cron uses exactly the same path interactive send_message mirroring uses. """ def test_mirror_writes_user_role_with_label_not_assistant(self): """Regression for #2221 / #2313: the cron brief must mirror as a USER turn (with a [Cron delivery: ...] label), NOT assistant — an assistant-role mirror lands as assistant->assistant after the agent's last turn and breaks strict alternation on non-Anthropic providers.""" from cron.scheduler import _maybe_mirror_cron_delivery with patch("gateway.mirror.mirror_to_session", return_value=True) as m: _maybe_mirror_cron_delivery( {"id": "j1", "name": "Morning Brief"}, "telegram", "123", "Market movers today", thread_id=None, enabled=True, ) m.assert_called_once() args, kwargs = m.call_args assert kwargs.get("role") == "user", "cron mirror must be a user turn, not assistant" # The brief text is prefixed with a human-readable cron-delivery label # so replay (where the mirror metadata is dropped at the SQLite # boundary) still distinguishes it from a genuine user message. assert args[2].startswith("[Cron delivery: Morning Brief]") assert "Market movers today" in args[2] def test_delivery_mirrors_clean_content_not_wrapped(self): """When enabled, the mirror receives the CLEAN agent output, not the cron header/footer-wrapped delivery text.""" from gateway.config import Platform pconfig = MagicMock() pconfig.enabled = True mock_cfg = MagicMock() mock_cfg.platforms = {Platform.TELEGRAM: pconfig} with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})), \ patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: job = { "id": "test-job", "name": "daily-report", "deliver": "origin", "origin": {"platform": "telegram", "chat_id": "123"}, "attach_to_session": True, } _deliver_result(job, "Here is today's summary.") mirror_mock.assert_called_once() mirrored_text = mirror_mock.call_args[0][2] # Clean content, no cron wrapper. assert "Here is today's summary." in mirrored_text assert "Cronjob Response:" not in mirrored_text assert "To stop or manage this job" not in mirrored_text # --- origin-scoping (mirror only into the conversation that created the job) --- # --- multi-participant parity with send_message (user_id passthrough) --- # --- continuable cron: thread-preferred (Teknium's interface) --- def test_open_thread_returns_id_on_thread_platform(self): """On a thread-capable adapter, _open_continuable_cron_thread returns the new thread id from create_handoff_thread.""" from cron.scheduler import _open_continuable_cron_thread adapter = MagicMock() adapter.create_handoff_thread = AsyncMock(return_value="9001") # safe_schedule_threadsafe hands the coro to the gateway loop and # returns a future. Patch it to close the coro and return a ready # future carrying the adapter's thread id. def _run_now(coro, _loop): coro.close() fut = MagicMock() fut.result.return_value = "9001" return fut with patch("agent.async_utils.safe_schedule_threadsafe", side_effect=_run_now): tid = _open_continuable_cron_thread( {"id": "j1", "name": "Brief"}, adapter, "123", loop=MagicMock(), ) assert tid == "9001" def test_seed_thread_session_creates_session_and_mirrors(self): """Seeding a freshly-opened thread creates the thread-keyed session via the adapter's live store and appends the brief via mirror_to_session.""" from cron.scheduler import _seed_cron_thread_session store = MagicMock() adapter = MagicMock() adapter._session_store = store with patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: _seed_cron_thread_session( {"id": "j1"}, adapter, "telegram", "123", "9001", "Daily brief Task #2", chat_name="Ops", ) # Session row created for the thread, then brief mirrored into it. store.get_or_create_session.assert_called_once() seeded_source = store.get_or_create_session.call_args[0][0] assert seeded_source.chat_type == "thread" assert seeded_source.thread_id == "9001" mirror_mock.assert_called_once() assert mirror_mock.call_args.kwargs.get("thread_id") == "9001" class TestCronContinuableSurfaceInChannel: """cron_continuable_surface: in_channel — deliver a continuable cron FLAT into a channel (no dedicated thread), so a plain channel reply continues the job via the shared-channel session (platform, chat_id, None). Design: decisions.md D1/D2/D6 + F5. The scheduler reads the per-platform key generically from pconfig.extra; the in_channel branch is gated on the adapter capability flag ``supports_inchannel_continuable`` (Slack=True, others fail SAFE to thread). In in_channel mode the thread-open branch is SKIPPED (thread_id stays None), then ``_seed_cron_channel_session`` CREATES the flat shared-channel session and mirrors the brief into it (the shipped mirror only APPENDS to an existing session, and the flat channel row is otherwise absent for a chat_postMessage delivery). """ def _slack_cfg(self, extra): """A mock GatewayConfig with a Slack pconfig carrying ``extra``.""" from gateway.config import Platform pconfig = MagicMock() pconfig.enabled = True pconfig.extra = extra mock_cfg = MagicMock() mock_cfg.platforms = {Platform.SLACK: pconfig} return mock_cfg def _run_inchannel_delivery(self, extra, adapter, *, mirror_ok=True, origin=None): """Drive _deliver_result down the live-adapter path for a Slack channel-origin job with the given ``extra`` config. Returns the _open_continuable_cron_thread mock and the mirror_to_session mock.""" from gateway.config import Platform from concurrent.futures import Future mock_cfg = self._slack_cfg(extra) loop = MagicMock() loop.is_running.return_value = True def fake_run_coro(coro, _loop): future = Future() try: import asyncio as _asyncio future.set_result(_asyncio.run(coro)) except BaseException as _e: # noqa: BLE001 future.set_exception(_e) return future job = { "id": "brief-job", "name": "Daily Brief", "deliver": "origin", # Channel origin: no thread_id (flat channel message scheduled it). # Carries the scheduling user's id — the in_channel seed must key # the flat channel session to THIS user (see build_session_key). "origin": origin or {"platform": "slack", "chat_id": "C123", "user_id": "U_HUMAN"}, # Opt into the continuable mirror. "attach_to_session": True, } with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ patch("cron.scheduler._open_continuable_cron_thread") as open_thread_mock, \ patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro), \ patch("gateway.mirror.mirror_to_session", return_value=mirror_ok) as mirror_mock: _deliver_result( job, "Here is today's brief.", adapters={Platform.SLACK: adapter}, loop=loop, ) return open_thread_mock, mirror_mock def _slack_adapter(self, supports_inchannel=True, with_store=True): adapter = AsyncMock() adapter.send.return_value = MagicMock(success=True) # Capability flag read via getattr in the scheduler. adapter.supports_inchannel_continuable = supports_inchannel # A live session store so the in_channel seed can CREATE the flat row # (the real bug: without a create step the mirror no-ops on a missing # session and the brief is lost). Use a plain MagicMock store. if with_store: adapter._session_store = MagicMock() return adapter def test_in_channel_skips_thread_open(self): """G2: in_channel mode must NOT open a handoff thread.""" adapter = self._slack_adapter(supports_inchannel=True) open_thread_mock, _ = self._run_inchannel_delivery( {"cron_continuable_surface": "in_channel"}, adapter, ) open_thread_mock.assert_not_called() # --- _seed_cron_channel_session: the create-then-mirror unit + the # KEY-MATCH invariant (seed key must equal the inbound reply's key) --- def test_seed_channel_session_key_matches_inbound_channel_reply(self): """The whole point: the flat session the seed CREATES must be keyed identically to what a plain inbound channel reply resolves to. Assert the invariant directly via build_session_key, not just call args.""" from cron.scheduler import _seed_cron_channel_session from gateway.session import build_session_key, SessionSource from gateway.config import Platform store = MagicMock() adapter = MagicMock() adapter._session_store = store with patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock: ok = _seed_cron_channel_session( {"id": "j1", "name": "Brief"}, adapter, "slack", "C123", "Daily brief", is_dm=False, user_id="U_HUMAN", chat_name="ops", ) assert ok is True seeded_source = store.get_or_create_session.call_args[0][0] seed_key = build_session_key(seeded_source) # What a plain top-level channel reply (reply_in_thread:false → thread # None) from the same user resolves to: inbound = SessionSource( platform=Platform.SLACK, chat_id="C123", chat_type="group", user_id="U_HUMAN", thread_id=None, ) assert seed_key == build_session_key(inbound), ( f"seed key {seed_key} != inbound reply key {build_session_key(inbound)} " "— the reply would NOT continue the seeded session" ) mirror_mock.assert_called_once() assert mirror_mock.call_args.kwargs.get("thread_id") is None assert mirror_mock.call_args.kwargs.get("user_id") == "U_HUMAN" class TestMultiTargetDeliveryContinuesOnFailure: """When delivery to one target fails inside the standalone thread-pool fallback, the loop must continue to the remaining targets (#47163). The fallback runs inside the `except RuntimeError` block of `_deliver_result`. Before the fix, an exception raised there (SMTP ConnectionError, future.result timeout) escaped the function entirely — it is NOT caught by the sibling `except Exception` — crashing the loop and silently dropping every subsequent target. """ def _email_cfg(self): from gateway.config import Platform pconfig = MagicMock() pconfig.enabled = True mock_cfg = MagicMock() mock_cfg.platforms = {Platform.EMAIL: pconfig} return mock_cfg def test_first_target_failure_does_not_crash_loop(self): """First email target fails in the fallback; the second is still attempted.""" job = { "id": "multi-email-job", "deliver": "email:a@example.com,email:b@example.com", } with patch("gateway.config.load_gateway_config", return_value=self._email_cfg()), \ patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ patch("asyncio.run", side_effect=RuntimeError("no running loop")), \ patch("concurrent.futures.ThreadPoolExecutor") as mock_pool_cls: mock_pool = MagicMock() mock_pool_cls.return_value = mock_pool fail_future = MagicMock() fail_future.result.side_effect = ConnectionError("SMTP connection refused") ok_future = MagicMock() ok_future.result.return_value = {"success": True} mock_pool.submit.side_effect = [fail_future, ok_future] result = _deliver_result(job, "Report content") # Both targets attempted — the loop did not crash after the first failure. assert mock_pool.submit.call_count == 2, ( f"expected 2 delivery attempts, got {mock_pool.submit.call_count}" ) # First target's failure is surfaced in the returned error string. assert result is not None assert "a@example.com" in result assert "SMTP connection refused" in result def test_all_targets_fail_returns_combined_errors(self): """When every target fails, the result reports all of them.""" job = { "id": "all-fail-job", "deliver": "email:a@example.com,email:b@example.com", } with patch("gateway.config.load_gateway_config", return_value=self._email_cfg()), \ patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ patch("asyncio.run", side_effect=RuntimeError("no running loop")), \ patch("concurrent.futures.ThreadPoolExecutor") as mock_pool_cls: mock_pool = MagicMock() mock_pool_cls.return_value = mock_pool fail_future = MagicMock() fail_future.result.side_effect = ConnectionError("connection refused") mock_pool.submit.return_value = fail_future result = _deliver_result(job, "Report content") assert result is not None assert "a@example.com" in result assert "b@example.com" in result assert mock_pool.submit.call_count == 2 class TestSetCronSessionTitle: """Robust cron session titling: #50535/#50536/#50537.""" def test_dedupes_on_duplicate_title(self): # First write collides (ValueError); helper falls back to lineage #N. from cron.scheduler import _set_cron_session_title db = MagicMock() db.set_session_title.side_effect = [ValueError("in use"), True] db.get_next_title_in_lineage.return_value = "Nightly Synthesis #2" out = _set_cron_session_title(db, "sess-1", "Nightly Synthesis") assert out == "Nightly Synthesis #2" db.get_next_title_in_lineage.assert_called_once_with("Nightly Synthesis")