diff --git a/cron/scheduler.py b/cron/scheduler.py index 77c2772762238..fade2048c2379 100644 --- a/cron/scheduler.py +++ b/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 diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index 4c20db01c6d42..0e712e586a019 100644 --- a/cron/scheduler_provider.py +++ b/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, diff --git a/gateway/run.py b/gateway/run.py index 4870a187cfc9e..eb873320dff68 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -20844,6 +20844,27 @@ 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, @@ -20851,6 +20872,7 @@ def _build_process_event_source(self, evt: dict): 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). diff --git a/gateway/session.py b/gateway/session.py index 144de458f71d0..2f46e301a7e5e 100644 --- a/gateway/session.py +++ b/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 "")