fix(cron): delivery uses owning profile's secret scope + adapters under multiplex

cron/scheduler.py

@@ -3065,7 +3065,48 @@ def _is_channel_dm_topic(
     return is_channel
 
 
-def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Optional[str]:
+def _deliver_adapters_for_job(job: dict, adapters=None, profile_adapters=None):
+    """Resolve the effective live-adapter dict for a cron delivery.
+
+    Prefers the owning profile's adapter map (``profile_adapters``) when the
+    job belongs to a secondary profile under multiplex, so delivery uses that
+    profile's live bot/chat instead of the default profile's shared
+    ``adapters`` dict. Falls back to the shared ``adapters`` dict for the
+    default profile / non-multiplex path (existing behavior unchanged).
+
+    Mirrors the lookup in gateway/authz_mixin (``_profile_adapters[profile]``),
+    but keyed off the job's active profile home resolved at delivery time.
+
+    Identity-boundary contract (review #83197): for ANY resolved non-default
+    profile, this returns that profile's non-empty map or ``{}`` — NEVER the
+    default profile's shared ``adapters`` map. Falling back to the default
+    map when the owning profile's registry is missing/empty would recreate
+    the exact wrong-bot/wrong-chat identity path this fix exists to remove.
+    ``{}`` still permits the existing standalone delivery path to use the
+    active profile's scoped credential. On profile-resolution failure the
+    authority-safe fallback is also ``{}``.
+    """
+    if profile_adapters is not None:
+        try:
+            from hermes_cli.profiles import get_active_profile_name
+            profile = get_active_profile_name() or "default"
+            if profile != "default":
+                profile_map = profile_adapters.get(profile)
+                if isinstance(profile_map, dict):
+                    return profile_map
+                return {}
+        except Exception:
+            logger.debug(
+                "profile-adapters resolution failed; returning empty adapter map "
+                "(never the default profile's shared adapters)",
+                exc_info=True,
+            )
+            return {}
+    return adapters or {}
+
+
+def _deliver_result(job: dict, content: str, adapters=None, loop=None,
+                    profile_adapters=None) -> Optional[str]:
     """
     Deliver job output to the configured target(s) (origin chat, specific platform, etc.).
 
@@ -3074,6 +3115,14 @@ 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`` (optional) is the gateway's per-profile adapter map
+    (Gateway._profile_adapters: ``{profile_name: {platform: adapter}}``) for
+    multiplex mode. When provided and the job's profile has a live adapter
+    entry, delivery routes through that profile's adapter instead of the
+    shared ``adapters`` dict (the default profile's), so a secondary-profile
+    cron job delivers from the correct bot/chat. Falls back to the shared
+    ``adapters`` dict for the default profile / non-multiplex path.
+
     Returns None on success, or an error string on failure.
     """
     targets = _resolve_delivery_targets(job)
@@ -3180,6 +3229,15 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
         logger.error("Job '%s': %s", job["id"], msg)
         return msg
 
+    # Under multiplex, delivery must use the job-owning profile's live adapter
+    # (its bot/chat), not the shared default-profile ``adapters`` dict — the
+    # gateway's per-profile adapter map is threaded through via
+    # ``profile_adapters``. Resolve it once here; both the transport lookup and
+    # the DeliveryRouter below consume it.
+    delivery_adapters = _deliver_adapters_for_job(
+        job, adapters=adapters, profile_adapters=profile_adapters
+    )
+
     delivery_errors = []
 
     for target in targets:
@@ -3265,7 +3323,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
@@ -3519,7 +3577,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),
@@ -6915,6 +6973,7 @@ def run_one_job(
     verbose: bool = False,
     extra_prompt: Optional[str] = None,
     cancel_event: Optional[_CancelEventLike] = None,
+    profile_adapters=None,
 ) -> bool:
     """Run ONE due job end-to-end: execute → save output → deliver → mark.
 
@@ -6953,6 +7012,7 @@ def run_one_job(
                 loop=loop,
                 verbose=verbose,
                 extra_prompt=extra_prompt,
+                profile_adapters=profile_adapters,
                 fire_claim_lost=(
                     _CombinedCancelEvent(lost_ownership, cancel_event)
                     if cancel_event is not None
@@ -6977,6 +7037,7 @@ def _run_one_job_body(
     loop=None,
     verbose: bool = False,
     extra_prompt: Optional[str] = None,
+    profile_adapters=None,
     fire_claim_lost: Optional[_CancelEventLike] = None,
     execution_token: Optional[object] = None,
 ) -> bool:
@@ -7090,9 +7151,20 @@ def _fire_claim_ownership_lost() -> bool:
             # still triggers teardown before propagating.
             for _deferred_agent in _deferred_agents:
                 _teardown_cron_agent(_deferred_agent, job["id"])
-            raise
-        finally:
+            # Restore the previous secret scope before propagating: this
+            # failure path never reaches the delivery block whose finally
+            # resets it, so without this a leaked scope would contaminate the
+            # next job running on the same worker thread.
             reset_secret_scope(_scope_token)
+            raise
+        # NOTE: the profile secret scope is intentionally NOT reset here — it
+        # must stay installed through _deliver_result below, which reads the
+        # job profile's credentials via load_gateway_config -> _getenv (e.g.
+        # TELEGRAM_BOT_TOKEN). Resetting it before delivery made multiplexed
+        # jobs deliver with NO scope, so the owning profile's .env was ignored
+        # and delivery used the default profile's/empty token (wrong bot/chat,
+        # lost thread). The reset now happens in the delivery block's finally,
+        # once run AND delivery are both complete.
 
         if _fire_claim_ownership_lost():
             for _deferred_agent in _deferred_agents:
@@ -7270,6 +7342,7 @@ def _fire_claim_ownership_lost() -> bool:
                             deliver_content,
                             adapters=adapters,
                             loop=loop,
+                            profile_adapters=profile_adapters,
                         )
                 except Exception as de:
                     if isinstance(de, _FireClaimLostDuringSideEffect):
@@ -7284,6 +7357,12 @@ def _fire_claim_ownership_lost() -> bool:
             # their subprocesses/clients (#10200).
             for _deferred_agent in _deferred_agents:
                 _teardown_cron_agent(_deferred_agent, job["id"])
+            # Reset the profile secret scope now that run AND delivery are both
+            # complete. Kept installed through _deliver_result above so the
+            # job profile's credentials (e.g. TELEGRAM_BOT_TOKEN) resolve
+            # correctly under multiplex; torn down here so a scope never leaks
+            # into the next job on this worker thread.
+            reset_secret_scope(_scope_token)
 
         if side_effect_ownership_lost or _fire_claim_ownership_lost():
             # Same transport-cancel distinction as the pre-side-effect path:
@@ -7560,6 +7639,7 @@ def tick(
     sync: bool = True,
     *,
     can_dispatch=None,
+    profile_adapters=None,
 ):
     """
     Check and run all due jobs.
