From 8091c4405438ce84b8f727c1de29de189f0b4d73 Mon Sep 17 00:00:00 2001
From: liuhao1024 <sunsky.lau@gmail.com>
Date: Wed, 8 Jul 2026 14:54:46 +0800
Subject: [PATCH] fix(gateway): install _profile_runtime_scope in
 _run_background_task when multiplexing is active

When multiplex_profiles is true, background tasks spawned by /background
command failed with UnscopedSecretError because _resolve_session_agent_runtime()
was called without a profile secret scope. This fix wraps the task in
_profile_runtime_scope, mirroring the pattern used by _run_agent.

Fixes #60726
---
 gateway/run.py                                | 27 +++++++
 .../test_multiplex_background_task_scope.py   | 74 +++++++++++++++++++
 2 files changed, 101 insertions(+)
 create mode 100644 tests/gateway/test_multiplex_background_task_scope.py

diff --git a/gateway/run.py b/gateway/run.py
index 6cf4566e044d..f711e69c8957 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -13763,6 +13763,33 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         event_message_id: Optional[str] = None,
         media_urls: Optional[List[str]] = None,
         media_types: Optional[List[str]] = None,
+    ) -> None:
+        """Profile-scoping wrapper around the background agent task.
+
+        When multiplexing is active, resolve the inbound source's profile and
+        run the whole task inside ``_profile_runtime_scope`` so credentials
+        resolve from that profile's secret scope. Mirrors the pattern in
+        ``_run_agent``.
+        """
+        if not getattr(getattr(self, "config", None), "multiplex_profiles", False):
+            return await self._run_background_task_inner(
+                prompt, source, task_id, event_message_id, media_urls, media_types,
+            )
+
+        profile_home = self._resolve_profile_home_for_source(source)
+        with _profile_runtime_scope(profile_home):
+            return await self._run_background_task_inner(
+                prompt, source, task_id, event_message_id, media_urls, media_types,
+            )
+
+    async def _run_background_task_inner(
+        self,
+        prompt: str,
+        source: "SessionSource",
+        task_id: str,
+        event_message_id: Optional[str] = None,
+        media_urls: Optional[List[str]] = None,
+        media_types: Optional[List[str]] = None,
     ) -> None:
         """Execute a background agent task and deliver the result to the chat."""
         from run_agent import AIAgent
diff --git a/tests/gateway/test_multiplex_background_task_scope.py b/tests/gateway/test_multiplex_background_task_scope.py
new file mode 100644
index 000000000000..fbafb6fe6cc8
--- /dev/null
+++ b/tests/gateway/test_multiplex_background_task_scope.py
@@ -0,0 +1,74 @@
+"""Regression: background tasks respect profile secret scope when multiplexing.
+
+Issue #60726: /background command runs _run_background_task without a profile
+scope, causing UnscopedSecretError when multiplexing is active and credentials
+are profile-scoped.
+"""
+import pytest
+from unittest import mock
+
+
+class TestBackgroundTaskProfileScope:
+    """_run_background_task installs _profile_runtime_scope when multiplexing is active."""
+
+    def test_background_task_calls_inner_wrapped_in_scope_when_multiplex_active(self):
+        """When multiplex_profiles is True, _run_background_task wraps call in _profile_runtime_scope."""
+        from gateway.run import GatewayRunner
+
+        config = {"multiplex_profiles": True}
+        gw = GatewayRunner(config=config)
+        gw._session_db = mock.MagicMock()
+        gw._adapter_for_source = mock.MagicMock(return_value=mock.MagicMock())
+
+        mock_inner = mock.AsyncMock(return_value=None)
+        gw._run_background_task_inner = mock_inner
+
+        import asyncio
+        source = mock.MagicMock()
+        source.profile_home = "/fake/profile"
+
+        # Mock _profile_runtime_scope
+        from gateway.run import _profile_runtime_scope
+        with mock.patch("gateway.run._profile_runtime_scope") as mock_scope:
+            mock_scope.return_value.__enter__ = mock.MagicMock()
+            mock_scope.return_value.__exit__ = mock.MagicMock()
+
+            asyncio.run(
+                gw._run_background_task(
+                    prompt="test",
+                    source=source,
+                    task_id="test_task",
+                )
+            )
+
+            # _profile_runtime_scope should have been called with profile_home
+            mock_scope.assert_called_once_with("/fake/profile")
+            mock_inner.assert_called_once()
+
+    def test_background_task_calls_inner_direct_when_multiplex_disabled(self):
+        """When multiplex_profiles is False, _run_background_task calls inner directly."""
+        from gateway.run import GatewayRunner
+
+        config = {"multiplex_profiles": False}
+        gw = GatewayRunner(config=config)
+        gw._session_db = mock.MagicMock()
+        gw._adapter_for_source = mock.MagicMock(return_value=mock.MagicMock())
+
+        mock_inner = mock.AsyncMock(return_value=None)
+        gw._run_background_task_inner = mock_inner
+
+        import asyncio
+        source = mock.MagicMock()
+
+        with mock.patch("gateway.run._profile_runtime_scope") as mock_scope:
+            asyncio.run(
+                gw._run_background_task(
+                    prompt="test",
+                    source=source,
+                    task_id="test_task",
+                )
+            )
+
+            # _profile_runtime_scope should NOT have been called
+            mock_scope.assert_not_called()
+            mock_inner.assert_called_once()
\ No newline at end of file