fix: multiplex cron delivery routes through correct profile bot

cron/scheduler.py

@@ -1447,7 +1447,8 @@ 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, profile_adapters=None, loop=None) -> Optional[str]:
+
     """
     Deliver job output to the configured target(s) (origin chat, specific platform, etc.).
 
@@ -1577,6 +1578,36 @@ 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)
+        # Cross-profile cron delivery: when running under multiplex, resolve
+        # the adapter from the profile that owns this cron job instead of
+        # defaulting to the main profile's adapter (#cross-profile-cron-delivery).
+        if profile_adapters:
+            try:
+                from hermes_cli.profiles import get_profile_dir
+                _profile_home = job.get("_profile_home")
+                if _profile_home:
+                    for pname, padapters in profile_adapters.items():
+                        if _profile_home == str(get_profile_dir(pname)):
+                            padapter = padapters.get(platform)
+                            if padapter is not None:
+                                from gateway.delivery import DeliveryTransport
+                                pconfig = padapter.config if hasattr(padapter, 'config') else None
+                                runtime_adapter = padapter
+                                transport = DeliveryTransport(
+                                    adapter=padapter,
+                                    config=pconfig,
+                                    transport_platform=platform,
+                                )
+                                # Swap adapters dict so DeliveryRouter uses the
+                                # profile-specific bot instead of the default one.
+                                adapters = padapters
+                                logger.info(
+                                    "Job '%s': using profile '%s' adapter for %s delivery",
+                                    job.get("id", "?"), pname, platform_name,
+                                )
+                            break
+            except Exception:
+                pass
         if transport is not None:
             pconfig = transport.config
             runtime_adapter = transport.adapter
@@ -3875,7 +3906,7 @@ def _teardown_cron_agent(agent, job_id: str) -> None:
         logger.debug("Job '%s': failed to reap stale auxiliary clients: %s", job_id, e)
 
 
-def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -> bool:
+def run_one_job(job: dict, *, adapters=None, profile_adapters=None, loop=None, verbose: bool = False) -> bool:
     """Run ONE due job end-to-end: execute → save output → deliver → mark.
 
     This is the shared firing body extracted from ``tick``'s per-job closure so
@@ -4008,7 +4039,7 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
                     and not _resolve_delivery_targets(job)
                 )
                 try:
-                    delivery_error = _deliver_result(job, deliver_content, adapters=adapters, loop=loop)
+                    delivery_error = _deliver_result(job, deliver_content, adapters=adapters, profile_adapters=profile_adapters, loop=loop)
                 except Exception as de:
                     delivery_error = str(de)
                     logger.error("Delivery failed for job %s: %s", job["id"], de)
