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
This commit is contained in:
liuhao1024 2026-07-08 14:54:46 +08:00 committed by Teknium
parent 58010c8b3d
commit 8091c44054
2 changed files with 101 additions and 0 deletions

View file

@ -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

View file

@ -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()