@@ -7779,6 +7859,7 @@ def _process_job(job: dict) -> bool:
                 adapters=adapters,
                 loop=loop,
                 verbose=verbose,
+                profile_adapters=profile_adapters,
             )
 
         # Partition due jobs: those with a per-job workdir mutate

cron/scheduler_provider.py

@@ -154,6 +154,7 @@ def fire_due(
         adapters: Any = None,
         loop: Any = None,
         force: bool = False,
+        profile_adapters: Any = None,
     ) -> bool:
         """Run a single job NOW via the shared orchestrator. Called by the
         inbound fire webhook when an external scheduler signals a job is due.
@@ -170,7 +171,10 @@ def fire_due(
         claimed_job = self.claim_fire(job_id, force=force)
         if claimed_job is None:
             return False
-        return self.fire_claimed(claimed_job, adapters=adapters, loop=loop)
+        return self.fire_claimed(
+            claimed_job, adapters=adapters, loop=loop,
+            profile_adapters=profile_adapters,
+        )
 
     def claim_fire(self, job_id: str, *, force: bool = False) -> dict | None:
         """Durably claim one fire and create its audit attempt before dispatch.
@@ -212,6 +216,7 @@ def fire_claimed(
         adapters: Any = None,
         loop: Any = None,
         cancel_event: Any = None,
+        profile_adapters: Any = None,
     ) -> bool:
         """Run an exact snapshot returned by ``claim_fire``.
 
@@ -227,6 +232,7 @@ def fire_claimed(
             adapters=adapters,
             loop=loop,
             cancel_event=cancel_event,
+            profile_adapters=profile_adapters,
         )
         return True
 
@@ -532,6 +538,7 @@ def start(
         interval=60,
         can_dispatch=None,
         profile_homes=None,
+        profile_adapters=None,
     ):
         import logging
         from cron.scheduler import tick as cron_tick
@@ -559,6 +566,7 @@ def start(
                 loop=loop,
                 interval=interval,
                 can_dispatch=can_dispatch,
+                profile_adapters=profile_adapters,
             )
             return
 
@@ -590,6 +598,7 @@ def start(
                         loop=loop,
                         sync=False,
                         can_dispatch=can_dispatch,
+                        profile_adapters=profile_adapters,
                     )
                 ok = True
             except BaseException as e:
@@ -630,6 +639,7 @@ def _start_multiplex(
         loop=None,
         interval=60,
         can_dispatch=None,
+        profile_adapters=None,
     ):
         """Tick every served profile's cron store when multiplex_profiles is on.
 
@@ -692,6 +702,7 @@ def _start_multiplex(
                                     loop=loop,
                                     sync=False,
                                     can_dispatch=can_dispatch,
+                                    profile_adapters=profile_adapters,
                                 )
                         finally:
                             reset_hermes_home_override(home_token)

gateway/run.py

@@ -31001,6 +31001,66 @@ def _env_home_value() -> Optional[str]:
     return False
 
 