@@ -4103,6 +4134,7 @@ def tick(
     sync: bool = True,
     *,
     can_dispatch=None,
+    profile_adapters=None,
 ):
     """
     Check and run all due jobs.
@@ -4144,6 +4176,14 @@ def tick(
 
         due_jobs = get_due_jobs()
 
+        # Stamp each job with the profile home so _deliver_result can route
+        # through the correct profile's adapter instead of defaulting to the
+        # main profile's bot (#cross-profile-cron-delivery).
+        _cron_profile_home = str(_get_hermes_home())
+        for _dj in due_jobs:
+            if "_profile_home" not in _dj:
+                _dj["_profile_home"] = _cron_profile_home
+
         if verbose and not due_jobs:
             logger.info("%s - No jobs due", _hermes_now().strftime('%H:%M:%S'))
             return 0
@@ -4191,7 +4231,7 @@ 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, profile_adapters=profile_adapters, loop=loop, verbose=verbose)
 
         # Partition due jobs: those with a per-job workdir mutate
         # os.environ["TERMINAL_CWD"] inside run_job, which is process-global, so

cron/scheduler_provider.py

@@ -182,6 +182,7 @@ def start(
         interval=60,
         can_dispatch=None,
         profile_homes=None,
+        profile_adapters=None,
     ):
         import logging
         from cron.scheduler import tick as cron_tick
@@ -206,6 +207,7 @@ def start(
                 stop_event,
                 profile_homes=profile_homes,
                 adapters=adapters,
+                profile_adapters=profile_adapters,
                 loop=loop,
                 interval=interval,
                 can_dispatch=can_dispatch,
@@ -266,6 +268,7 @@ def _start_multiplex(
         *,
         profile_homes,
         adapters=None,
+        profile_adapters=None,
         loop=None,
         interval=60,
         can_dispatch=None,
@@ -326,6 +329,7 @@ def _start_multiplex(
                                 cron_tick(
                                     verbose=False,
                                     adapters=adapters,
+                                    profile_adapters=profile_adapters,
                                     loop=loop,
                                     sync=False,
                                     can_dispatch=can_dispatch,

gateway/run.py

@@ -20844,13 +20844,35 @@ def _build_process_event_source(self, evt: dict):
             )
             return None
 
+        # Extract profile from session_key so that background-process
+        # notifications (async delegation completions, watch patterns)
+        # are routed through the correct profile's adapter in multiplex
+        # mode.  Without this, completions from non-default profiles
+        # (e.g. CONTENT, CHASE) always fall back to the default adapter
+        # and are delivered through the wrong bot.
+        _derived_profile: Optional[str] = None
+        if session_key:
+            try:
+                # Extract profile namespace from session key.
+                # Keys follow agent:{namespace}:{platform}:... where namespace
+                # is "main" (default) or the profile name (e.g. "content").
+                _parts = str(session_key).split(":")
+                if len(_parts) >= 2 and _parts[0] == "agent":
+                    _ns = _parts[1] or "main"
+                    _derived_profile = None if _ns == "main" else _ns
+            except Exception:
+                pass
+        if _derived_profile == "default":
+            _derived_profile = None
+
         return SessionSource(
             platform=platform,
             chat_id=chat_id,
             chat_type=chat_type,
             thread_id=str(evt.get("thread_id") or "").strip() or None,
             user_id=str(evt.get("user_id") or "").strip() or None,
             user_name=str(evt.get("user_name") or "").strip() or None,
+            profile=_derived_profile,
         )
 
     async def _inject_watch_notification(
@@ -20910,11 +20932,21 @@ async def _inject_watch_notification(
             )
             return None
         platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform)
-        adapter = None
-        for p, a in self.adapters.items():
-            if p.value == platform_name:
-                adapter = a
-                break
+        # Use profile-aware adapter resolution so that background-process
+        # notifications (delegation completions, watch patterns) from
+        # non-default profiles are delivered through the correct bot.
+        # _adapter_for_source checks _registered_transport_adapter first,
+        # then _authorization_adapter which uses _profile_adapters for
+        # secondary profiles — matching the same routing path as live
+        # inbound messages.
+        adapter = self._adapter_for_source(source)
+        if adapter is None:
+            # Fallback: try self.adapters (default profile) for platforms
+            # where _adapter_for_source returned None (e.g. relay).
+            for p, a in self.adapters.items():
+                if p.value == platform_name:
+                    adapter = a
+                    break
         if not adapter:
             return None
         from gateway.wake import adapter_supports_push as _wake_push_ok
@@ -21383,11 +21415,15 @@ async def _run_process_watcher(self, watcher: dict) -> None:
                         f"[Background process {session_id} finished with exit code {session.exit_code}~ "
                         f"Here's the final output:\n{new_output}]"
                     )
-                    adapter = None
-                    for p, a in self.adapters.items():
-                        if p.value == platform_name:
-                            adapter = a
-                            break
+                    # Profile-aware adapter resolution for background process
+                    # completion notifications — matches the same routing as
+                    # _inject_watch_notification and live inbound messages.
+                    adapter = self._adapter_for_source(source) if source else None
+                    if adapter is None:
+                        for p, a in self.adapters.items():
+                            if p.value == platform_name:
+                                adapter = a
+                                break
                     if adapter and chat_id:
                         try:
                             send_meta = {"thread_id": thread_id} if thread_id else None
@@ -21413,11 +21449,13 @@ async def _run_process_watcher(self, watcher: dict) -> None:
                     f"[Background process {session_id} is still running~ "
                     f"New output:\n{new_output}]"
                 )
-                adapter = None
-                for p, a in self.adapters.items():
-                    if p.value == platform_name:
-                        adapter = a
-                        break
+                # Profile-aware adapter resolution for running-process updates.
+                adapter = self._adapter_for_source(source) if source else None
+                if adapter is None:
+                    for p, a in self.adapters.items():
+                        if p.value == platform_name:
+                            adapter = a
+                            break
                 if adapter and chat_id:
                     try:
                         send_meta = {"thread_id": thread_id} if thread_id else None
@@ -25541,6 +25579,11 @@ def restart_signal_handler():
     cron_stop = threading.Event()
     cron_provider = resolve_cron_scheduler()
     cron_start_kwargs: Dict[str, Any] = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()}
+    # Pass profile-specific adapters so multiplexed profile crons deliver
+    # through the right bot (#cross-profile-cron-delivery).
+    profile_adapters = getattr(runner, "_profile_adapters", None)
+    if profile_adapters:
+        cron_start_kwargs["profile_adapters"] = profile_adapters
 
     # Multiplex profiles: tell the built-in ticker which profile homes to
     # tick so secondary-profile cron jobs actually fire (#69377).

gateway/session.py

@@ -1550,8 +1550,28 @@ def _recovered_row_allowed_for_active_profile(
         requested_session_key: str,
         recovered: Dict[str, Any],
     ) -> bool:
-        """Prevent non-multiplexed gateways from reviving another profile's row."""
+        """Prevent non-multiplexed gateways from reviving another profile's row.
+
+        When multiplexing is on, the fallback session-recovery query matches
+        on (source, user_id, chat_id, chat_type, thread_id) WITHOUT a profile
+        namespace filter, so a session created by one profile can be recovered
+        by another.  Guard against this by checking that the recovered row's
+        profile namespace matches the requested key's profile namespace.
+        """
         if getattr(self.config, "multiplex_profiles", False):
+            recovered_key = str(recovered.get("session_key") or "")
+            if recovered_key and recovered_key != requested_session_key:
+                recovered_profile = self._profile_from_session_key(recovered_key)
+                requested_profile = self._profile_from_session_key(requested_session_key)
+                if recovered_profile != requested_profile:
+                    logger.warning(
+                        "Gateway session DB recovery ignored %s for %s because "
+                        "multiplex_profiles is on and profiles don't match "
+                        "(recovered=%s, requested=%s)",
+                        recovered_key, requested_session_key,
+                        recovered_profile, requested_profile,
+                    )
+                    return False
             return True
 
         recovered_key = str(recovered.get("session_key") or "")