From 0d4ceb7fa4a62cf370fdf8175d0e04c436d132fb Mon Sep 17 00:00:00 2001 From: dongjiang Date: Tue, 11 Aug 2026 08:52:48 +0800 Subject: [PATCH] fix(cron): multiplex delivery uses owning profile's bot token and adapter (#83182) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two root causes under multiplex gateway: 1. Secret scope reset before delivery: run_one_job's inner finally block reset the profile's secret scope after run_job returned but BEFORE _deliver_result ran. load_gateway_config → _getenv fell back to os.environ (empty in the multiplex unit) — TELEGRAM_BOT_TOKEN resolved to the wrong bot and delivery routed to the wrong chat. Fix: move reset_secret_scope to an outer finally so the scope stays installed through BOTH execution and delivery. 2. Shared adapters dict for delivery: The cron ticker received only runner.adapters (default profile's live adapters). Secondary-profile jobs therefore delivered via the default profile's adapter — even though Gateway._profile_adapters[profile] had the right per-profile adapter live and ready. Fix: gateway/run.py builds profile_adapters_by_home (resolved hermes home path → per-profile adapter map) and passes it through the cron chain: scheduler_provider → tick → run_one_job → _deliver_result. run_one_job resolves the owning profile's adapter map via the current hermes home (set by _start_multiplex's per-profile override), so delivery picks the right bot token. Files changed: cron/scheduler.py - scope restructure + profile_adapters param chain cron/scheduler_provider.py - propagate profile_adapters_by_home gateway/run.py - build profile_adapters_by_home from runner tests/cron/test_cron_multiplex_delivery_83182.py - 6 new regression tests tests/cron/test_run_one_job.py - fake_deliver signature tests/cron/test_preflight_config.py - fake_deliver signature Fixes #83182 Signed-off-by: dongjiang --- cron/scheduler.py | 276 +++++++++------ cron/scheduler_provider.py | 9 + gateway/run.py | 23 ++ .../test_cron_multiplex_delivery_83182.py | 313 ++++++++++++++++++ tests/cron/test_preflight_config.py | 4 +- tests/cron/test_run_one_job.py | 2 +- 6 files changed, 514 insertions(+), 113 deletions(-) create mode 100644 tests/cron/test_cron_multiplex_delivery_83182.py diff --git a/cron/scheduler.py b/cron/scheduler.py index 0102f1f64364e..04567f61fcfe3 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -34,7 +34,7 @@ except ImportError: msvcrt = None from pathlib import Path -from typing import Any, List, Optional +from typing import Any, Dict, List, Optional # Add parent directory to path for imports BEFORE repo-level imports. # Without this, standalone invocations (e.g. after `hermes update` reloads @@ -1602,7 +1602,10 @@ def _is_channel_dm_topic( return is_channel -def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Optional[str]: +def _deliver_result( + job: dict, content: str, adapters=None, loop=None, + profile_adapters: Optional[Dict] = None, +) -> Optional[str]: """ Deliver job output to the configured target(s) (origin chat, specific platform, etc.). @@ -1611,6 +1614,11 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option the standalone HTTP path cannot encrypt. Falls back to standalone send if the adapter path fails or is unavailable. + ``profile_adapters`` is the owning profile's live adapter map (multiplex + mode). When set, it is preferred over the shared ``adapters`` dict so that + secondary-profile cron delivers via that profile's bot token, not the + default profile's (#83182). + Returns None on success, or an error string on failure. """ targets = _resolve_delivery_targets(job) @@ -1687,6 +1695,15 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option logger.error("Job '%s': %s", job["id"], msg) return msg + # Multiplex-aware adapter selection (#83182). When the cron scheduler was + # started with per-profile adapter maps (``profile_adapters_by_home``), + # prefer the owning profile's live adapters over the shared default dict. + # The secret scope (still active through delivery thanks to the Part-1 + # restructure) ensures load_gateway_config() above already returned the + # correct profile's gateway.yaml — using the matching adapter map keeps + # bot-token selection in lockstep with the config. + delivery_adapters = profile_adapters if profile_adapters is not None else adapters + delivery_errors = [] for target in targets: @@ -1731,7 +1748,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option from gateway.delivery import resolve_delivery_transport - transport = resolve_delivery_transport(platform, config, adapters) + transport = resolve_delivery_transport(platform, config, delivery_adapters) if transport is not None: pconfig = transport.config runtime_adapter = transport.adapter @@ -1941,7 +1958,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option if text_to_send: from agent.async_utils import safe_schedule_threadsafe - router = DeliveryRouter(config, adapters) + router = DeliveryRouter(config, delivery_adapters) route_target = DeliveryTarget( platform=platform, chat_id=str(chat_id), @@ -4539,6 +4556,7 @@ def _teardown_cron_agent(agent, job_id: str) -> None: def run_one_job( job: dict, *, adapters=None, loop=None, verbose: bool = False, extra_prompt: Optional[str] = None, + profile_adapters_by_home: Optional[Dict] = None, ) -> bool: """Run ONE due job end-to-end: execute → save output → deliver → mark. @@ -4605,115 +4623,145 @@ def run_one_job( # below once delivery is done. Defense-in-depth alongside the # interpreter-shutdown guard in _deliver_result. _deferred_agents: list = [] + # The secret scope MUST stay installed through _deliver_result. + # Resetting it between run_job and delivery breaks multiplex cron: + # delivery calls load_gateway_config() → _getenv() which falls back + # to os.environ (empty in the multiplex unit) instead of the owning + # profile's .env — TELEGRAM_BOT_TOKEN resolves to the wrong bot + # (or nothing), and delivery routes to the wrong chat (#83182). + # The scope is torn down in the outer finally so BOTH execution and + # delivery see the owning profile's secrets. try: - success, output, final_response, error = run_job( - job, defer_agent_teardown=_deferred_agents, - extra_prompt=extra_prompt, - ) - except BaseException: - # run_job's finally still hands back the agent when it raises; tear - # it down here so a failed run never leaks its async resources - # (#10200), then re-raise into the outer handler. BaseException - # (not just Exception) so a KeyboardInterrupt/SystemExit mid-run - # still triggers teardown before propagating. - for _deferred_agent in _deferred_agents: - _teardown_cron_agent(_deferred_agent, job["id"]) - raise - finally: - reset_secret_scope(_scope_token) - - # Everything from here through delivery runs with the agent still live - # (deferred teardown). Wrap it ALL in a try/finally so that if any step - # between run_job returning and delivery — save_job_output, the [SILENT] - # / empty-response computation, or _deliver_result itself — raises, the - # deferred agent is still torn down. Otherwise the outer `except` would - # swallow the error and leak the agent's subprocesses/clients (#10200). - delivery_error = None - blocked_config = False - try: - output_file = save_job_output(job["id"], output) - if verbose: - logger.info("Output saved to: %s", output_file) - - # If the gateway shutdown killed this job's tool subprocess - # mid-flight (#60432), the agent may still have produced a - # plausible-looking final_response from the truncated output -- - # force the failure path so the delivered message is an honest - # "this run was interrupted" summary instead of that response. - # Peek-only: the flag stays set for the authoritative check - # right before mark_job_run below. - if success and _is_interrupted(job["id"]): - success = False - error = ( - "Interrupted by gateway shutdown before the run finished " - "(tool subprocess was killed mid-flight)." + try: + success, output, final_response, error = run_job( + job, defer_agent_teardown=_deferred_agents, + extra_prompt=extra_prompt, ) + except BaseException: + # run_job's finally still hands back the agent when it raises; + # tear it down here so a failed run never leaks its async + # resources (#10200), then re-raise into the outer handler. + # BaseException (not just Exception) so a + # KeyboardInterrupt/SystemExit mid-run still triggers teardown + # before propagating. + for _deferred_agent in _deferred_agents: + _teardown_cron_agent(_deferred_agent, job["id"]) + raise - # Deliver the final response to the origin/target chat. - # If the agent responded with [SILENT], skip delivery (but - # output is already saved above). Failed jobs always deliver. - # - # Exception: a run blocked by pre-dispatch config validation - # (T1-26) alerts exactly ONCE — the silent marker means the - # operator was already told on a previous tick, so re-delivering - # the same alert every tick would be spam (#73506 alert-once - # shape). - blocked_config_silent = ( - bool(error) and BLOCKED_CONFIG_SILENT_MARKER in str(error) - ) - blocked_config = blocked_config_silent or ( - bool(error) and BLOCKED_CONFIG_MARKER in str(error) - ) - if blocked_config and not success: - # Blocked-config alert: bypass the generic failure summarizer - # (whose auth/timeout heuristics would mislabel this as a - # provider runtime failure) — say plainly that config - # validation blocked the run and nothing was spent. - _pf_text = re.sub( - r"\[blocked_config[^\]]*\]\s*", "", str(error) - ).strip() - deliver_content = ( - f"⛔ Cron '{job.get('name') or job['id']}' blocked by " - f"configuration validation (no LLM call was made): " - f"{_pf_text} " - "This alert is sent once; the job stays blocked until " - "the configuration is fixed." + # Everything from here through delivery runs with the agent still + # live (deferred teardown) AND the secret scope still installed. + # Wrap it ALL in a try/finally so that if any step between + # run_job returning and delivery — save_job_output, the [SILENT] + # / empty-response computation, or _deliver_result itself — + # raises, the deferred agent is still torn down. Otherwise the + # outer `except` would swallow the error and leak the agent's + # subprocesses/clients (#10200). + delivery_error = None + blocked_config = False + try: + output_file = save_job_output(job["id"], output) + if verbose: + logger.info("Output saved to: %s", output_file) + + # If the gateway shutdown killed this job's tool subprocess + # mid-flight (#60432), the agent may still have produced a + # plausible-looking final_response from the truncated output -- + # force the failure path so the delivered message is an honest + # "this run was interrupted" summary instead of that response. + # Peek-only: the flag stays set for the authoritative check + # right before mark_job_run below. + if success and _is_interrupted(job["id"]): + success = False + error = ( + "Interrupted by gateway shutdown before the run finished " + "(tool subprocess was killed mid-flight)." + ) + + # Deliver the final response to the origin/target chat. + # If the agent responded with [SILENT], skip delivery (but + # output is already saved above). Failed jobs always deliver. + # + # Exception: a run blocked by pre-dispatch config validation + # (T1-26) alerts exactly ONCE — the silent marker means the + # operator was already told on a previous tick, so re-delivering + # the same alert every tick would be spam (#73506 alert-once + # shape). + blocked_config_silent = ( + bool(error) and BLOCKED_CONFIG_SILENT_MARKER in str(error) ) - else: - deliver_content = final_response if success else _summarize_cron_failure_for_delivery(job, error) - # Treat whitespace-only final responses the same as empty - # responses: do not deliver a blank message, and let the - # empty-response guard below mark the run as a soft failure. - should_deliver = bool(deliver_content.strip()) - if blocked_config_silent: - should_deliver = False - unresolved_origin = False - # Cron silence suppression — see _is_cron_silence_response. Replaces the - # old `SILENT_MARKER in ...upper()` substring check, which both leaked - # bracketless near-markers ("SILENT" / "NO_REPLY") and wrongly swallowed - # a real report that merely quoted "[SILENT]" mid-sentence (#51438, - # #46917). Keeps the intentional bracketed-prefix / trailing-line - # tolerance the cron contract relies on. - if should_deliver and success and _is_cron_silence_response(deliver_content): - logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER) - should_deliver = False - - if should_deliver: - unresolved_origin = ( - _normalize_deliver_value(job.get("deliver", "local")) == "origin" - and not _resolve_delivery_targets(job) + blocked_config = blocked_config_silent or ( + bool(error) and BLOCKED_CONFIG_MARKER in str(error) ) - try: - delivery_error = _deliver_result(job, deliver_content, adapters=adapters, loop=loop) - except Exception as de: - delivery_error = str(de) - logger.error("Delivery failed for job %s: %s", job["id"], de) + if blocked_config and not success: + # Blocked-config alert: bypass the generic failure summarizer + # (whose auth/timeout heuristics would mislabel this as a + # provider runtime failure) — say plainly that config + # validation blocked the run and nothing was spent. + _pf_text = re.sub( + r"\[blocked_config[^\]]*\]\s*", "", str(error) + ).strip() + deliver_content = ( + f"⛔ Cron '{job.get('name') or job['id']}' blocked by " + f"configuration validation (no LLM call was made): " + f"{_pf_text} " + "This alert is sent once; the job stays blocked until " + "the configuration is fixed." + ) + else: + deliver_content = final_response if success else _summarize_cron_failure_for_delivery(job, error) + # Treat whitespace-only final responses the same as empty + # responses: do not deliver a blank message, and let the + # empty-response guard below mark the run as a soft failure. + should_deliver = bool(deliver_content.strip()) + if blocked_config_silent: + should_deliver = False + unresolved_origin = False + # Cron silence suppression — see _is_cron_silence_response. Replaces the + # old `SILENT_MARKER in ...upper()` substring check, which both leaked + # bracketless near-markers ("SILENT" / "NO_REPLY") and wrongly swallowed + # a real report that merely quoted "[SILENT]" mid-sentence (#51438, + # #46917). Keeps the intentional bracketed-prefix / trailing-line + # tolerance the cron contract relies on. + if should_deliver and success and _is_cron_silence_response(deliver_content): + logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER) + should_deliver = False + + if should_deliver: + unresolved_origin = ( + _normalize_deliver_value(job.get("deliver", "local")) == "origin" + and not _resolve_delivery_targets(job) + ) + # Resolve the owning profile's live adapter map for + # multiplex delivery (#83182). The cron ticker's + # ``_start_multiplex`` sets the hermes-home override + # per profile before ticking, so _get_hermes_home() + # returns the job-owning profile's resolved path — + # use it to look up the matching adapter map. + _profile_adapters = None + if profile_adapters_by_home: + _home_key = str(_get_hermes_home().resolve()) + _profile_adapters = profile_adapters_by_home.get(_home_key) + try: + delivery_error = _deliver_result( + job, deliver_content, + adapters=adapters, loop=loop, + profile_adapters=_profile_adapters, + ) + except Exception as de: + delivery_error = str(de) + logger.error("Delivery failed for job %s: %s", job["id"], de) + finally: + # Tear down the deferred agent(s) now that save + delivery have + # run (or raised). Must happen on every path so cron agents + # never leak their subprocesses/clients (#10200). + for _deferred_agent in _deferred_agents: + _teardown_cron_agent(_deferred_agent, job["id"]) finally: - # Tear down the deferred agent(s) now that save + delivery have run - # (or raised). Must happen on every path so cron agents never leak - # their subprocesses/clients (#10200). - for _deferred_agent in _deferred_agents: - _teardown_cron_agent(_deferred_agent, job["id"]) + # Tear down the secret scope last — AFTER both run_job AND + # _deliver_result complete. This is the fix for #83182: delivery + # must see the owning profile's secrets (TELEGRAM_BOT_TOKEN etc.) + # so multiplex cron delivers via the correct bot. + reset_secret_scope(_scope_token) # Treat empty final_response as a soft failure so last_status # is not "ok" — the agent ran but produced nothing useful. @@ -4850,19 +4898,24 @@ def tick( sync: bool = True, *, can_dispatch=None, + profile_adapters_by_home=None, ): """ Check and run all due jobs. - + Uses a file lock so only one tick runs at a time, even if the gateway's in-process ticker and a standalone daemon or manual tick overlap. - + Args: verbose: Whether to print status messages adapters: Optional dict mapping Platform → live adapter (from gateway) loop: Optional asyncio event loop (from gateway) for live adapter sends can_dispatch: Optional synchronous gate; false leaves due jobs untouched for the next allowed tick + profile_adapters_by_home: Optional mapping of resolved profile home + paths to that profile's live adapter dict (multiplex mode). Used + by ``run_one_job`` to select the owning profile's adapter map + instead of the shared default (#83182). Returns: Number of jobs executed (0 if another tick is already running) @@ -4961,7 +5014,10 @@ def _process_job(job: dict) -> bool: module-level ``run_one_job`` so ``tick`` and external providers (Chronos ``fire_due``) use the identical execute→save→deliver→mark body.""" - return run_one_job(job, adapters=adapters, loop=loop, verbose=verbose) + return run_one_job( + job, adapters=adapters, loop=loop, verbose=verbose, + profile_adapters_by_home=profile_adapters_by_home, + ) # Partition due jobs: those with a per-job workdir mutate # os.environ["TERMINAL_CWD"] inside run_job, which is process-global, so diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index db3641a8c9633..d13e4912f1b6d 100644 --- a/cron/scheduler_provider.py +++ b/cron/scheduler_provider.py @@ -192,6 +192,7 @@ def start( interval=60, can_dispatch=None, profile_homes=None, + profile_adapters_by_home=None, ): import logging from cron.scheduler import tick as cron_tick @@ -219,6 +220,7 @@ def start( loop=loop, interval=interval, can_dispatch=can_dispatch, + profile_adapters_by_home=profile_adapters_by_home, ) return @@ -279,6 +281,7 @@ def _start_multiplex( loop=None, interval=60, can_dispatch=None, + profile_adapters_by_home=None, ): """Tick every served profile's cron store when multiplex_profiles is on. @@ -287,6 +290,11 @@ def _start_multiplex( agent execution to that profile's home — mirroring how ``_profile_runtime_scope`` scopes the multiplexed inbound path and ``web_server.py`` scopes per-profile cron API calls. + + ``profile_adapters_by_home`` is a mapping of resolved profile home + paths to that profile's live adapter dict (multiplex-only). Passed + through to ``run_one_job`` so cron delivery uses the owning profile's + adapter instead of the shared default dict (#83182). """ import logging from cron.scheduler import tick as cron_tick @@ -339,6 +347,7 @@ def _start_multiplex( loop=loop, sync=False, can_dispatch=can_dispatch, + profile_adapters_by_home=profile_adapters_by_home, ) finally: reset_hermes_home_override(home_token) diff --git a/gateway/run.py b/gateway/run.py index 11e6fd7f0fbfb..ff3b20524df79 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -27967,6 +27967,29 @@ def restart_signal_handler(): len(profile_homes), [p[0] if isinstance(p, tuple) else p for p in profile_homes], ) + # Build per-profile live-adapter map (#83182). Cron delivery + # uses this to pick the owning profile's adapter (Telegram + # bot token, Matrix client, etc.) instead of the shared + # default dict. Keyed by resolved home path so run_one_job + # can look up the right map via _get_hermes_home(). + from hermes_cli.profiles import get_active_profile_name + + _profile_adapters_by_home: Dict[str, Dict[Any, Any]] = {} + for _pname, _phome in profile_homes: + _resolved_home = str(Path(_phome).resolve()) + _amap = getattr(runner, "_profile_adapters", {}).get(_pname) + if _amap is not None: + _profile_adapters_by_home[_resolved_home] = dict(_amap) + elif _pname == (get_active_profile_name() or "default"): + # Default profile's adapters live on runner.adapters. + if runner.adapters: + _profile_adapters_by_home[_resolved_home] = dict(runner.adapters) + if _profile_adapters_by_home: + cron_start_kwargs["profile_adapters_by_home"] = _profile_adapters_by_home + logger.info( + "Cron delivery will use per-profile adapters for %d profile(s)", + len(_profile_adapters_by_home), + ) except Exception as exc: logger.warning( "Could not resolve profile homes for multiplex cron: %s", diff --git a/tests/cron/test_cron_multiplex_delivery_83182.py b/tests/cron/test_cron_multiplex_delivery_83182.py new file mode 100644 index 0000000000000..dc65b21047084 --- /dev/null +++ b/tests/cron/test_cron_multiplex_delivery_83182.py @@ -0,0 +1,313 @@ +"""Regression tests for #83182 — cron delivery must use the owning profile's +bot token and adapter under multiplex. + +Two root causes, both fixed: +1. Secret scope was reset in run_one_job's inner finally BEFORE _deliver_result + ran. load_gateway_config() → _getenv() then fell back to os.environ (empty + in the multiplex unit) — TELEGRAM_BOT_TOKEN resolved to the wrong bot. +2. Cron delivery used the shared runner.adapters dict (default profile) even + when multiplexing. Secondary profile jobs delivered via the default bot. + +Fix: + - Part 1: Scope reset moved to outer finally, covers execution + delivery. + - Part 2: profile_adapters_by_home propagated through gateway → cron chain; + _deliver_result prefers the owning profile's adapter map. +""" +import importlib +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Part 1: Secret scope stays installed through delivery +# --------------------------------------------------------------------------- + + +class TestSecretScopeThroughDelivery: + """run_one_job must keep the profile's secret scope installed when + _deliver_result runs, so load_gateway_config picks up the right + TELEGRAM_BOT_TOKEN from the profile's .env. + """ + + def test_scope_active_during_delivery(self, tmp_path, monkeypatch): + """Monkey-patch run_job and _deliver_result to observe scope state + at delivery time. + """ + import hermes_constants + from cron import scheduler as sched_mod + + # Point hermes home at our tmp_path + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + importlib.reload(hermes_constants) + + # Stub out heavy imports/modules that run_job and delivery touch + monkeypatch.setattr(sched_mod, "run_job", lambda *a, **kw: (True, "out", "response", None)) + monkeypatch.setattr(sched_mod, "save_job_output", lambda *a, **kw: "/tmp/fake") + monkeypatch.setattr(sched_mod, "_is_interrupted", lambda *a, **kw: False) + monkeypatch.setattr(sched_mod, "_consume_interrupted_flag", lambda *a, **kw: False) + monkeypatch.setattr(sched_mod, "mark_job_run", lambda *a, **kw: None) + monkeypatch.setattr(sched_mod, "claim_dispatch", lambda *a, **kw: True) + monkeypatch.setattr(sched_mod, "mark_execution_running", lambda *a, **kw: None) + monkeypatch.setattr(sched_mod, "create_execution", lambda *a, **kw: {"id": "exec-1"}) + monkeypatch.setattr(sched_mod, "finish_execution", lambda *a, **kw: None) + + # Track the secret scope state at delivery time + scope_state_at_delivery = {} + + def fake_deliver(job, content, **kwargs): + from agent.secret_scope import current_secret_scope + scope_state_at_delivery["scope"] = current_secret_scope() + return None + + monkeypatch.setattr(sched_mod, "_deliver_result", fake_deliver) + + job = { + "id": "job-1", + "name": "test-job", + "deliver": "local", # local delivery skips the send path + } + + sched_mod.run_one_job(job, adapters=None, loop=None) + + # The key assertion: secret scope is still installed when delivery runs. + # (Not None — which was the pre-fix state.) + # Note: may be None if no .env exists in the test hermes home, but the + # important thing is that set_secret_scope was called and reset was + # NOT called before delivery. We test the ordering separately below. + # For now, just assert run_one_job completed without raising. + + def test_scope_reset_after_delivery_not_before(self, tmp_path, monkeypatch): + """Verify the ordering: reset_secret_scope fires AFTER _deliver_result, + not before. We track the call sequence explicitly. + """ + import hermes_constants + from cron import scheduler as sched_mod + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + importlib.reload(hermes_constants) + + monkeypatch.setattr(sched_mod, "run_job", lambda *a, **kw: (True, "out", "response", None)) + monkeypatch.setattr(sched_mod, "save_job_output", lambda *a, **kw: "/tmp/fake") + monkeypatch.setattr(sched_mod, "_is_interrupted", lambda *a, **kw: False) + monkeypatch.setattr(sched_mod, "_consume_interrupted_flag", lambda *a, **kw: False) + monkeypatch.setattr(sched_mod, "mark_job_run", lambda *a, **kw: None) + monkeypatch.setattr(sched_mod, "claim_dispatch", lambda *a, **kw: True) + monkeypatch.setattr(sched_mod, "mark_execution_running", lambda *a, **kw: None) + monkeypatch.setattr(sched_mod, "create_execution", lambda *a, **kw: {"id": "exec-1"}) + monkeypatch.setattr(sched_mod, "finish_execution", lambda *a, **kw: None) + + events = [] + + def fake_deliver(*a, **kw): + events.append("deliver") + return None + + # Wrap reset_secret_scope to observe the call + from agent import secret_scope as scope_mod + _orig_reset = scope_mod.reset_secret_scope + + def tracking_reset(token): + events.append("reset_scope") + return _orig_reset(token) + + monkeypatch.setattr(sched_mod, "_deliver_result", fake_deliver) + monkeypatch.setattr(scope_mod, "reset_secret_scope", tracking_reset) + + job = {"id": "job-1", "name": "test-job", "deliver": "local"} + sched_mod.run_one_job(job) + + # Verify ordering: deliver must come BEFORE reset_scope + assert "deliver" in events, f"deliver was not called: {events}" + assert "reset_scope" in events, f"reset_scope was not called: {events}" + assert events.index("deliver") < events.index("reset_scope"), ( + f"delivery ({events.index('deliver')}) should happen BEFORE " + f"scope reset ({events.index('reset_scope')}); events={events}" + ) + + +# --------------------------------------------------------------------------- +# Part 2: Profile-specific adapter selection +# --------------------------------------------------------------------------- + + +class TestProfileAdapterSelection: + """_deliver_result must use profile-specific adapters when available, + falling back to the shared dict when not. + """ + + def test_profile_adapters_preferred_over_shared(self, monkeypatch): + """When profile_adapters is set, it wins over the shared adapters.""" + from cron.scheduler import _deliver_result + + captured = {} + + def fake_resolve(platform, config, adapters): + captured["adapters"] = adapters + return None # no live transport → fall through + + monkeypatch.setattr( + "gateway.delivery.resolve_delivery_transport", + fake_resolve, + ) + # Stub _resolve_delivery_targets (module-level in cron.scheduler) + monkeypatch.setattr( + "cron.scheduler._resolve_delivery_targets", + lambda job: [{"platform": "telegram", "chat_id": "123"}], + ) + # Stub config loaders (imported inside _deliver_result) + monkeypatch.setattr( + "gateway.config.load_gateway_config", + lambda: MagicMock(platforms={}), + ) + monkeypatch.setattr( + "cron.scheduler.load_config", + lambda: {"cron": {"wrap_response": False}}, + ) + + shared_adapters = {"shared": "adapter"} + profile_adapters = {"telegram": MagicMock()} + + job = {"id": "job-1", "deliver": "telegram"} + + # Call with both — profile_adapters should win + _deliver_result( + job, "hello", + adapters=shared_adapters, loop=None, + profile_adapters=profile_adapters, + ) + + assert captured.get("adapters") is profile_adapters, ( + "profile_adapters should be preferred over shared adapters" + ) + + def test_fallback_to_shared_adapters(self, monkeypatch): + """When profile_adapters is None, shared adapters are used (back-compat).""" + from cron.scheduler import _deliver_result + + captured = {} + + def fake_resolve(platform, config, adapters): + captured["adapters"] = adapters + return None + + monkeypatch.setattr( + "gateway.delivery.resolve_delivery_transport", + fake_resolve, + ) + monkeypatch.setattr( + "cron.scheduler._resolve_delivery_targets", + lambda job: [{"platform": "telegram", "chat_id": "123"}], + ) + monkeypatch.setattr( + "gateway.config.load_gateway_config", + lambda: MagicMock(platforms={}), + ) + monkeypatch.setattr( + "cron.scheduler.load_config", + lambda: {"cron": {"wrap_response": False}}, + ) + + shared_adapters = {"shared": "adapter"} + + job = {"id": "job-1", "deliver": "telegram"} + + # Call without profile_adapters — shared should be used + _deliver_result( + job, "hello", + adapters=shared_adapters, loop=None, + profile_adapters=None, + ) + + assert captured.get("adapters") is shared_adapters, ( + "shared adapters should be used when profile_adapters is None" + ) + + +# --------------------------------------------------------------------------- +# run_one_job profile_adapters_by_home resolution +# --------------------------------------------------------------------------- + + +class TestRunOneJobProfileAdaptersResolution: + """run_one_job must look up the right profile adapters from the map + using the current hermes home. + """ + + def test_resolves_adapters_by_home(self, tmp_path, monkeypatch): + """run_one_job resolves the owning profile's adapters from the + profile_adapters_by_home map keyed by resolved hermes home path. + """ + import hermes_constants + from cron import scheduler as sched_mod + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + importlib.reload(hermes_constants) + + # Stubs + monkeypatch.setattr(sched_mod, "run_job", lambda *a, **kw: (True, "out", "response", None)) + monkeypatch.setattr(sched_mod, "save_job_output", lambda *a, **kw: "/tmp/fake") + monkeypatch.setattr(sched_mod, "_is_interrupted", lambda *a, **kw: False) + monkeypatch.setattr(sched_mod, "_consume_interrupted_flag", lambda *a, **kw: False) + monkeypatch.setattr(sched_mod, "mark_job_run", lambda *a, **kw: None) + monkeypatch.setattr(sched_mod, "claim_dispatch", lambda *a, **kw: True) + monkeypatch.setattr(sched_mod, "mark_execution_running", lambda *a, **kw: None) + monkeypatch.setattr(sched_mod, "create_execution", lambda *a, **kw: {"id": "exec-1"}) + monkeypatch.setattr(sched_mod, "finish_execution", lambda *a, **kw: None) + + captured = {} + + def fake_deliver(job, content, adapters=None, loop=None, profile_adapters=None): + captured["profile_adapters"] = profile_adapters + captured["shared_adapters"] = adapters + return None + + monkeypatch.setattr(sched_mod, "_deliver_result", fake_deliver) + + # Build the profile_adapters_by_home map keyed by resolved tmp_path + profile_adapter = {"telegram": "profile-specific-adapter"} + profile_adapters_by_home = {str(tmp_path.resolve()): profile_adapter} + shared_adapters = {"shared": "adapter"} + + job = {"id": "job-1", "name": "test-job", "deliver": "local"} + + sched_mod.run_one_job( + job, adapters=shared_adapters, loop=None, + profile_adapters_by_home=profile_adapters_by_home, + ) + + assert captured.get("profile_adapters") is profile_adapter, ( + "run_one_job should resolve the owning profile's adapter map" + ) + assert captured.get("shared_adapters") is shared_adapters + + def test_no_profile_adapters_map(self, tmp_path, monkeypatch): + """When profile_adapters_by_home is not provided, profile_adapters is None.""" + import hermes_constants + from cron import scheduler as sched_mod + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + importlib.reload(hermes_constants) + + monkeypatch.setattr(sched_mod, "run_job", lambda *a, **kw: (True, "out", "response", None)) + monkeypatch.setattr(sched_mod, "save_job_output", lambda *a, **kw: "/tmp/fake") + monkeypatch.setattr(sched_mod, "_is_interrupted", lambda *a, **kw: False) + monkeypatch.setattr(sched_mod, "_consume_interrupted_flag", lambda *a, **kw: False) + monkeypatch.setattr(sched_mod, "mark_job_run", lambda *a, **kw: None) + monkeypatch.setattr(sched_mod, "claim_dispatch", lambda *a, **kw: True) + monkeypatch.setattr(sched_mod, "mark_execution_running", lambda *a, **kw: None) + monkeypatch.setattr(sched_mod, "create_execution", lambda *a, **kw: {"id": "exec-1"}) + monkeypatch.setattr(sched_mod, "finish_execution", lambda *a, **kw: None) + + captured = {} + + def fake_deliver(job, content, adapters=None, loop=None, profile_adapters=None): + captured["profile_adapters"] = profile_adapters + return None + + monkeypatch.setattr(sched_mod, "_deliver_result", fake_deliver) + + job = {"id": "job-1", "name": "test-job", "deliver": "local"} + sched_mod.run_one_job(job) + + assert captured.get("profile_adapters") is None diff --git a/tests/cron/test_preflight_config.py b/tests/cron/test_preflight_config.py index 495072ee10817..d756e06deca6a 100644 --- a/tests/cron/test_preflight_config.py +++ b/tests/cron/test_preflight_config.py @@ -129,7 +129,7 @@ def test_single_alert_across_two_ticks_and_blocked_status(self, tmp_path): job = _job() deliveries = [] - def fake_deliver(job, content, adapters=None, loop=None): + def fake_deliver(job, content, adapters=None, loop=None, profile_adapters=None): deliveries.append(content) return None @@ -233,7 +233,7 @@ def test_preflight_false_restores_old_behavior(self, tmp_path): job = _job() deliveries = [] - def fake_deliver(job, content, adapters=None, loop=None): + def fake_deliver(job, content, adapters=None, loop=None, profile_adapters=None): deliveries.append(content) return None diff --git a/tests/cron/test_run_one_job.py b/tests/cron/test_run_one_job.py index a61866c54413d..05c0ad62f2faf 100644 --- a/tests/cron/test_run_one_job.py +++ b/tests/cron/test_run_one_job.py @@ -27,7 +27,7 @@ def fake_save(jid, out): calls.append(("save", jid)) return f"/tmp/{jid}.txt" - def fake_deliver(job, content, adapters=None, loop=None): + def fake_deliver(job, content, adapters=None, loop=None, profile_adapters=None): calls.append(("deliver", job["id"])) return None