+def _build_cron_start_kwargs(
+    runner: Any,
+    cron_provider: Any,
+    *,
+    multiplex_cron: bool = False,
+) -> Dict[str, Any]:
+    """Build the kwargs for ``cron_provider.start`` for the gateway ticker.
+
+    The shared defaults (``adapters`` + ``loop``) apply to every provider.
+    ``profile_adapters`` and ``profile_homes`` are multiplex-only additions
+    consumed exclusively by ``InProcessCronScheduler``; they are injected ONLY
+    when the resolved provider is the in-process ticker. External providers
+    (Chronos) implement ``start(stop_event, *, adapters, loop, interval)`` and
+    must never receive them — widening the generic provider call broke
+    external providers with a deterministic TypeError (review #83197).
+    """
+    from cron.scheduler_provider import InProcessCronScheduler
+
+    cron_start_kwargs: Dict[str, Any] = {
+        "adapters": runner.adapters,
+        "loop": asyncio.get_running_loop(),
+    }
+    if isinstance(cron_provider, InProcessCronScheduler):
+        # Multiplex profiles: expose the gateway's per-profile live-adapter
+        # map so cron delivery for a secondary-profile job routes through THAT
+        # profile's bot/chat (not the default profile's shared
+        # ``runner.adapters`` dict). Without this, a secondary-profile cron
+        # delivery picked the wrong adapter (or none) even when the owning
+        # profile's token resolved correctly. Deliberately scoped to the
+        # in-process provider: it is the ONLY provider whose ``start()``
+        # consumes ``profile_adapters`` (interface rule: start() signature
+        # growth must not break providers).
+        _profile_adapters = getattr(runner, "_profile_adapters", None)
+        if _profile_adapters:
+            cron_start_kwargs["profile_adapters"] = _profile_adapters
+        if multiplex_cron:
+            # Tell the built-in ticker which profile homes to tick so
+            # secondary-profile cron jobs actually fire (#69377).
+            try:
+                profile_homes = _multiplex_profile_homes(runner.config)
+                if profile_homes:
+                    cron_start_kwargs["profile_homes"] = profile_homes
+                    logger.info(
+                        "Cron scheduler will tick %d profile(s) under multiplex: %s",
+                        len(profile_homes),
+                        [p[0] if isinstance(p, tuple) else p for p in profile_homes],
+                    )
+            except Exception as exc:
+                logger.warning(
+                    "Could not resolve profile homes for multiplex cron: %s",
+                    exc,
+                )
+        # External cron providers own their remote scheduling contract. Only
+        # the in-process ticker polls local due jobs, so only it receives the
+        # local external-drain dispatch gate.
+        cron_start_kwargs["can_dispatch"] = lambda: not (
+            runner._draining or runner._external_drain_active
+        )
+    return cron_start_kwargs
+
 
 async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = False, verbosity: Optional[int] = 0) -> bool:
     """
@@ -31610,40 +31670,11 @@ def _request() -> None:
         resolve_cron_scheduler(),
         multiplex_profiles=multiplex_cron,
     )
-    cron_start_kwargs: Dict[str, Any] = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()}
-
-    # Multiplex profiles: tell the built-in ticker which profile homes to
-    # tick so secondary-profile cron jobs actually fire (#69377).
-    # Without this, only the process-global HERMES_HOME (default profile)
-    # is iterated and every secondary profile's cron store is silently
-    # ignored — jobs show as "scheduled" with a valid next_run_at but
-    # never execute because no ticker owns that store.
-    if (
-        isinstance(cron_provider, InProcessCronScheduler)
-        and multiplex_cron
-    ):
-        try:
-            profile_homes = _multiplex_profile_homes(runner.config)
-            if profile_homes:
-                cron_start_kwargs["profile_homes"] = profile_homes
-                logger.info(
-                    "Cron scheduler will tick %d profile(s) under multiplex: %s",
-                    len(profile_homes),
-                    [p[0] if isinstance(p, tuple) else p for p in profile_homes],
-                )
-        except Exception as exc:
-            logger.warning(
-                "Could not resolve profile homes for multiplex cron: %s",
-                exc,
-            )
-
-    # External cron providers own their remote scheduling contract. Only the
-    # in-process ticker polls local due jobs, so only it receives the local
-    # external-drain dispatch gate.
-    if isinstance(cron_provider, InProcessCronScheduler):
-        cron_start_kwargs["can_dispatch"] = lambda: not (
-            runner._draining or runner._external_drain_active
-        )
+    cron_start_kwargs = _build_cron_start_kwargs(
+        runner,
+        cron_provider,
+        multiplex_cron=multiplex_cron,
+    )
     cron_thread = threading.Thread(
         target=cron_provider.start,
         args=(cron_stop,),

plugins/cron_providers/chronos/__init__.py

@@ -239,13 +239,15 @@ def fire_claimed(
         adapters: Any = None,
         loop: Any = None,
         cancel_event: Any = None,
+        profile_adapters: Any = None,
     ) -> bool:
         job_id = claimed_job["id"]
         ran = super().fire_claimed(
             claimed_job,
             adapters=adapters,
             loop=loop,
             cancel_event=cancel_event,
+            profile_adapters=profile_adapters,
         )
         if ran:
             from cron.jobs import get_job