"""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 ( SILENT_MARKER, _build_job_prompt, _deliver_result, _merge_mcp_into_per_job_toolsets, _resolve_cron_enabled_toolsets, _resolve_delivery_target, _resolve_origin, _send_media_via_adapter, _summarize_cron_failure_for_delivery, run_job, ) from tools.env_passthrough import clear_env_passthrough from tools.credential_files import clear_credential_files class TestSummarizeCronFailureForDelivery: def test_embedded_429_in_source_identifier_is_not_a_rate_limit(self): summary = _summarize_cron_failure_for_delivery( {"name": "LLM Wiki Incremental Index", "no_agent": True}, "Script failed: path/hash429abc.md source snapshot failure", ) assert "provider rate limit" not in summary assert "hash429abc.md" in summary def test_http_429_is_still_classified_as_a_rate_limit(self): summary = _summarize_cron_failure_for_delivery( {"name": "provider-backed job"}, "HTTP 429: Too Many Requests", ) assert "provider rate limit" in summary # Chain wording is now honest (#85508): either the exhausted phrase # (chain configured) or the "No fallback chain configured" guidance. assert "fallback chain" in summary.lower() def test_no_agent_rate_limit_does_not_claim_a_fallback_chain(self): summary = _summarize_cron_failure_for_delivery( {"name": "script job", "no_agent": True}, "HTTP 429: Too Many Requests", ) # Composed with #77648: a no_agent job never gets provider-shaped # classification at all — the generic cleaner reports the script's # own error instead. assert "provider" not in summary.lower() assert "fallback chain" not in summary.lower() def test_no_agent_timeout_is_identified_as_a_script_timeout(self): summary = _summarize_cron_failure_for_delivery( {"name": "script job", "no_agent": True}, "Script timed out after 3600s", ) assert "script timed out" in summary assert "No model was invoked" in summary assert "provider timeout" not in summary assert "fallback chain" not in summary.lower() 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) def test_resolver_keeps_memory_in_per_job_list(self): result = _resolve_cron_enabled_toolsets( {"enabled_toolsets": ["memory", "file"]}, {"mcp_servers": {}}, ) assert "memory" in result assert "file" in result def test_resolver_keeps_memory_from_platform_fallback(self): job = {"enabled_toolsets": None} with patch( "hermes_cli.tools_config._get_platform_tools", return_value={"web", "memory", "file"}, ): result = _resolve_cron_enabled_toolsets(job, {}) assert result == ["file", "memory", "web"] 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_unresolved_target_still_delivered_as_written(self): """A stored job's platform-native target keeps delivering when neither parser nor directory recognizes it. Routing cron through resolve_send_target turned these into a warning plus a silently dropped delivery; pass_unresolved_references hands the raw id to the adapter again.""" job = {"deliver": "telegram:ops-room"} with patch( "gateway.channel_directory.resolve_channel_name", return_value=None, ): result = _resolve_delivery_target(job) assert result == { "platform": "telegram", "chat_id": "ops-room", "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 # Not a real RelayAdapter: keep the auto-created accessor from # shadowing the scalar False (MagicMock fabricates truthy callables). relay.supports_inchannel_continuable_for_platform = None 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_run_job_memory_enabled_in_cron(self, tmp_path): """Cron agents get memory like any other agent run. skip_memory=False and the memory toolset is not policy-denied, so MEMORY.md/USER.md load and the memory tool follows normal toolset resolution. """ job = { "id": "memory-enabled-job", "name": "test", "prompt": "hello", } with self._run_job_patches(tmp_path) as (fake_db, mock_agent_cls): run_job(job) kwargs = mock_agent_cls.call_args.kwargs assert kwargs["skip_memory"] is False assert "memory" not in (kwargs["disabled_toolsets"] or []), ( "memory toolset must not be policy-denied in cron" ) def test_run_job_keeps_per_job_memory_toolset(self, tmp_path): """A per-job enabled_toolsets naming memory keeps it.""" job = { "id": "memory-toolset-job", "name": "test", "prompt": "remember what you learn", "enabled_toolsets": ["memory", "file"], } with self._run_job_patches(tmp_path) as (fake_db, mock_agent_cls): run_job(job) kwargs = mock_agent_cls.call_args.kwargs assert kwargs["skip_memory"] is False assert "memory" in (kwargs["enabled_toolsets"] or []) assert "file" in (kwargs["enabled_toolsets"] or []) assert "memory" not in kwargs["disabled_toolsets"] 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.claim_job_for_fire", return_value=True ) as claim, patch("cron.scheduler.run_one_job") as run_one: assert tick(verbose=False, sync=True, can_dispatch=lambda: False) == 0 claim.assert_not_called() run_one.assert_not_called() def test_tick_marks_empty_response_as_error(self, tmp_path): """When run_job returns success=True but final_response is empty, tick() should mark the job as error so last_status != 'ok'. (issue #8585) """ from cron.scheduler import tick job = { "id": "empty-job", "name": "empty-test", "prompt": "do something", "schedule": "every 1h", "enabled": True, "next_run_at": "2020-01-01T00:00:00", "deliver": "local", "last_status": None, } fake_db = MagicMock() with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler.get_due_jobs", return_value=[job]), \ patch("cron.scheduler.claim_job_for_fire", return_value=True), \ patch("cron.scheduler.mark_job_run") as mock_mark, \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("cron.scheduler.run_job", return_value=(True, "output", "", None)): tick(verbose=False) # Should be called with success=False because final_response is empty mock_mark.assert_called_once() call_args = mock_mark.call_args assert call_args[0][0] == "empty-job" assert call_args[0][1] is False # success should be False assert "empty" in call_args[0][2].lower() # error should mention empty def test_run_job_sets_auto_delivery_env_from_dotenv_home_channel(self, tmp_path, monkeypatch): job = { "id": "test-job", "name": "test", "prompt": "hello", "deliver": "telegram", } fake_db = MagicMock() seen = {} (tmp_path / ".env").write_text("TELEGRAM_HOME_CHANNEL=-2002\n") monkeypatch.delenv("TELEGRAM_HOME_CHANNEL", raising=False) monkeypatch.delenv("HERMES_CRON_AUTO_DELIVER_PLATFORM", raising=False) monkeypatch.delenv("HERMES_CRON_AUTO_DELIVER_CHAT_ID", raising=False) monkeypatch.delenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID", raising=False) class FakeAgent: def __init__(self, *args, **kwargs): pass def run_conversation(self, *args, **kwargs): from gateway.session_context import get_session_env seen["platform"] = get_session_env("HERMES_CRON_AUTO_DELIVER_PLATFORM") or None seen["chat_id"] = get_session_env("HERMES_CRON_AUTO_DELIVER_CHAT_ID") or None seen["thread_id"] = get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID") or None return {"final_response": "ok"} with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._preflight_job_config", return_value=None), \ 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("run_agent.AIAgent", FakeAgent): success, output, final_response, error = run_job(job) assert success is True assert error is None assert final_response == "ok" assert "ok" in output assert seen == { "platform": "telegram", "chat_id": "-2002", "thread_id": None, } assert os.getenv("HERMES_CRON_AUTO_DELIVER_PLATFORM") is None assert os.getenv("HERMES_CRON_AUTO_DELIVER_CHAT_ID") is None assert os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID") is None fake_db.close.assert_called_once() def test_run_job_preserves_slack_origin_thread_for_same_explicit_channel(self, tmp_path, monkeypatch): job = { "id": "slack-thread-job", "name": "slack-thread", "prompt": "hello", "deliver": "slack:C0B3KEP3SD6", "origin": { "platform": "slack", "chat_id": "C0B3KEP3SD6", "thread_id": "1778485067.844139", }, } fake_db = MagicMock() seen = {} monkeypatch.delenv("HERMES_CRON_AUTO_DELIVER_PLATFORM", raising=False) monkeypatch.delenv("HERMES_CRON_AUTO_DELIVER_CHAT_ID", raising=False) monkeypatch.delenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID", raising=False) class FakeAgent: def __init__(self, *args, **kwargs): pass def run_conversation(self, *args, **kwargs): from gateway.session_context import get_session_env seen["platform"] = get_session_env("HERMES_CRON_AUTO_DELIVER_PLATFORM") or None seen["chat_id"] = get_session_env("HERMES_CRON_AUTO_DELIVER_CHAT_ID") or None seen["thread_id"] = get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID") or None return {"final_response": "ok"} with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._preflight_job_config", return_value=None), \ 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("run_agent.AIAgent", FakeAgent): success, output, final_response, error = run_job(job) assert success is True assert error is None assert final_response == "ok" assert "ok" in output assert seen == { "platform": "slack", "chat_id": "C0B3KEP3SD6", "thread_id": "1778485067.844139", } assert os.getenv("HERMES_CRON_AUTO_DELIVER_PLATFORM") is None assert os.getenv("HERMES_CRON_AUTO_DELIVER_CHAT_ID") is None assert os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID") is None fake_db.close.assert_called_once() @pytest.mark.parametrize("timeout_value", ["600", "0"]) def test_run_job_heartbeats_oneshot_claim_in_both_wait_modes( self, tmp_path, monkeypatch, timeout_value ): """Timed and unlimited one-shot monitors both refresh their owned claim.""" job = { "id": "heartbeat-job", "name": "heartbeat", "prompt": "hello", "schedule": {"kind": "once", "run_at": "2026-07-10T12:00:00Z"}, "run_claim": {"at": "2026-07-10T12:00:00Z", "by": "owner-token"}, } fake_db = MagicMock() class FakeAgent: def __init__(self, *args, **kwargs): pass def run_conversation(self, *args, **kwargs): return {"final_response": "ok"} class FakeFuture: def result(self): return {"final_response": "ok"} fake_future = FakeFuture() fake_pool = MagicMock() fake_pool.submit.return_value = fake_future wait_results = [(set(), set()), ({fake_future}, set())] monotonic_ticks = itertools.count(step=61.0) monkeypatch.setenv("HERMES_CRON_TIMEOUT", timeout_value) with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._preflight_job_config", return_value=None), \ 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("run_agent.AIAgent", FakeAgent), \ patch("cron.scheduler.concurrent.futures.ThreadPoolExecutor", return_value=fake_pool), \ patch("cron.scheduler.concurrent.futures.wait", side_effect=wait_results), \ patch("cron.scheduler.time.monotonic", side_effect=monotonic_ticks.__next__), \ patch("cron.scheduler.heartbeat_run_claim", return_value=True) as heartbeat: success, _output, final_response, error = run_job(job) assert success is True assert error is None assert final_response == "ok" heartbeat.assert_called_once_with( "heartbeat-job", expected_owner="owner-token" ) def test_run_job_resets_secret_source_cache_before_reload(self, tmp_path, monkeypatch): """Each run must clear the secret-source cache before re-reading the env, so a long-running gateway re-resolves Bitwarden/BSM-backed secrets instead of leaving the startup .env placeholder in place (#33465). A bare ``load_dotenv`` re-load can't do this: startup already recorded this HERMES_HOME in ``_APPLIED_HOMES``, so the external-secret pull no-ops and only the placeholder is re-applied. The scheduler must call ``reset_secret_source_cache()`` (forcing the re-pull) and route through ``load_hermes_dotenv`` (which then re-applies external secret sources). """ job = {"id": "bsm-job", "name": "bsm", "prompt": "hello"} fake_db = MagicMock() call_order = [] def _record_reset(): call_order.append("reset") def _record_load(*args, **kwargs): call_order.append("load") return [] with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.reset_secret_source_cache", _record_reset), \ patch("hermes_cli.env_loader.load_hermes_dotenv", _record_load), \ 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("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, error = run_job(job) assert success is True assert error is None # reset MUST precede the reload, else _APPLIED_HOMES no-ops the re-pull. assert call_order[:2] == ["reset", "load"], call_order def test_run_job_clears_stale_auto_delivery_thread_id_between_jobs(self, tmp_path, monkeypatch): jobs = [ { "id": "threaded-job", "name": "threaded", "prompt": "hello", "deliver": "telegram:-1001:42", }, { "id": "threadless-job", "name": "threadless", "prompt": "hello again", "deliver": "telegram:-2002", }, ] fake_db = MagicMock() seen = [] monkeypatch.delenv("HERMES_CRON_AUTO_DELIVER_PLATFORM", raising=False) monkeypatch.delenv("HERMES_CRON_AUTO_DELIVER_CHAT_ID", raising=False) monkeypatch.delenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID", raising=False) class FakeAgent: def __init__(self, *args, **kwargs): pass def run_conversation(self, *args, **kwargs): from gateway.session_context import get_session_env seen.append( { "platform": get_session_env("HERMES_CRON_AUTO_DELIVER_PLATFORM") or None, "chat_id": get_session_env("HERMES_CRON_AUTO_DELIVER_CHAT_ID") or None, "thread_id": get_session_env("HERMES_CRON_AUTO_DELIVER_THREAD_ID") or None, } ) return {"final_response": "ok"} with patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler._preflight_job_config", return_value=None), \ 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("run_agent.AIAgent", FakeAgent): for job in jobs: success, output, final_response, error = run_job(job) assert success is True assert error is None assert final_response == "ok" assert "ok" in output assert seen == [ { "platform": "telegram", "chat_id": "-1001", "thread_id": "42", }, { "platform": "telegram", "chat_id": "-2002", "thread_id": None, }, ] assert os.getenv("HERMES_CRON_AUTO_DELIVER_PLATFORM") is None assert os.getenv("HERMES_CRON_AUTO_DELIVER_CHAT_ID") is None assert os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID") is None assert fake_db.close.call_count == 2 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_transient_dns_fallback_switches_provider_and_model_together(self, tmp_path): """DNS blip during primary OAuth resolve must still walk fallback_providers. Regression for Daily Focus Kickoff 2026-08-11: xai-oauth token refresh raised httpx.ConnectError ([Errno 8] nodename nor servname provided) and the scheduler only tried fallbacks on AuthError, so the job died before XAI_API_KEY / Anthropic could rescue it. """ import httpx (tmp_path / "config.yaml").write_text( "model:\n" " default: grok-4.5\n" " provider: xai-oauth\n" "fallback_providers:\n" " - provider: xai\n" " model: grok-4.5\n" " - provider: anthropic\n" " model: claude-opus-5\n", encoding="utf-8", ) job = { "id": "dns-fallback", "name": "dns fallback", "prompt": "hi", "provider": "xai-oauth", "model": "grok-4.5", } fake_db = MagicMock() requested = [] def resolve_runtime(**kwargs): requested.append(kwargs.get("requested")) if kwargs.get("requested") in (None, "xai-oauth"): raise httpx.ConnectError( "[Errno 8] nodename nor servname provided, or not known" ) # First fallback rung (xai API key) succeeds. assert kwargs["requested"] == "xai" assert kwargs["target_model"] == "grok-4.5" return {**self._RUNTIME, "provider": "xai", "api_mode": "chat_completions"} 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, error assert error is None assert requested == ["xai-oauth", "xai"] kwargs = mock_agent_cls.call_args.kwargs assert kwargs["provider"] == "xai" assert kwargs["model"] == "grok-4.5" 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.claim_job_for_fire", return_value=True), \ 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_silent_with_note_suppresses_delivery(self): with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \ patch("cron.scheduler.claim_job_for_fire", return_value=True), \ patch("cron.scheduler.run_job", return_value=(True, "# output", "[SILENT] No changes detected", 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_not_called() def test_silent_trailing_suppresses_delivery(self): """Agent appended [SILENT] after explanation text — must still suppress.""" response = "2 deals filtered out (like<10, reply<15).\n\n[SILENT]" with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \ patch("cron.scheduler.claim_job_for_fire", return_value=True), \ 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_not_called() def test_silent_is_case_insensitive(self): with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \ patch("cron.scheduler.claim_job_for_fire", return_value=True), \ patch("cron.scheduler.run_job", return_value=(True, "# output", "[silent] nothing new", 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_not_called() def test_bracketless_silent_variants_suppress(self): """Bracketless near-markers the model emits when it drops brackets must still suppress delivery (#51438, #46917).""" from cron.scheduler import tick for marker in ("SILENT", "NO_REPLY", "NO REPLY", "no_reply"): with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \ patch("cron.scheduler.claim_job_for_fire", return_value=True), \ patch("cron.scheduler.run_job", return_value=(True, "# output", marker, 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"): tick(verbose=False) deliver_mock.assert_not_called() 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.claim_job_for_fire", return_value=True), \ 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.claim_job_for_fire", return_value=True), \ 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_output_saved_even_when_delivery_suppressed(self): with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \ patch("cron.scheduler.claim_job_for_fire", return_value=True), \ patch("cron.scheduler.run_job", return_value=(True, "# full output", "[SILENT]", None)), \ patch("cron.scheduler.save_job_output") as save_mock, \ patch("cron.scheduler._deliver_result") as deliver_mock, \ patch("cron.scheduler.mark_job_run"): save_mock.return_value = "/tmp/out.md" from cron.scheduler import tick tick(verbose=False) save_mock.assert_called_once_with("monitor-job", "# full output") deliver_mock.assert_not_called() 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.claim_job_for_fire", return_value=True), \ 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_job_for_fire", return_value=True), \ 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 def test_refused_claim_skips_run_job(self): with patch("cron.scheduler.get_due_jobs", return_value=[self._oneshot()]), \ patch("cron.scheduler.claim_job_for_fire", return_value=True), \ patch("cron.scheduler.claim_dispatch", return_value=False), \ patch("cron.scheduler.run_job") as run_mock, \ patch("cron.scheduler.save_job_output"), \ 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) run_mock.assert_not_called() deliver_mock.assert_not_called() mark_mock.assert_not_called() 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({ "id": "cron-task", "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 assert all( call.kwargs == {"task_id": "cron-task"} for call in mock_bump.call_args_list ) 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, **kw): """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.claim_job_for_fire", return_value=True), \ 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, **kw): 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.claim_job_for_fire", return_value=True), \ 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"} def test_max_parallel_env_var(self, monkeypatch): """HERMES_CRON_MAX_PARALLEL=1 should restore serial behaviour.""" monkeypatch.setenv("HERMES_CRON_MAX_PARALLEL", "1") call_times = [] def mock_run_job(job, *, defer_agent_teardown=None, **_kw): import time call_times.append(("start", job["id"], time.monotonic())) time.sleep(0.05) call_times.append(("end", job["id"], time.monotonic())) return (True, "output", "response", None) jobs = [ {"id": "s1", "name": "s1", "deliver": "local"}, {"id": "s2", "name": "s2", "deliver": "local"}, ] with patch("cron.scheduler.get_due_jobs", return_value=jobs), \ patch("cron.scheduler.claim_job_for_fire", return_value=True), \ 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 # With max_workers=1, second job starts after first ends end_s1 = [t for action, jid, t in call_times if action == "end" and jid == "s1"][0] start_s2 = [t for action, jid, t in call_times if action == "start" and jid == "s2"][0] assert start_s2 >= end_s1, "Jobs ran concurrently despite max_parallel=1" 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, attach_to_session=True): """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 (parameterized: the seed must # NOT depend on this — see the no-attach regression test). "attach_to_session": attach_to_session, } 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, message_id="msg_1", raw_response=None, ) # Capability flag read via getattr in the scheduler. adapter.supports_inchannel_continuable = supports_inchannel # Pin the per-platform accessor OFF: an unspecced AsyncMock would # auto-create it as a truthy callable, silently routing every test # through the relay accessor branch instead of the native scalar # fallback these tests describe (and making supports_inchannel=False # unenforceable). None -> not callable -> scalar path, like a real # native adapter that never defines the method. adapter.supports_inchannel_continuable_for_platform = None # 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" def test_in_channel_seed_fires_without_attach_to_session(self): """REGRESSION (live, Alice 2026-08-19): the in_channel seed was gated on mirror_this_target (mirror_enabled AND origin match), so a continuable in_channel cron created WITHOUT attach_to_session (and with the cron.mirror_delivery global at its default False) delivered the brief flat but never seeded the flat session — the next plain reply hit a blank session and the agent had no idea about its own delivery message. in_channel IS the continuation surface: the seed must fire on origin match alone. attach_to_session stays the opt-in for the SEPARATE default-surface mirror behavior; it must not be required here.""" from cron.scheduler import _deliver_result # noqa: F401 (driven via helper) adapter = self._slack_adapter(supports_inchannel=True) with patch("cron.scheduler._seed_cron_channel_session", return_value=True) as seed_mock: self._run_inchannel_delivery( {"slack": {"cron_continuable_surface": "in_channel"}}, adapter, attach_to_session=False, ) seed_mock.assert_called_once() # user_id must ride along even without the mirror opt-in — the flat # session key includes it on per-user-isolated chats. assert seed_mock.call_args.kwargs.get("user_id") == "U_HUMAN" def test_in_channel_flattens_thread_without_attach_to_session(self): """REGRESSION: the thread-id-clearing gate was `mirror_this_target` while the seed below had been decoupled to origin-match alone. With the default knobs off (attach_to_session=False, cron.mirror_delivery unset) and an origin carrying a REAL thread_id, the brief still delivered INTO the origin thread while the flat (thread_id=None) session got seeded — brief and continuation surface in different places. The flatten must use the same gate as the seed: origin_target. Asserts on the routed DeliveryTarget, not just that the seed ran.""" captured = {} class _SpyRouter: def __init__(self, *a, **k): pass async def _deliver_to_platform(self, target, text, metadata): captured["target"] = target return {"success": True, "message_id": "msg_1"} adapter = self._slack_adapter(supports_inchannel=True) origin_with_thread = { "platform": "slack", "chat_id": "C123", "user_id": "U_HUMAN", # Genuine origin thread (job created from inside a thread). "thread_id": "1787188000.000100", } with patch("gateway.delivery.DeliveryRouter", _SpyRouter), \ patch("cron.scheduler._seed_cron_channel_session", return_value=True) as seed_mock: self._run_inchannel_delivery( {"slack": {"cron_continuable_surface": "in_channel"}}, adapter, attach_to_session=False, origin=origin_with_thread, ) seed_mock.assert_called_once() assert captured["target"].thread_id is None, ( "in_channel delivery routed into the origin thread " f"({captured['target'].thread_id}) — the flat seeded session " "does not match where the brief actually landed" ) def test_origin_scope_id_rides_delivery_metadata(self): """REGRESSION (restart-shaped): the connector's fail-closed tenant guard resolves the workspace from metadata.scope_id. After a gateway restart the RelayAdapter's per-chat scope cache is cold, and DeliveryRouter stamps scope only for the configured HOME channel — so a scoped Slack origin that is NOT the home chat egressed with no scope_id and could be rejected before delivery. The scheduler must stamp the persisted origin scope onto origin-matching routing metadata (and never onto fan-out targets, which the origin-match gate already excludes).""" captured = {} class _SpyRouter: def __init__(self, *a, **k): pass async def _deliver_to_platform(self, target, text, metadata): captured["metadata"] = metadata return {"success": True, "message_id": "msg_1"} adapter = self._slack_adapter(supports_inchannel=True) scoped_origin = { "platform": "slack", "chat_id": "C123", "user_id": "U_HUMAN", # Persisted workspace scope (captured at job creation). C123 is # not any configured home channel in this harness. "scope_id": "T0AAAA111", } with patch("gateway.delivery.DeliveryRouter", _SpyRouter), \ patch("cron.scheduler._seed_cron_channel_session", return_value=True): self._run_inchannel_delivery( {"slack": {"cron_continuable_surface": "in_channel"}}, adapter, attach_to_session=False, origin=scoped_origin, ) assert captured["metadata"].get("scope_id") == "T0AAAA111", ( "persisted origin scope_id did not reach the delivery metadata — " "a cold-cache relay egress has no tenant discriminator" ) def test_legacy_origin_without_scope_stamps_nothing(self): """Legacy jobs (origin persisted before scope capture) must not gain a scope_id key — the relay's per-chat cache / home-channel stamping remain the only sources, exactly today's behavior.""" captured = {} class _SpyRouter: def __init__(self, *a, **k): pass async def _deliver_to_platform(self, target, text, metadata): captured["metadata"] = metadata return {"success": True, "message_id": "msg_1"} adapter = self._slack_adapter(supports_inchannel=True) with patch("gateway.delivery.DeliveryRouter", _SpyRouter), \ patch("cron.scheduler._seed_cron_channel_session", return_value=True): self._run_inchannel_delivery( {"slack": {"cron_continuable_surface": "in_channel"}}, adapter, attach_to_session=False, ) assert "scope_id" not in captured["metadata"] def test_native_adapter_scalar_false_fails_safe_to_thread(self): """D6 fallback boundary with a REAL (non-mock) adapter shape: a native adapter defines only the scalar supports_inchannel_continuable and no per-platform accessor — MagicMock-based fixtures can't prove this branch because they fabricate a truthy accessor. Scalar False must fail safe to thread mode: the in_channel seed never fires.""" class _NativeShapedAdapter: supports_inchannel_continuable = False _session_store = None def __init__(self): self.sent = [] async def send(self, chat_id, content, metadata=None, **kwargs): self.sent.append((chat_id, content)) return MagicMock(success=True, message_id="msg_1", raw_response=None) adapter = _NativeShapedAdapter() assert not callable( getattr(adapter, "supports_inchannel_continuable_for_platform", None) ) with patch("cron.scheduler._seed_cron_channel_session") as seed_mock: self._run_inchannel_delivery( {"slack": {"cron_continuable_surface": "in_channel"}}, adapter, attach_to_session=False, ) # Delivery must have gone through the LIVE adapter — otherwise a # broken harness that never delivers would also leave the seed # uncalled and this test would pass for the wrong reason. assert len(adapter.sent) == 1 assert adapter.sent[0][0] == "C123" # Capability absent -> surface fails safe to thread -> flat seed # must NOT run (D6). seed_mock.assert_not_called() def test_seed_mirrors_into_exact_created_session_on_populated_chat(self): """REGRESSION (live, Alice 2026-08-19 19:17): 'in_channel seed did NOT land'. The seed created the flat session row, then mirror_to_session re-discovered the target via origin heuristics — and on a populated chat (flat session + N per-message thread sessions sharing chat_id, mixed user_ids) find_session_by_origin's multi-candidate bail-out returned None, silently dropping the brief. The seed must mirror into the EXACT session row it just created, no rediscovery.""" from cron.scheduler import _seed_cron_channel_session store = MagicMock() created = MagicMock() created.session_id = "sess-flat-exact" store.get_or_create_session.return_value = created 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", "D0BJTDCSR7C", "Daily brief", is_dm=True, user_id="U_HUMAN", chat_name=None, ) assert ok is True # The mirror received the exact created session id — origin-scan # heuristics (and their populated-chat bail-out) are out of the path. assert mirror_mock.call_args.kwargs.get("session_id") == "sess-flat-exact" def test_in_channel_also_seeds_thread_surface_of_delivered_brief(self): """REGRESSION (live, Alice 2026-08-19 19:19): the user replied IN THE BRIEF'S THREAD (Slack's natural affordance on a flat message). That reply keys to (chat, thread=) — a session in_channel mode never seeded, so the agent had no idea about its own brief. The flat delivery's message_id must anchor a companion thread-surface seed.""" adapter = self._slack_adapter(supports_inchannel=True) with patch("cron.scheduler._seed_cron_channel_session", return_value=True), \ patch("cron.scheduler._seed_cron_thread_session") as thread_seed_mock: self._run_inchannel_delivery( {"slack": {"cron_continuable_surface": "in_channel"}}, adapter, attach_to_session=False, ) thread_seed_mock.assert_called_once() # Anchored on the delivered message id (the router's SendResult). assert thread_seed_mock.call_args.args[4] == "msg_1" 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 TestBuildJobPromptExtraPrompt: """Regression: _build_job_prompt merges extra_prompt into the assembled prompt.""" def test_extra_prompt_appended_with_header(self): """extra_prompt appears under a '## Run Context' header.""" job = {"prompt": "stored prompt"} result = _build_job_prompt(job, extra_prompt="CONTEXT: client=Foo") assert "stored prompt" in result assert "## Run Context" in result assert "CONTEXT: client=Foo" in result def test_extra_prompt_does_not_mutate_job(self): """The job dict's 'prompt' field must remain unchanged.""" job = {"prompt": "original"} _build_job_prompt(job, extra_prompt="transient context") assert job["prompt"] == "original" def test_no_extra_prompt_omits_header(self): """Without extra_prompt, no '## Run Context' header is injected.""" job = {"prompt": "just the stored prompt"} result = _build_job_prompt(job) assert "## Run Context" not in result assert "just the stored prompt" in result 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") class TestFailureStreakNudge: """Poke-inspired repeated-failure review nudge (_failure_streak_nudge).""" def _job(self, streak, kind="cron", name="scout"): return { "id": "j1", "name": name, "failure_streak": streak, "schedule": {"kind": kind}, } def test_nudges_at_threshold(self): from cron.scheduler import _failure_streak_nudge # stored streak 2 + this run = 3 >= default threshold 3 with patch("cron.scheduler.load_config", return_value={}): out = _failure_streak_nudge(self._job(2)) assert "failed 3 runs in a row" in out assert "hermes cron pause scout" in out def test_silent_below_threshold(self): from cron.scheduler import _failure_streak_nudge with patch("cron.scheduler.load_config", return_value={}): assert _failure_streak_nudge(self._job(0)) == "" assert _failure_streak_nudge(self._job(1)) == "" def test_oneshot_never_nudges(self): from cron.scheduler import _failure_streak_nudge with patch("cron.scheduler.load_config", return_value={}): assert _failure_streak_nudge(self._job(10, kind="once")) == "" def test_config_threshold_and_disable(self): from cron.scheduler import _failure_streak_nudge cfg5 = {"cron": {"failure_nudge_threshold": 5}} with patch("cron.scheduler.load_config", return_value=cfg5): assert _failure_streak_nudge(self._job(3)) == "" assert "failed 5 runs" in _failure_streak_nudge(self._job(4)) with patch("cron.scheduler.load_config", return_value={"cron": {"failure_nudge_threshold": 0}}): assert _failure_streak_nudge(self._job(50)) == "" def test_missing_streak_field_backcompat(self): from cron.scheduler import _failure_streak_nudge job = {"id": "old", "schedule": {"kind": "interval"}} # pre-field job with patch("cron.scheduler.load_config", return_value={}): assert _failure_streak_nudge(job) == "" def test_config_load_failure_falls_back(self): from cron.scheduler import _failure_streak_nudge with patch("cron.scheduler.load_config", side_effect=RuntimeError("boom")): assert "failed 3 runs" in _failure_streak_nudge(self._job(2))