From b24a741e9255bb11708e161ebbbc9a004efa0a03 Mon Sep 17 00:00:00 2001 From: xbrxr03 Date: Fri, 7 Aug 2026 03:13:46 -0400 Subject: [PATCH 1/4] fix: multiplex cron delivery routes through correct profile bot When multiplex_profiles is enabled, all cron job deliveries route through the default profile's Telegram adapter regardless of which profile owns the cron job. This causes SCOUT's hourly reports to appear in JARVIS's chat, CHASE's EOD briefs to leak into the wrong DM, etc. Root cause: resolve_delivery_transport() in _deliver_result() receives only the default profile's dict. The per-profile adapters (_profile_adapters on GatewayRunner) are never passed through the cron delivery chain, so every delivery resolves to the main profile's bot. Fix: - gateway/run.py: pass runner._profile_adapters to the cron scheduler - cron/scheduler_provider.py: thread profile_adapters through start() and _start_multiplex() to cron_tick() - cron/scheduler.py: - Add profile_adapters param to tick(), run_one_job(), _deliver_result() - Stamp each job with _profile_home (the profile's HERMES_HOME) in tick() before dispatching to ThreadPoolExecutor, since ContextVars don't propagate to worker threads in Python 3.11 - In _deliver_result(), when profile_adapters is provided, match the job's _profile_home to the correct profile's adapter and construct a DeliveryTransport that routes through that profile's bot Verified: agent.log confirms 'Job X: using profile Y adapter for telegram delivery' for chase and outreach crons. Messages now appear in the correct profile's Telegram chat. --- cron/scheduler.py | 44 ++++++++++++++++++++++++++++++++++---- cron/scheduler_provider.py | 4 ++++ gateway/run.py | 5 +++++ 3 files changed, 49 insertions(+), 4 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index 77c2772762238..e1deee6a0e691 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,32 @@ 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 + transport = DeliveryTransport( + adapter=padapter, + config=pconfig, + transport_platform=platform, + ) + 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 +3902,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 +4035,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 +4130,7 @@ def tick( sync: bool = True, *, can_dispatch=None, + profile_adapters=None, ): """ Check and run all due jobs. @@ -4144,6 +4172,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 +4227,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..ebdf6d195393b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -25541,6 +25541,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). From 6df70f23ca2a6456429c16094708c5ff5db5094e Mon Sep 17 00:00:00 2001 From: xbrxr03 Date: Fri, 7 Aug 2026 03:13:46 -0400 Subject: [PATCH 2/4] fix: multiplex cron delivery routes through correct profile bot When multiplex_profiles is enabled, all cron job deliveries route through the default profile's Telegram adapter regardless of which profile owns the cron job. This causes SCOUT's hourly reports to appear in JARVIS's chat, CHASE's EOD briefs to leak into the wrong DM, etc. Root cause: resolve_delivery_transport() in _deliver_result() receives only the default profile's dict. The per-profile adapters (_profile_adapters on GatewayRunner) are never passed through the cron delivery chain, so every delivery resolves to the main profile's bot. Fix: - gateway/run.py: pass runner._profile_adapters to the cron scheduler - cron/scheduler_provider.py: thread profile_adapters through start() and _start_multiplex() to cron_tick() - cron/scheduler.py: - Add profile_adapters param to tick(), run_one_job(), _deliver_result() - Stamp each job with _profile_home (the profile's HERMES_HOME) in tick() before dispatching to ThreadPoolExecutor, since ContextVars don't propagate to worker threads in Python 3.11 - In _deliver_result(), when profile_adapters is provided, match the job's _profile_home to the correct profile's adapter and construct a DeliveryTransport that routes through that profile's bot Verified: agent.log confirms 'Job X: using profile Y adapter for telegram delivery' for chase and outreach crons. Messages now appear in the correct profile's Telegram chat. --- cron/scheduler.py | 44 ++++++++++++++++++++++++++++++++++---- cron/scheduler_provider.py | 4 ++++ gateway/run.py | 5 +++++ 3 files changed, 49 insertions(+), 4 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index 77c2772762238..e1deee6a0e691 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,32 @@ 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 + transport = DeliveryTransport( + adapter=padapter, + config=pconfig, + transport_platform=platform, + ) + 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 +3902,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 +4035,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 +4130,7 @@ def tick( sync: bool = True, *, can_dispatch=None, + profile_adapters=None, ): """ Check and run all due jobs. @@ -4144,6 +4172,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 +4227,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..ebdf6d195393b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -25541,6 +25541,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). From 9bb9a1ef2b960037ddf276c791a9541425c3d754 Mon Sep 17 00:00:00 2001 From: xbrxr03 Date: Sat, 8 Aug 2026 00:58:26 -0400 Subject: [PATCH 3/4] fix: also swap runtime_adapter and adapters dict in cron delivery Previous patch set transport and pconfig but left runtime_adapter and the adapters dict pointing to the default profile. DeliveryRouter uses self.adapters to find the platform bot, so it still sent through JARVIS's bot. Now swaps adapters to the profile's dict and sets runtime_adapter to the profile's adapter. --- cron/scheduler.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/cron/scheduler.py b/cron/scheduler.py index e1deee6a0e691..fade2048c2379 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1592,11 +1592,15 @@ def _deliver_result(job: dict, content: str, adapters=None, profile_adapters=Non 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, From 6ab9a1040514c58d5ce9c35e2c687c015172fcc6 Mon Sep 17 00:00:00 2001 From: xbrxr03 Date: Sat, 15 Aug 2026 16:16:00 -0400 Subject: [PATCH 4/4] fix: profile-aware routing for background process notifications Two bugs fixed: 1. _build_process_event_source: extract profile from session_key so delegation completions and watch patterns from non-default profiles (CONTENT, CHASE, etc.) carry source.profile for correct adapter routing. Previously, background-process SessionSource objects had no profile, causing _adapter_for_source to fall back to the default adapter. 2. _inject_watch_notification + _run_process_watcher: replace self.adapters iteration with _adapter_for_source(source) for profile-aware adapter resolution. Without this, delegation completions from secondary profiles always route through the default (JARVIS) bot. Same root cause as the cron delivery bug (PR #80876): multiplex profiles all share self.adapters, so any code path that iterates self.adapters instead of using _adapter_for_source sends through the wrong bot. --- gateway/run.py | 68 ++++++++++++++++++++++++++++++++++++---------- gateway/session.py | 22 ++++++++++++++- 2 files changed, 74 insertions(+), 16 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index ebdf6d195393b..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 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 "")