feat(agent): isolate interactive execution from background services (#7083)

2026-11-03 22:04:57 +08:00 · 2026-10-19 18:12:05 +08:00
parent 8b5e536452
commit c85f9b262f
29 changed files with 980 additions and 38 deletions
--- a/app/adapters/web/priority.py
+++ b/app/adapters/web/priority.py
@@ -0,0 +1,31 @@
+"""
+为宿主 Agent 的本机 HTTP 调用恢复临时执行归属。
+"""
+
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+from app.runtime.priority import PRIORITY_HEADER, propagated_execution
+
+
+class ExecutionPriorityMiddleware:
+    """
+    仅凭本进程登记的临时随机凭据恢复优先级，正常鉴权仍由下游负责。
+    """
+
+    def __init__(self, app: ASGIApp) -> None:
+        """
+        保存要恢复执行上下文的下游应用。
+        """
+        self._app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        """
+        覆盖完整 HTTP 响应生命周期，其他连接类型直接转交下游。
+        """
+        if scope.get("type") != "http":
+            await self._app(scope, receive, send)
+            return
+        headers = dict(scope.get("headers") or [])
+        key = headers.get(PRIORITY_HEADER.lower().encode("ascii"), b"")
+        with propagated_execution(key.decode("ascii", errors="ignore")):
+            await self._app(scope, receive, send)
--- a/app/agent/api/executor.py
+++ b/app/agent/api/executor.py
@@ -20,6 +20,7 @@ from app.agent.policy.contracts import ActionEffect
 from app.agent.policy.sanitizer import stable_type_name
 from app.application.security.token import create_access_token
 from app.runtime.log import logger
+from app.runtime.priority import PRIORITY_HEADER, interactive_key
 from app.runtime.settings import get_runtime_setting

 _DEFAULT_API_REQUEST_TIMEOUT_SECONDS = 30
@@ -156,6 +157,9 @@ class MoviePilotApiExecutor:
            headers["X-MoviePilot-Agent-Channel"] = quote(self._context.channel, safe="")
        if self._context.source:
            headers["X-MoviePilot-Agent-Source"] = quote(self._context.source, safe="")
+        execution_key = interactive_key()
+        if execution_key:
+            headers[PRIORITY_HEADER] = execution_key
        return headers

    @classmethod
--- a/app/agent/middleware/config.py
+++ b/app/agent/middleware/config.py
@@ -2,6 +2,7 @@
 动态注入 Agent 根层运行时配置的中间件。
 """

+import asyncio
 from collections.abc import Awaitable, Callable

 from langchain.agents.middleware.types import (
@@ -14,6 +15,7 @@ from langchain.agents.middleware.types import (

 from app.agent.middleware.utils import append_to_system_message
 from app.agent.runtime import agent_runtime_manager
+from app.runtime.settings import get_runtime_setting


 class RuntimeConfigMiddleware(AgentMiddleware[dict, ContextT, ResponseT]):  # noqa
@@ -45,7 +47,9 @@ class RuntimeConfigMiddleware(AgentMiddleware[dict, ContextT, ResponseT]):  # no
        """
        在异步模型调用前注入当前运行配置，并返回模型执行结果。
        """
-        return await handler(self.modify_request(request))
+        timeout = float(get_runtime_setting("AGENT_LLM_TIMEOUT", 180))
+        async with asyncio.timeout(timeout if timeout > 0 else None):
+            return await handler(self.modify_request(request))


 __all__ = ["RuntimeConfigMiddleware"]
--- a/app/agent/orchestrator.py
+++ b/app/agent/orchestrator.py
@@ -102,6 +102,7 @@ from app.runtime.events import eventmanager
 from app.runtime.execution import run_in_threadpool
 from app.runtime.log import logger
 from app.runtime.observability import record_metric
+from app.runtime.priority import interactive_execution
 from app.runtime.settings import get_runtime_setting
 from app.schemas.event import AgentLLMProviderEventData, AgentTokensUsageEventData
 from app.schemas.message import Message
@@ -1643,6 +1644,8 @@ class MoviePilotAgent:
        将 Agent 执行异常转换为用户可读消息。
        回复只携带主错误信息，完整 traceback 保留在日志中排查。
        """
+        if isinstance(error, TimeoutError):
+            return "本次请求执行超时，请稍后重试。已开始的后台操作可能仍在收尾，请先确认状态，避免重复执行。"
        message = cls._primary_exception_message(error)
        if not message:
            return AGENT_EXECUTION_ERROR_MESSAGE
@@ -2211,7 +2214,9 @@ class MoviePilotAgent:
            self._scheduled_terminal_scopes.add(scope)
        steering_token = bind_steering_inbox(self._steering_inbox)
        try:
-            with bind_terminal_scope(scope):
+            with bind_terminal_scope(scope), interactive_execution(
+                enabled=not self.is_background and scope.kind != "scheduled"
+            ):
                return await self._process(
                    message, images=images, files=files, has_audio_input=has_audio_input
                )
--- a/app/agent/tools/base.py
+++ b/app/agent/tools/base.py
@@ -31,6 +31,7 @@ from app.application.notification import get_notification_configs
 from app.chain.base import ChainBase
 from app.foundation.identity import build_user_memory_key
 from app.runtime.log import logger
+from app.runtime.priority import is_interactive
 from app.runtime.settings import get_runtime_setting
 from app.schemas.message import Message
 from app.schemas.types import MessageType, NotificationChannel
@@ -299,7 +300,8 @@ _BLOCKING_BUCKET_LIMITS = {
 }
 _blocking_semaphores = {
    bucket: asyncio.Semaphore(limit)
-    for bucket, limit in _BLOCKING_BUCKET_LIMITS.items()
+    for name, limit in _BLOCKING_BUCKET_LIMITS.items()
+    for bucket in (name, f"interactive:{name}")
 }
 _blocking_executors: dict[str, ThreadPoolExecutor] = {}
 _blocking_retiring_executors: set[ThreadPoolExecutor] = set()
@@ -333,7 +335,7 @@ def _submit_blocking_call(
            raise RuntimeError("Agent 工具阻塞执行器正在关闭，不能再提交新任务")
        executor = _blocking_executors.get(bucket)
        if executor is None:
-            limit = _BLOCKING_BUCKET_LIMITS[bucket]
+            limit = _BLOCKING_BUCKET_LIMITS[bucket.removeprefix("interactive:")]
            executor = ThreadPoolExecutor(
                max_workers=limit,
                thread_name_prefix=f"agent-tool-{bucket}",
@@ -469,6 +471,8 @@ async def run_agent_blocking(
    新任务，把同一类慢 IO 的线程池持续打满。
    """
    bucket_name = bucket if bucket in _BLOCKING_BUCKET_LIMITS else "default"
+    if is_interactive():
+        bucket_name = f"interactive:{bucket_name}"
    semaphore = _blocking_semaphores[bucket_name]
    bound_call = partial(func, *args, **kwargs)
    loop = asyncio.get_running_loop()
@@ -481,6 +485,9 @@ async def run_agent_blocking(
        raise

    def _release_semaphore(_future: ConcurrentFuture[object]) -> None:
+        """
+        同步工具真实结束后才归还所属执行组的并发额度。
+        """
        try:
            _future.exception()
        except Exception:
--- a/app/application/messaging/chat.py
+++ b/app/application/messaging/chat.py
@@ -15,6 +15,7 @@ from app.application.database import (
 )
 from app.application.messaging.recall import RecallService
 from app.runtime.observability import record_metric
+from app.runtime.priority import INTERACTIVE_DATABASE_CAPACITY, is_interactive
 from app.schemas.agent import AgentChatSessionDetail, AgentChatSessionSummary
 from app.schemas.exception import AgentChatPersistenceUnavailableError

@@ -523,6 +524,7 @@ class AgentChatPersistenceService:
        self._capacity = capacity
        self._session_capacity = session_capacity
        self._pending_writes = 0
+        self._pending_interactive_writes = 0
        self._pending_by_session: dict[str, int] = {}
        self._active_tasks: set[asyncio.Task[object]] = set()
        self._closing = False
@@ -561,16 +563,21 @@ class AgentChatPersistenceService:
                "AgentChat 持久化服务当前不可接收任务"
            )
        session_pending = self._pending_by_session.get(session_id, 0)
+        interactive = is_interactive()
+        pending = self._pending_interactive_writes if interactive else self._pending_writes - self._pending_interactive_writes
+        capacity = INTERACTIVE_DATABASE_CAPACITY if interactive else self._capacity
        if (
-            self._pending_writes >= self._capacity
+            pending >= capacity
            or session_pending >= self._session_capacity
        ):
            record_metric("agent.chat.persistence.rejected")
            raise AgentChatPersistenceUnavailableError(
-                f"AgentChat 写入容量已用尽（全局上限 {self._capacity}，"
+                f"AgentChat 写入容量已用尽（当前执行组上限 {capacity}，"
                f"单会话上限 {self._session_capacity}）"
            )
        self._pending_writes += 1
+        if interactive:
+            self._pending_interactive_writes += 1
        self._pending_by_session[session_id] = session_pending + 1
        current = asyncio.current_task()
        if current is not None:
@@ -589,6 +596,8 @@ class AgentChatPersistenceService:
                await self._async_executor.run(execute)
        finally:
            self._pending_writes -= 1
+            if interactive:
+                self._pending_interactive_writes -= 1
            remaining = self._pending_by_session.get(session_id, 1) - 1
            if remaining:
                self._pending_by_session[session_id] = remaining
--- a/app/application/scheduling.py
+++ b/app/application/scheduling.py
@@ -78,6 +78,7 @@ class JobSpec:
            "name": self.name,
            "func": self.func,
            "owner": self.owner,
+            "execution_group": "agent" if self.owner == "agent" else "default",
            "overlap": self.overlap.value,
            "timeout_seconds": self.timeout_seconds,
            "manual": self.manual,
--- a/app/application/settings/contract.py
+++ b/app/application/settings/contract.py
@@ -555,7 +555,7 @@ def _runtime_dependencies(key: str) -> tuple[str, ...]:
        return ("AI_AGENT_ENABLE", "LLM_SUPPORT_AUDIO_INPUT")
    if key.startswith("AUDIO_OUTPUT_"):
        return ("AI_AGENT_ENABLE", "LLM_SUPPORT_AUDIO_OUTPUT")
-    if key.startswith("LLM_") or key.startswith("AI_AGENT_"):
+    if key.startswith(("LLM_", "AI_AGENT_", "AGENT_")):
        return () if key == "AI_AGENT_ENABLE" else ("AI_AGENT_ENABLE",)
    if key.startswith("DB_POSTGRESQL_") or key.startswith("DB_SQLITE_"):
        return ("DB_TYPE",)
--- a/app/application/settings/resources/system_settings_catalog.json
+++ b/app/application/settings/resources/system_settings_catalog.json
@@ -13,6 +13,24 @@
      "source_file": "app/runtime/config.py",
      "source_line": 410
    },
+    "AGENT_BACKGROUND_DEFER_SECONDS": {
+      "description": "交互活跃时非关键后台服务最多避让的秒数，0为不避让；不暂停已开始的任务",
+      "group": "ai_agent",
+      "source_file": "app/runtime/config.py",
+      "source_line": 713
+    },
+    "AGENT_INTERACTIVE_PRIORITY": {
+      "description": "为交互式智能助手隔离执行容量，并让新启动的非关键后台服务有限避让",
+      "group": "ai_agent",
+      "source_file": "app/runtime/config.py",
+      "source_line": 712
+    },
+    "AGENT_LLM_TIMEOUT": {
+      "description": "Agent单次模型请求的总超时秒数，包含供应商重试；0为不限制",
+      "group": "ai_agent",
+      "source_file": "app/runtime/config.py",
+      "source_line": 714
+    },
    "AI_AGENT_ENABLE": {
      "description": "AI智能体开关",
      "group": "ai_agent",
@@ -1297,6 +1315,12 @@
      "source_file": "app/runtime/config.py",
      "source_line": 757
    },
+    "SCHEDULER_JOB_TIMEOUT": {
+      "description": "协程调度任务默认执行超时秒数，0为不限制；不强行终止同步线程",
+      "group": "performance",
+      "source_file": "app/runtime/config.py",
+      "source_line": 715
+    },
    "SCRAP_FOLLOW_TMDB": {
      "description": "新增已入库媒体是否跟随TMDB信息变化",
      "group": "transfer",
--- a/app/db/worker.py
+++ b/app/db/worker.py
@@ -11,9 +11,14 @@ import time
 from concurrent.futures import Future, ThreadPoolExecutor
 from contextvars import copy_context
 from dataclasses import dataclass
-from typing import Callable, TypeVar, cast
+from typing import Callable, Optional, TypeVar, cast

 from app.runtime.observability import record_metric
+from app.runtime.priority import (
+    INTERACTIVE_DATABASE_CAPACITY,
+    INTERACTIVE_DATABASE_WORKERS,
+    is_interactive,
+)
 from app.schemas.exception import (
    DatabaseWorkerClosedError,
    DatabaseWorkerOverloadedError,
@@ -28,7 +33,7 @@ DATABASE_WORKER_CAPACITY = 32
@dataclass(frozen=True, slots=True)
 class DatabaseWorkerStats:
    """
-    数据库执行器当前的容量和任务数量快照。
+    单个数据库执行组当前的容量和任务数量快照。

    :param max_workers: 最大工作线程数
    :param capacity: 包含排队和运行中任务的总容量上限
@@ -69,9 +74,14 @@ class DatabaseWorker:
        *,
        max_workers: int = DATABASE_WORKER_MAX_WORKERS,
        capacity: int = DATABASE_WORKER_CAPACITY,
+        reserve_interactive: bool = True,
    ) -> None:
        """
        保存容量配置，线程只在显式启动后创建。
+
+        :param max_workers: 普通数据库事务的工作线程上限
+        :param capacity: 普通事务的运行和排队总容量
+        :param reserve_interactive: 是否额外保留独立的交互短事务 worker 和队列
        """
        if max_workers < 1:
            raise ValueError("数据库 worker 线程数必须大于 0")
@@ -91,6 +101,13 @@ class DatabaseWorker:
        self._reported_running = 0
        self._rejected = 0
        self._closing = False
+        self._interactive_worker = (
+            DatabaseWorker(
+                max_workers=INTERACTIVE_DATABASE_WORKERS,
+                capacity=INTERACTIVE_DATABASE_CAPACITY,
+                reserve_interactive=False,
+            ) if reserve_interactive else None
+        )

    async def start(self) -> None:
        """
@@ -106,11 +123,13 @@ class DatabaseWorker:
            thread_name_prefix="moviepilot-db",
        )
        self._closing = False
+        if self._interactive_worker is not None:
+            await self._interactive_worker.start()
        self._record_depth()

    def snapshot(self) -> DatabaseWorkerStats:
        """
-        返回无需访问任务对象的低基数运行快照。
+        返回普通执行组的低基数快照；交互保留组通过 interactive_snapshot 单独读取。
        """
        with self._state_lock:
            return DatabaseWorkerStats(
@@ -122,6 +141,12 @@ class DatabaseWorker:
                closing=self._closing,
            )

+    def interactive_snapshot(self) -> Optional[DatabaseWorkerStats]:
+        """
+        返回交互保留组的容量和负载；未配置保留组时返回 None。
+        """
+        return self._interactive_worker.snapshot() if self._interactive_worker is not None else None
+
    async def run(self, operation: Callable[[], T]) -> T:
        """
        执行短事务；结果发布和取消返回均晚于容量回收，取消仍等待已开始事务收口。
@@ -130,6 +155,8 @@ class DatabaseWorker:
        executor = self._executor
        if executor is None or self._loop is not loop or self._closing:
            raise DatabaseWorkerClosedError("数据库 worker 当前不可接收任务")
+        if self._interactive_worker is not None and is_interactive():
+            return await self._interactive_worker.run(operation)

        with self._state_lock:
            if self._queued + self._running >= self._capacity:
@@ -284,6 +311,8 @@ class DatabaseWorker:
        if self._loop is not asyncio.get_running_loop():
            raise RuntimeError("数据库 worker 必须在所属事件循环中停止")
        self._closing = True
+        if self._interactive_worker is not None:
+            await self._interactive_worker.shutdown()
        with self._state_lock:
            futures = tuple(self._futures.items())
        for future, _state in futures:
--- a/app/factory.py
+++ b/app/factory.py
@@ -17,6 +17,7 @@ from app.adapters.web.docs import install_api_docs_routes
 from app.adapters.web.health import install_health_routes
 from app.adapters.web.metrics import HttpMetricsMiddleware
 from app.adapters.web.plugin.routes import FastAPIDynamicRouteRegistry
+from app.adapters.web.priority import ExecutionPriorityMiddleware
 from app.adapters.web.security.access import (
    configure_token_codec,
    verify_apikey,
@@ -409,6 +410,7 @@ def create_app() -> FastAPI:
        allow_headers=["*"],
    )
    _app.add_middleware(CorrelationIdMiddleware)
+    _app.add_middleware(ExecutionPriorityMiddleware)
    _app.add_middleware(HttpMetricsMiddleware)

    @_app.middleware("http")
--- a/app/runtime/config.py
+++ b/app/runtime/config.py
@@ -709,6 +709,10 @@ class ConfigModel(BaseModel):

    # AI智能体配置
    AI_AGENT_ENABLE: bool = Field(default=False, description='AI智能体开关')
+    AGENT_INTERACTIVE_PRIORITY: bool = Field(default=True, description='为交互式智能助手隔离执行容量，并让新启动的非关键后台服务有限避让')
+    AGENT_BACKGROUND_DEFER_SECONDS: int = Field(default=30, ge=0, description='交互活跃时非关键后台服务最多避让的秒数，0为不避让；不暂停已开始的任务')
+    AGENT_LLM_TIMEOUT: int = Field(default=180, ge=0, description='Agent单次模型请求的总超时秒数，包含供应商重试；0为不限制')
+    SCHEDULER_JOB_TIMEOUT: int = Field(default=3600, ge=0, description='协程调度任务默认执行超时秒数，0为不限制；不强行终止同步线程')
    AI_AGENT_GLOBAL: bool = Field(default=False, description='合局AI智能体')
    AI_AGENT_HIDE_ENTRY: bool = Field(default=False, description='是否隐藏前端全局智能体入口')
    AI_AGENT_PET: str = Field(default="", description='默认助手形象，格式为 插件ID:形象key，留空使用内置机器人；用户个人选择优先')
--- a/app/runtime/execution.py
+++ b/app/runtime/execution.py
@@ -16,6 +16,7 @@ from typing import Any, Awaitable, Callable, ParamSpec, TypeVar, cast
 from anyio.to_thread import run_sync

 from app.runtime.gc import flush_thread_allocator_cache
+from app.runtime.priority import background_thread_limiter, interactive_thread_limiter, is_interactive
 from app.schemas.exception import ImmediateException

 TaskResult = TypeVar("TaskResult")
@@ -224,6 +225,8 @@ async def run_in_threadpool(
    """
    在线程中执行同步函数，保持 FastAPI 旧帮助函数的参数语义。

+    交互调用使用保留配额；普通宿主调用使用独立后台配额，避免耗尽框架同步入口的令牌。
+
    :param func: 在线程中执行的同步回调
    :param args: 回调的位置参数
    :param kwargs: 回调的关键字参数
@@ -234,9 +237,13 @@ async def run_in_threadpool(
    context = copy_context()

    def invoke() -> _Result:
+        """
+        在线程内恢复提交方上下文，确保嵌套调用沿用执行归属。
+        """
        return context.run(func, *args)

-    result: _Result = await run_sync(invoke)
+    limiter = interactive_thread_limiter() if is_interactive() else background_thread_limiter()
+    result: _Result = await run_sync(invoke, limiter=limiter)
    return result


--- a/app/runtime/priority.py
+++ b/app/runtime/priority.py
@@ -0,0 +1,173 @@
+"""
+交互轮次的执行归属、临时 HTTP 凭据和后台协作避让。
+
+仅宿主入口可以建立作用域；复制的上下文在轮次结束后自动失效。
+此处不抢占已开始的线程或事务，也不改变用户鉴权。
+"""
+
+import asyncio
+import threading
+import time
+import uuid
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+from typing import Optional
+
+from anyio import CapacityLimiter
+from anyio.lowlevel import RunVar
+from anyio.to_thread import current_default_thread_limiter
+
+from app.runtime.settings import get_runtime_setting
+
+PRIORITY_HEADER = "X-MoviePilot-Execution"
+INTERACTIVE_THREAD_WORKERS = 4
+INTERACTIVE_DATABASE_WORKERS = 1
+INTERACTIVE_DATABASE_CAPACITY = 8
+_condition = threading.Condition()
+
+
+@dataclass
+class _InteractiveScope:
+    """
+    记录一次可信交互的存活期。
+
+    :param key: 仅在本进程存活期间有效的随机 HTTP 传播凭据
+    :param active: 是否仍处于原始交互轮次内
+    """
+
+    key: str
+    active: bool = True
+
+
+_current: ContextVar[Optional[_InteractiveScope]] = ContextVar("interactive_execution", default=None)
+_scopes: dict[str, _InteractiveScope] = {}
+# RunVar 按事件循环而非请求保存 limiter，避免不同会话各建一份配额。
+_limiter: RunVar[CapacityLimiter] = RunVar("interactive_thread_limiter")
+_background_limiter: RunVar[CapacityLimiter] = RunVar("background_thread_limiter")
+
+
+def is_interactive() -> bool:
+    """
+    返回当前调用是否属于仍存活且已启用优先策略的交互轮次。
+    """
+    scope = _current.get()
+    return bool(scope and scope.active and get_runtime_setting("AGENT_INTERACTIVE_PRIORITY", True))
+
+
+def interactive_key() -> Optional[str]:
+    """
+    返回供本机 API 传播使用的临时凭据；普通后台调用返回 None。
+    """
+    scope = _current.get()
+    return scope.key if scope and is_interactive() else None
+
+
+@contextmanager
+def interactive_execution(enabled: bool = True) -> Iterator[None]:
+    """
+    为宿主确认的交互轮次登记归属，异常和取消均撤销凭据。
+
+    :param enabled: False 时隔离继承的上下文，供自主任务入口使用
+    """
+    if enabled and is_interactive():
+        yield
+        return
+    scope = _InteractiveScope(uuid.uuid4().hex) if enabled else None
+    token = _current.set(scope)
+    if scope is not None:
+        with _condition:
+            _scopes[scope.key] = scope
+    try:
+        yield
+    finally:
+        if scope is not None:
+            with _condition:
+                scope.active = False
+                _scopes.pop(scope.key, None)
+                _condition.notify_all()
+        _current.reset(token)
+
+
+@contextmanager
+def propagated_execution(key: Optional[str]) -> Iterator[None]:
+    """
+    仅恢复本进程登记且尚未结束的交互凭据，伪造或过期值按普通调用处理。
+    """
+    with _condition:
+        scope = _scopes.get(key or "")
+    token = _current.set(scope)
+    try:
+        yield
+    finally:
+        _current.reset(token)
+
+
+def interactive_thread_limiter() -> CapacityLimiter:
+    """
+    返回当前事件循环共享的交互线程配额，不消费 anyio 默认后台令牌。
+    """
+    limiter = _limiter.get(None)
+    if limiter is None:
+        limiter = CapacityLimiter(INTERACTIVE_THREAD_WORKERS)
+        _limiter.set(limiter)
+    return limiter
+
+
+def background_thread_limiter() -> CapacityLimiter:
+    """
+    隔离宿主后台同步调用，避免其耗尽 FastAPI 同步鉴权和端点的默认令牌。
+
+    普通容量跟随 anyio 默认配置；不修改框架 limiter 或第三方库的线程入口。
+    """
+    capacity = current_default_thread_limiter().total_tokens
+    limiter = _background_limiter.get(None)
+    if limiter is None:
+        limiter = CapacityLimiter(capacity)
+        _background_limiter.set(limiter)
+    else:
+        limiter.total_tokens = capacity
+    return limiter
+
+
+def background_defer_seconds() -> float:
+    """
+    返回非关键后台任务的最长避让秒数；当前交互主动触发的调用不自我等待。
+    """
+    if is_interactive() or not get_runtime_setting("AGENT_INTERACTIVE_PRIORITY", True):
+        return 0.0
+    return float(get_runtime_setting("AGENT_BACKGROUND_DEFER_SECONDS", 30))
+
+
+def _busy() -> bool:
+    """
+    返回是否仍有交互轮次在思考或执行工具。
+    """
+    with _condition:
+        return bool(_scopes)
+
+
+def wait_for_interactive_idle(timeout: float, accepting: Callable[[], bool]) -> None:
+    """
+    同步后台入口有限避让；按短间隔检查关闭，避免关停等待整段预算。
+    """
+    deadline = time.monotonic() + timeout
+    while _busy() and accepting():
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            break
+        with _condition:
+            _condition.wait(min(remaining, 0.05))
+
+
+async def await_interactive_idle(timeout: float, accepting: Callable[[], bool]) -> None:
+    """
+    协程后台入口有限避让，等待不占用主循环且保留取消传播。
+    """
+    deadline = time.monotonic() + timeout
+    while _busy() and accepting():
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            break
+        await asyncio.sleep(min(remaining, 0.05))
--- a/app/runtime/thread.py
+++ b/app/runtime/thread.py
@@ -3,12 +3,14 @@
 """

 import threading
+import time
 from concurrent.futures import Future
 from contextvars import copy_context
 from typing import Any, Callable, TypeVar, cast

 from app.foundation.singleton import Singleton
 from app.runtime.execution import OwnedThreadPoolExecutor
+from app.runtime.priority import INTERACTIVE_THREAD_WORKERS, is_interactive
 from app.runtime.settings import get_runtime_setting

 _Result = TypeVar("_Result")
@@ -20,15 +22,16 @@ _OwnedThreadPoolExecutor = OwnedThreadPoolExecutor

 class ThreadHelper(metaclass=Singleton):
    """
-    共享后台线程池 owner，负责关联上下文传播和生命周期收敛。
+    普通及交互保留线程池的共同 owner，负责上下文传播和生命周期收敛。
    """

    def __init__(self) -> None:
        """
-        按系统配置创建共享后台线程池。
+        按系统配置创建普通线程池，并保留独立的交互容量。
        """
        self._lifecycle_lock = threading.RLock()
        self.pool = self._new_pool()
+        self._interactive_pool = OwnedThreadPoolExecutor(max_workers=INTERACTIVE_THREAD_WORKERS)

    @staticmethod
    def _new_pool() -> OwnedThreadPoolExecutor:
@@ -46,7 +49,7 @@ class ThreadHelper(metaclass=Singleton):
            **kwargs: Any,
    ) -> Future[_Result]:
        """
-        向共享后台线程池提交任务，并传播提交时捕获的上下文。
+        按执行归属向普通或交互线程池提交任务，并传播提交时捕获的上下文。

        :param func: 需要在线程池中执行的同步函数
        :param args: 传给目标函数的位置参数
@@ -58,22 +61,27 @@ class ThreadHelper(metaclass=Singleton):
            # mypy 严格检查跳过 execution 实现导入，需要在兼容门面恢复 Future 泛型。
            return cast(
                Future[_Result],
-                self.pool.submit(context.run, func, *args, **kwargs),
+                (self._interactive_pool if is_interactive() else self.pool).submit(
+                    context.run, func, *args, **kwargs
+                ),
            )

    def reopen(self) -> bool:
        """
        为新的应用生命周期恢复线程池提交准入。

-        只有上一 owner 已完全收敛时才会构造新线程池，避免跨 lifespan
-        同时保留两组 worker。
+        只有上一 owner 的两组 worker 已完全收敛时才会构造新线程池，
+        避免同一执行组跨 lifespan 同时保留新旧 worker。
        """
        with self._lifecycle_lock:
-            if self.pool.accepting:
+            if self.pool.accepting and self._interactive_pool.accepting:
                return True
-            if not self.pool.shutdown_bounded(timeout=0.0):
+            normal_stopped = self.pool.shutdown_bounded(timeout=0.0)
+            interactive_stopped = self._interactive_pool.shutdown_bounded(timeout=0.0)
+            if not normal_stopped or not interactive_stopped:
                return False
            self.pool = self._new_pool()
+            self._interactive_pool = OwnedThreadPoolExecutor(max_workers=INTERACTIVE_THREAD_WORKERS)
            return True

    def shutdown(
@@ -87,4 +95,10 @@ class ThreadHelper(metaclass=Singleton):
        :return: 全部任务和 worker 均已终止时返回 True，否则返回 False
        """
        with self._lifecycle_lock:
-            return bool(self.pool.shutdown_bounded(timeout=timeout))
+            deadline = time.monotonic() + max(0.0, timeout)
+            # 两组执行器先共同封口，再在同一个预算内等待，超时仍保留原 owner。
+            self.pool.shutdown_bounded(timeout=0.0)
+            self._interactive_pool.shutdown_bounded(timeout=0.0)
+            normal_stopped = self.pool.shutdown_bounded(timeout=max(0.0, deadline - time.monotonic()))
+            interactive_stopped = self._interactive_pool.shutdown_bounded(timeout=max(0.0, deadline - time.monotonic()))
+            return bool(normal_stopped and interactive_stopped)
--- a/app/scheduler/admission.py
+++ b/app/scheduler/admission.py
@@ -0,0 +1,62 @@
+"""
+非关键调度任务的有限避让，保留原调度所有权和重入保护。
+"""
+
+import asyncio
+from collections.abc import Callable
+from typing import Any
+
+from app.runtime.priority import (
+    await_interactive_idle,
+    background_defer_seconds,
+    interactive_execution,
+    wait_for_interactive_idle,
+)
+
+# 安全检查、持久副作用补偿与生命周期维护不得等待用户对话结束。
+_CRITICAL_OWNERS = frozenset({"outbox", "security", "runtime", "database"})
+
+
+def _defer_budget(job: dict[str, Any]) -> float:
+    """
+    只对自动业务任务避让，手动专用任务和关键维护保留立即执行语义。
+    """
+    if job.get("manual") or job.get("owner") in _CRITICAL_OWNERS:
+        return 0.0
+    return background_defer_seconds()
+
+
+async def defer_coroutine(job: dict[str, Any], accepting: Callable[[], bool]) -> None:
+    """
+    协程等待期间持续持有调度句柄，取消或关闭仍由原 owner 收口。
+    """
+    job["deferred"] = True
+    try:
+        await await_interactive_idle(_defer_budget(job), accepting)
+        if not accepting():
+            raise asyncio.CancelledError()
+    finally:
+        job["deferred"] = False
+
+
+def invoke_background(
+    job: dict[str, Any], accepting: Callable[[], bool], func: Callable[..., Any],
+    *args: Any, **kwargs: Any,
+) -> Any:
+    """
+    在线程入口有限避让后以普通执行归属调用业务函数。
+
+    已在事件循环中的旧同步入口不能阻塞等待；这类入口沿用原立即调用合同。
+    """
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        job["deferred"] = True
+        try:
+            wait_for_interactive_idle(_defer_budget(job), accepting)
+        finally:
+            job["deferred"] = False
+    if not accepting():
+        raise RuntimeError("调度器正在关闭，尚未开始的任务已取消")
+    with interactive_execution(enabled=False):
+        return func(*args, **kwargs)
--- a/app/scheduler/catalog.py
+++ b/app/scheduler/catalog.py
@@ -87,6 +87,10 @@ def _wallpaper_job_specs(


 # APScheduler 未提供类型注解，基类在 mypy 中是 Any
+# 自主 Agent 任务只需独立的有限触发容量；协程执行仍由原主循环和句柄 owner 托管。
+_AGENT_SCHEDULER_WORKERS = 2
+
+
 class _SchedulerThreadPoolExecutor(BasePoolExecutor):  # type: ignore[misc]
    """
    APScheduler 线程池执行器：定时任务线程按需增长、不收缩，空闲时交还各自的分配器线程缓存。
@@ -269,7 +273,11 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):

        self._scheduler = BackgroundScheduler(
            timezone=config.timezone,
-            executors={"default": _SchedulerThreadPoolExecutor(config.scheduler_workers)},
+            executors={
+                "default": _SchedulerThreadPoolExecutor(config.scheduler_workers),
+                "agent": _SchedulerThreadPoolExecutor(_AGENT_SCHEDULER_WORKERS),
+            },
+            job_defaults={"coalesce": True, "max_instances": 1, "misfire_grace_time": None},
        )

        self._register_database_backup_job(config)
@@ -527,6 +535,7 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase):
                self.start,
                "interval",
                id="agent_heartbeat",
+                executor="agent",
                name="智能体定时任务",
                hours=config.ai_agent_job_interval,
                kwargs={"job_id": "agent_heartbeat"},
--- a/app/scheduler/execution.py
+++ b/app/scheduler/execution.py
@@ -19,8 +19,11 @@ from app.runtime.events import eventmanager
 from app.runtime.log import logger
 from app.runtime.loop import main_loop_registry
 from app.runtime.observability import record_metric
+from app.runtime.priority import interactive_execution
 from app.runtime.progress import ProgressHelper
 from app.runtime.scheduling import TimerUtils
+from app.runtime.settings import get_runtime_setting
+from app.scheduler.admission import defer_coroutine, invoke_background
 from app.scheduler.contract import _SchedulerOwnerBase
 from app.schemas.dashboard import ScheduleInfo as _SchemaScheduleInfo
 from app.schemas.types import EventType
@@ -50,6 +53,9 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
        把注册 generation 写入可变运行时状态。
        """
        self._registry.assign_generation(job_id, job)
+        if inspect.iscoroutinefunction(job.get("func")) and job.get("timeout_seconds") is None:
+            timeout = int(get_runtime_setting("SCHEDULER_JOB_TIMEOUT", 3600))
+            job["timeout_seconds"] = timeout if timeout > 0 else None

    def _is_job_active(self, job_id: str) -> bool:
        """
@@ -179,10 +185,12 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
        success = True
        error = None
        try:
-            result = await JobExecutionState.await_result(
-                coro_factory(),
-                timeout_seconds=job.get("timeout_seconds"),
-            )
+            await defer_coroutine(job, self._accepting_submissions)
+            with interactive_execution(enabled=False):
+                result = await JobExecutionState.await_result(
+                    coro_factory(),
+                    timeout_seconds=job.get("timeout_seconds"),
+                )
            error = self._get_result_error(result)
            success = error is None
        except asyncio.TimeoutError as err:
@@ -329,7 +337,7 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
                p.join()
            else:
                # 普通函数
-                result = func(*args, **kwargs)
+                result = invoke_background(job, self._accepting_submissions, func, *args, **kwargs)
                error = self._get_result_error(result)
                success = error is None
        except Exception as e:
@@ -401,7 +409,8 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
                            id=job_id,
                            name=name,
                            provider=provider_name,
-                            status="正在运行",
+                            status="等待交互结束" if service.get("deferred") else "正在运行",
+                            execution_group=service.get("execution_group", "default"),
                            progress=progress.value if progress else 0,
                            progress_text=progress.text if progress else None,
                            progress_enable=progress.enable if progress else False,
@@ -429,6 +438,7 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
                        name=job.name,
                        provider=service_state.get("provider_name", "[系统]"),
                        status=status,
+                        execution_group=service_state.get("execution_group", "default"),
                        next_run=next_run,
                        progress=progress.value if progress else 0,
                        progress_text=progress.text if progress else None,
@@ -450,6 +460,7 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
                        name=service.get("name"),
                        provider=service.get("provider_name", "[系统]"),
                        status=("正在运行" if self._is_job_active(job_id) else "等待"),
+                        execution_group=service.get("execution_group", "default"),
                        progress=progress.value if progress else 0,
                        progress_text=progress.text if progress else None,
                        progress_enable=progress.enable if progress else False,
--- a/app/scheduler/reconcile.py
+++ b/app/scheduler/reconcile.py
@@ -163,6 +163,7 @@ class SchedulerReconcileOwner(_SchedulerOwnerBase):
                trigger=trigger,
                id=job_id,
                name=task.name,
+                executor="agent",
                kwargs={"job_id": job_id, "task_id": task_id},
                coalesce=True,
                max_instances=1,
--- a/app/schemas/dashboard.py
+++ b/app/schemas/dashboard.py
@@ -82,6 +82,7 @@ class ScheduleProgress(BaseModel):
    name_i18n: Optional[str] = Field(default=None, description="多语言名称")
    provider: Optional[str] = Field(default=None, description="提供者")
    provider_i18n: Optional[str] = Field(default=None, description="多语言提供者")
+    execution_group: str = Field(default="default", description="调度执行组：agent为自主Agent独立线程池，default为普通后台任务；交互轮次使用另外保留的容量")
    enable: Optional[bool] = Field(default=False, description="是否正在执行")
    value: Optional[float] = Field(default=0.0, description="当前完成百分比")
    text: Optional[str] = Field(default=None, description="当前进度文本")
--- a/docs/agent-execution.md
+++ b/docs/agent-execution.md
@@ -0,0 +1,62 @@
+# 智能助手执行优先策略
+
+交互轮次从宿主 `MoviePilotAgent.process` 进入，到本轮推理及工具调用收口后结束。
+网页、消息渠道、CLI 和兼容 API 的前台请求使用交互容量；自主定时任务与心跳保持后台归属，
+即使配置了结果回传渠道也不会升级为交互轮次。
+
+默认启用 `AGENT_INTERACTIVE_PRIORITY=true`。这是协作式资源隔离：正在运行的同步线程、
+数据库事务和后台协程不会被抢占。主事件循环仍采用 asyncio 的协作调度，
+因此同步代码阻塞事件循环、CPU 饱和、数据库锁竞争和模型网络延迟仍可能影响响应。
+
+| 执行边界 | 交互容量 | 普通后台容量 |
+| --- | --- | --- |
+| `ThreadHelper.submit` | 独立 4 线程 | 保留现有 `CONF.threadpool` |
+| 宿主 `run_in_threadpool` | 当前事件循环共享 4 个 anyio 令牌 | 独立后台 limiter，容量跟随 anyio 默认配置（通常 40） |
+| `DatabaseWorker.run` | 独立 1 线程、运行及排队总计 8 项 | 保留 4 线程、总计 32 项默认值 |
+| Agent 同步工具 | 各能力域独立的交互桶 | 保留原能力域限额，供自主任务使用 |
+| APScheduler | 自主 Agent 使用 `agent` 执行组，2 线程 | 其他任务使用原 `default` 组 |
+
+以上容量按需启动工作线程，关闭时沿用各自生命周期 owner 的封口和收敛合同。
+取消不会提前归还仍在执行的同步工具、线程调用或数据库事务的容量。
+聊天持久化的全局写入准入也区分交互和后台；同一会话的写锁及单会话上限继续共享，
+以免并发覆盖历史。`DatabaseWorker.snapshot()` 返回普通组，`interactive_snapshot()` 返回保留组；
+数据库队列和执行指标包含两组的增减量。
+
+执行归属通过上下文复制传入线程。`moviepilot_api` 的本机 HTTP 调用携带本进程随机登记的
+临时执行凭据，由 ASGI 中间件恢复；普通自填请求头或已经结束的轮次无法取得优先级。
+用户令牌与权限检查照常执行。子进程、外部 MCP 服务和插件自建线程池不受本策略控制。
+框架同步鉴权与同步端点保留 anyio 默认 limiter，宿主后台线程调用不再消耗该组令牌；
+框架自身的 HTTP 请求之间仍共用默认配额，不能保证在任意 HTTP 洪峰下立即响应。
+
+## 后台避让
+
+交互活跃时，新进入调度执行入口的非关键自动任务最多等待
+`AGENT_BACKGROUND_DEFER_SECONDS` 秒，默认 30 秒，设为 `0` 禁用等待。
+轮次结束后等待任务继续执行；持续对话超过预算时也会继续，避免后台任务饥饿。
+期间保留原任务 generation、重入保护和协程句柄，不重复注册或补投任务。
+调度器明确使用合并过期触发、单实例和无限 misfire 宽限，避免有限避让产生的排队
+被默认一秒宽限误判为过期丢弃；任务自己的重入保护继续生效。
+调度服务列表以“等待交互结束”显示正在避让的任务，API 的 `execution_group` 返回实际执行组。
+
+Outbox 补偿、安全检查、数据库维护、运行时维护和手动专用任务不避让。
+交互主动触发的同步调用不会等待自己结束；业务调度函数执行时会撤销继承的交互归属。
+已有主循环同步入口沿用原调用合同，不在事件循环中同步等待。
+避让只作用于新进入执行入口的任务，不会暂停或重放已开始的业务操作。
+
+APScheduler 3 的标准线程池没有任意数值 `priority` 的调度能力。
+这里通过独立执行组和实际准入等待实现隔离，不向 `add_job` 传入无效的优先级参数。
+独立调度执行组仅隔离触发线程，协程仍回投到同一主循环。
+
+## 超时
+
+- `AGENT_LLM_TIMEOUT`：单次根 Agent 模型请求总预算，默认 180 秒，覆盖流式与非流式生成
+  以及该请求的供应商重试；`0` 表示不限时。工具执行时间由原 `LLM_TOOL_TIMEOUT` 管理。
+- `SCHEDULER_JOB_TIMEOUT`：协程调度任务的默认执行预算，默认 3600 秒，等待避让不计入执行时间；
+  `0` 表示不限时。显式 `JobSpec.timeout_seconds` 优先于此默认值。
+
+超时通过协程取消收口，不能杀死同步线程或撤销已经发生的外部副作用。
+数据库和线程调用仍等待实际操作收尾后释放容量，因而收尾时间可能超过超时预算。
+超时后应先查询操作状态，再决定是否重试写入。
+
+关闭 `AGENT_INTERACTIVE_PRIORITY` 可让 Agent 使用普通执行容量并停用避让；两项超时配置独立生效。
+这些配置由部署环境或 `app.env` 提供，修改后重启生效。
--- a/docs/architecture-overview.md
+++ b/docs/architecture-overview.md
@@ -816,8 +816,8 @@ flowchart LR

 | 指标 | 当前值 |
 |---|---:|
-| Python 模块 | 1124 |
-| 内部导入边 | 9,621 |
+| Python 模块 | 1127 |
+| 内部导入边 | 9,640 |
 | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） |
--- a/docs/development-setup.md
+++ b/docs/development-setup.md
@@ -285,6 +285,8 @@ CI 上传 `complexity-report` 工件，门禁失败时也保留已生成的指

 ### Agent 学习验证

+交互执行隔离、后台避让及超时配置见 [智能助手执行优先策略](agent-execution.md)。
+
 `tests/test_agent_learning_skills.py`、`tests/test_agent_learning_memory.py` 和
 `tests/test_agent_learning_review.py` 用临时目录与确定性模型核对 Hermes 学习契约，
 不访问实际 LLM 或个人记忆。匹配器、威胁模式和提示词的上游来源及 MIT 许可在
--- a/docs/rules/05-architecture.md
+++ b/docs/rules/05-architecture.md
@@ -158,6 +158,9 @@ upload/domain failures to HTTP responses, and frames Application events as SSE.
 | `app/runtime/resources.py` | Provider-neutral acquisition, observation and shutdown facade for process-owned optional resources |
 | `app/runtime/tasks.py` | Lifespan-scoped ownership, cancellation and bounded shutdown waiting for in-process background tasks |
 | `app/runtime/execution.py` | Shared sync/async execution and cross-thread submission boundary with correlation propagation |
+| `app/runtime/priority.py` | 交互执行作用域、有限后台避让及每个事件循环共享的交互 anyio 配额；仅传播临时执行归属，不替代身份鉴权 |
+| `app/adapters/web/priority.py` | 从本进程登记的存活临时凭据恢复本机 Agent API 执行归属 |
+| `app/scheduler/admission.py` | 关键任务豁免、协程和同步调度入口的有限避让；保留原 generation 和句柄 owner |
 | `app/runtime/correlation.py` | Request/cross-thread correlation context and safe propagation into logs and child work |
 | `app/runtime/state.py` | Process restart and update state |
 | `app/runtime/dependencies/` | Runtime dependency profile selection in `profile.py` and loaded native payload activation detection in `native.py`; the package root contains no implementation or host re-exports |
--- a/tests/fixtures/architecture/dependency-baseline.json
+++ b/tests/fixtures/architecture/dependency-baseline.json
@@ -975,8 +975,8 @@
      "runtime_only": true
    }
  },
-  "edge_count": 9621,
-  "edge_sha256": "30d190a87ebc01743763a67be2f52f9b88e59bc553279fb65b66b758131c344e",
+  "edge_count": 9640,
+  "edge_sha256": "e282d6307ea8c6f0cd2843f5c2faae04995bc2630c945e408d1dd490e337fcfe",
  "edges": [
    "app -> app.foundation",
    "app -> app.foundation.environment",
@@ -1216,6 +1216,8 @@
    "app.adapters.web.health -> app.runtime.health",
    "app.adapters.web.metrics -> app.runtime",
    "app.adapters.web.metrics -> app.runtime.observability",
+    "app.adapters.web.priority -> app.runtime",
+    "app.adapters.web.priority -> app.runtime.priority",
    "app.adapters.web.security.access -> app.runtime",
    "app.adapters.web.security.access -> app.runtime.log",
    "app.adapters.web.security.access -> app.runtime.settings",
@@ -1237,6 +1239,7 @@
    "app.agent.api.executor -> app.application.security.token",
    "app.agent.api.executor -> app.runtime",
    "app.agent.api.executor -> app.runtime.log",
+    "app.agent.api.executor -> app.runtime.priority",
    "app.agent.api.executor -> app.runtime.settings",
    "app.agent.callback -> app.agent",
    "app.agent.callback -> app.agent.policy",
@@ -1528,6 +1531,8 @@
    "app.agent.middleware.config -> app.agent.middleware",
    "app.agent.middleware.config -> app.agent.middleware.utils",
    "app.agent.middleware.config -> app.agent.runtime",
+    "app.agent.middleware.config -> app.runtime",
+    "app.agent.middleware.config -> app.runtime.settings",
    "app.agent.middleware.guardrails -> app.agent",
    "app.agent.middleware.guardrails -> app.agent.code",
    "app.agent.middleware.guardrails -> app.agent.code.authority",
@@ -1747,6 +1752,7 @@
    "app.agent.orchestrator -> app.runtime.execution",
    "app.agent.orchestrator -> app.runtime.log",
    "app.agent.orchestrator -> app.runtime.observability",
+    "app.agent.orchestrator -> app.runtime.priority",
    "app.agent.orchestrator -> app.runtime.settings",
    "app.agent.orchestrator -> app.schemas",
    "app.agent.orchestrator -> app.schemas.event",
@@ -1886,6 +1892,7 @@
    "app.agent.tools.base -> app.foundation.identity",
    "app.agent.tools.base -> app.runtime",
    "app.agent.tools.base -> app.runtime.log",
+    "app.agent.tools.base -> app.runtime.priority",
    "app.agent.tools.base -> app.runtime.settings",
    "app.agent.tools.base -> app.schemas",
    "app.agent.tools.base -> app.schemas.message",
@@ -3769,6 +3776,7 @@
    "app.application.messaging.chat -> app.application.messaging.recall",
    "app.application.messaging.chat -> app.runtime",
    "app.application.messaging.chat -> app.runtime.observability",
+    "app.application.messaging.chat -> app.runtime.priority",
    "app.application.messaging.chat -> app.schemas",
    "app.application.messaging.chat -> app.schemas.agent",
    "app.application.messaging.chat -> app.schemas.exception",
@@ -6799,6 +6807,7 @@
    "app.db.session -> app.runtime.settings",
    "app.db.worker -> app.runtime",
    "app.db.worker -> app.runtime.observability",
+    "app.db.worker -> app.runtime.priority",
    "app.db.worker -> app.schemas",
    "app.db.worker -> app.schemas.exception",
    "app.doctor.checks -> app.adapters",
@@ -6988,6 +6997,7 @@
    "app.factory -> app.adapters.web.metrics",
    "app.factory -> app.adapters.web.plugin",
    "app.factory -> app.adapters.web.plugin.routes",
+    "app.factory -> app.adapters.web.priority",
    "app.factory -> app.adapters.web.security",
    "app.factory -> app.adapters.web.security.access",
    "app.factory -> app.api",
@@ -8955,6 +8965,7 @@
    "app.runtime.events -> app.schemas.types",
    "app.runtime.execution -> app.runtime",
    "app.runtime.execution -> app.runtime.gc",
+    "app.runtime.execution -> app.runtime.priority",
    "app.runtime.execution -> app.schemas",
    "app.runtime.execution -> app.schemas.exception",
    "app.runtime.extensions.module.adapter -> app.runtime",
@@ -9171,6 +9182,8 @@
    "app.runtime.observability -> app.runtime",
    "app.runtime.observability -> app.runtime.deprecation",
    "app.runtime.observability -> app.runtime.deprecation.policy",
+    "app.runtime.priority -> app.runtime",
+    "app.runtime.priority -> app.runtime.settings",
    "app.runtime.progress -> app.runtime",
    "app.runtime.progress -> app.runtime.cache",
    "app.runtime.progress -> app.runtime.localization",
@@ -9195,12 +9208,15 @@
    "app.runtime.thread -> app.foundation.singleton",
    "app.runtime.thread -> app.runtime",
    "app.runtime.thread -> app.runtime.execution",
+    "app.runtime.thread -> app.runtime.priority",
    "app.runtime.thread -> app.runtime.settings",
    "app.runtime.version -> app.foundation",
    "app.runtime.version -> app.foundation.environment",
    "app.runtime.version -> app.runtime",
    "app.runtime.version -> app.runtime.log",
    "app.runtime.version -> app.runtime.settings",
+    "app.scheduler.admission -> app.runtime",
+    "app.scheduler.admission -> app.runtime.priority",
    "app.scheduler.bridge -> app.runtime",
    "app.scheduler.bridge -> app.runtime.loop",
    "app.scheduler.bridge -> app.scheduler",
@@ -9240,9 +9256,12 @@
    "app.scheduler.execution -> app.runtime.log",
    "app.scheduler.execution -> app.runtime.loop",
    "app.scheduler.execution -> app.runtime.observability",
+    "app.scheduler.execution -> app.runtime.priority",
    "app.scheduler.execution -> app.runtime.progress",
    "app.scheduler.execution -> app.runtime.scheduling",
+    "app.scheduler.execution -> app.runtime.settings",
    "app.scheduler.execution -> app.scheduler",
+    "app.scheduler.execution -> app.scheduler.admission",
    "app.scheduler.execution -> app.scheduler.contract",
    "app.scheduler.execution -> app.schemas",
    "app.scheduler.execution -> app.schemas.dashboard",
@@ -10600,7 +10619,7 @@
    "app.workflow.actions.transfer_file -> app.workflow",
    "app.workflow.actions.transfer_file -> app.workflow.actions"
  ],
-  "module_count": 1124,
+  "module_count": 1127,
  "modules": [
    "app",
    "app.adapters",
@@ -10661,6 +10680,7 @@
    "app.adapters.web.metrics",
    "app.adapters.web.plugin",
    "app.adapters.web.plugin.routes",
+    "app.adapters.web.priority",
    "app.adapters.web.security",
    "app.adapters.web.security.access",
    "app.agent",
@@ -11565,6 +11585,7 @@
    "app.runtime.log",
    "app.runtime.loop",
    "app.runtime.observability",
+    "app.runtime.priority",
    "app.runtime.progress",
    "app.runtime.rate",
    "app.runtime.reload",
@@ -11579,6 +11600,7 @@
    "app.runtime.version",
    "app.runtime.webpush",
    "app.scheduler",
+    "app.scheduler.admission",
    "app.scheduler.bridge",
    "app.scheduler.catalog",
    "app.scheduler.chain",
--- a/tests/fixtures/architecture/startup-performance-baseline.json
+++ b/tests/fixtures/architecture/startup-performance-baseline.json
@@ -6,7 +6,7 @@
  "repeat": 3,
  "targets": {
    "app.startup.lifecycle": {
-      "loaded_app_module_count": 588,
+      "loaded_app_module_count": 590,
      "max_ms": 1293.338,
      "median_ms": 1156.239,
      "min_ms": 1102.806,
@@ -17,7 +17,7 @@
      ]
    },
    "app.factory": {
-      "loaded_app_module_count": 601,
+      "loaded_app_module_count": 604,
      "max_ms": 1127.911,
      "median_ms": 1122.382,
      "min_ms": 1119.221,
@@ -28,7 +28,7 @@
      ]
    },
    "app.main": {
-      "loaded_app_module_count": 603,
+      "loaded_app_module_count": 606,
      "max_ms": 1188.652,
      "median_ms": 1183.509,
      "min_ms": 1174.522,
--- a/tests/test_interactive_priority.py
+++ b/tests/test_interactive_priority.py
@@ -0,0 +1,448 @@
+"""
+交互执行隔离、HTTP 传播、有限避让与关闭合同的离线回归。
+"""
+
+import asyncio
+import threading
+import time
+from contextvars import copy_context
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from anyio.to_thread import current_default_thread_limiter, run_sync
+
+from app.adapters.web.priority import ExecutionPriorityMiddleware
+from app.agent.api import executor as api_module
+from app.agent.api.executor import ApiExecutionContext, MoviePilotApiExecutor
+from app.agent.middleware.config import RuntimeConfigMiddleware
+from app.agent.orchestrator import MoviePilotAgent
+from app.agent.terminal.ownership import TerminalScope
+from app.agent.tools import base as tool_module
+from app.application.messaging.chat import AgentChatPersistenceService
+from app.db.worker import DatabaseWorker
+from app.runtime import priority
+from app.runtime.execution import run_in_threadpool
+from app.runtime.thread import ThreadHelper
+from app.scheduler import execution as scheduler_execution
+from app.scheduler.admission import defer_coroutine, invoke_background
+from app.schemas.exception import DatabaseWorkerOverloadedError
+
+
+@pytest.fixture(autouse=True)
+def execution_settings(monkeypatch):
+    """
+    固定本文件的执行策略，避免依赖真实配置或等待生产默认预算。
+    """
+    settings = {"AGENT_INTERACTIVE_PRIORITY": True, "AGENT_BACKGROUND_DEFER_SECONDS": 0.1}
+    monkeypatch.setattr(priority, "get_runtime_setting", lambda key, default=None: settings.get(key, default))
+    return settings
+
+
+async def _entered(event: threading.Event) -> None:
+    """
+    有界等待线程开始；超时明确失败，避免挂住测试进程。
+    """
+    async with asyncio.timeout(2):
+        while not event.is_set():
+            await asyncio.sleep(0.001)
+
+
+def test_scope_restoration_expiry_and_disabled_policy(execution_settings) -> None:
+    """
+    轮次结束撤销复制上下文和 HTTP 凭据，禁用策略后不再取得保留容量。
+    """
+    assert not priority.is_interactive()
+    with priority.interactive_execution():
+        key = priority.interactive_key()
+        copied = copy_context()
+        with priority.interactive_execution():
+            assert priority.interactive_key() == key
+        with priority.interactive_execution(enabled=False):
+            assert not priority.is_interactive()
+            with priority.propagated_execution(key):
+                assert priority.is_interactive()
+            with priority.propagated_execution("forged"):
+                assert not priority.is_interactive()
+        assert priority.background_defer_seconds() == 0
+        execution_settings["AGENT_INTERACTIVE_PRIORITY"] = False
+        assert priority.interactive_key() is None
+        assert priority.background_defer_seconds() == 0
+    assert copied.run(priority.is_interactive) is False
+    with priority.propagated_execution(key):
+        assert not priority.is_interactive()
+
+
+@pytest.mark.asyncio
+async def test_http_only_restores_live_process_registered_key(monkeypatch) -> None:
+    """
+    仅真实宿主轮次可经 API 传播优先级，普通请求、过期和伪造头均无效。
+    """
+    monkeypatch.setattr(api_module, "create_access_token", lambda **_kwargs: "test-auth")
+    executor = MoviePilotApiExecutor(context=ApiExecutionContext("1", "admin", True))
+    observed = []
+
+    async def app(_scope, _receive, _send):
+        """
+        观察下游请求实际看到的执行归属。
+        """
+        observed.append(priority.is_interactive())
+
+    middleware = ExecutionPriorityMiddleware(app)
+    with priority.interactive_execution():
+        headers = executor._build_headers()
+        key = headers[priority.PRIORITY_HEADER]
+        assert headers["Authorization"] == "Bearer test-auth"
+        with priority.interactive_execution(enabled=False):
+            await middleware({"type": "http", "headers": [(b"x-moviepilot-execution", key.encode())]}, None, None)
+            assert not priority.is_interactive()
+    assert priority.PRIORITY_HEADER not in executor._build_headers()
+    for value in (key, "forged", ""):
+        await middleware({"type": "http", "headers": [(b"x-moviepilot-execution", value.encode())]}, None, None)
+    await middleware({"type": "lifespan"}, None, None)
+    assert observed == [True, False, False, False, False]
+
+
+@pytest.mark.asyncio
+async def test_anyio_interactive_and_gateway_capacity_survive_background_saturation() -> None:
+    """
+    普通宿主令牌用尽时框架入口仍执行，两组均忙时交互工具仍可取得保留容量。
+    """
+    limiter = current_default_thread_limiter()
+    original = limiter.total_tokens
+    limiter.total_tokens = 1
+    entered = threading.Event()
+    framework_entered = threading.Event()
+    release = threading.Event()
+    task = asyncio.create_task(run_in_threadpool(lambda: entered.set() or release.wait(3)))
+    framework = None
+    try:
+        await _entered(entered)
+        # FastAPI 的同步依赖和端点走库默认入口，也不能等待宿主后台调用归还令牌。
+        assert await asyncio.wait_for(run_sync(lambda: "gateway"), 1) == "gateway"
+        framework = asyncio.create_task(run_sync(lambda: framework_entered.set() or release.wait(3)))
+        await _entered(framework_entered)
+        with priority.interactive_execution():
+            assert priority.interactive_thread_limiter() is priority.interactive_thread_limiter()
+            assert await asyncio.wait_for(run_in_threadpool(priority.is_interactive), 1) is True
+        assert priority.background_thread_limiter() is priority.background_thread_limiter()
+        assert not task.done()
+    finally:
+        release.set()
+        await task
+        if framework is not None:
+            await framework
+        limiter.total_tokens = original
+
+
+@pytest.mark.asyncio
+async def test_database_reservation_is_bounded_and_cancelled_transaction_keeps_owner() -> None:
+    """
+    普通 DB 队列满时交互仍执行，交互容量也有上限，取消等待真实事务结束。
+    """
+    worker = DatabaseWorker(max_workers=1, capacity=1)
+    await worker.start()
+    background_entered = threading.Event()
+    interactive_entered = threading.Event()
+    release = threading.Event()
+    interactive_release = threading.Event()
+    background = asyncio.create_task(worker.run(lambda: background_entered.set() or release.wait(3)))
+    tasks = []
+    try:
+        await _entered(background_entered)
+        with pytest.raises(DatabaseWorkerOverloadedError):
+            await worker.run(lambda: None)
+        with priority.interactive_execution():
+            assert await asyncio.wait_for(worker.run(priority.is_interactive), 1) is True
+            active = asyncio.create_task(worker.run(lambda: interactive_entered.set() or interactive_release.wait(3)))
+            tasks.append(active)
+            await _entered(interactive_entered)
+            for _ in range(priority.INTERACTIVE_DATABASE_CAPACITY - 1):
+                tasks.append(asyncio.create_task(worker.run(lambda: None)))
+            await asyncio.sleep(0)
+            with pytest.raises(DatabaseWorkerOverloadedError):
+                await worker.run(lambda: None)
+            active.cancel()
+            await asyncio.sleep(0)
+            assert not active.done()
+            interactive_release.set()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            assert await worker.run(lambda: "reused") == "reused"
+            assert worker.interactive_snapshot().running == 0
+            assert worker._interactive_worker.interactive_snapshot() is None
+        assert not background.done()
+    finally:
+        release.set()
+        interactive_release.set()
+        await asyncio.gather(background, *tasks, return_exceptions=True)
+        await worker.shutdown()
+
+
+def test_thread_helper_reservation_shutdown_and_reopen(monkeypatch) -> None:
+    """
+    普通池占满时使用交互池；关闭先共同封口，未收敛时不得创建替代 owner。
+    """
+    from app.runtime import thread as thread_module
+
+    monkeypatch.setattr(thread_module, "get_runtime_setting", lambda _key: SimpleNamespace(threadpool=1))
+    helper = object.__new__(ThreadHelper)
+    helper.__init__()
+    entered = threading.Event()
+    release = threading.Event()
+    future = helper.submit(lambda: entered.set() or release.wait(3))
+    try:
+        assert entered.wait(1)
+        with priority.interactive_execution():
+            assert helper.submit(priority.is_interactive).result(1) is True
+            assert helper.shutdown(timeout=0.01) is False
+            with pytest.raises(RuntimeError):
+                helper.submit(lambda: None)
+        assert helper.reopen() is False
+    finally:
+        release.set()
+        future.result(1)
+        assert helper.shutdown(timeout=1)
+    assert helper.reopen()
+    with priority.interactive_execution():
+        assert helper.submit(lambda: "restored").result(1) == "restored"
+    assert helper.shutdown(timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_tool_bucket_background_saturation_does_not_block_interactive() -> None:
+    """
+    自主任务占满某工具桶后，交互工具仍获得独立额度并保留归属。
+    """
+    release = threading.Event()
+    entered = [threading.Event() for _ in range(tool_module._BLOCKING_BUCKET_LIMITS["web"])]
+    background = [asyncio.create_task(tool_module.run_agent_blocking("web", lambda e=e: e.set() or release.wait(3))) for e in entered]
+    try:
+        for event in entered:
+            await _entered(event)
+        with priority.interactive_execution():
+            assert await asyncio.wait_for(tool_module.run_agent_blocking("web", priority.is_interactive), 1) is True
+        assert all(not task.done() for task in background)
+    finally:
+        release.set()
+        await asyncio.gather(*background)
+        assert await tool_module.close_blocking_executors(timeout_seconds=1)
+        assert tool_module.reopen_blocking_executors()
+
+
+@pytest.mark.asyncio
+async def test_background_defer_resumes_and_does_not_starve(execution_settings) -> None:
+    """
+    活跃轮次使后台等待，交互结束立即恢复；持续对话只等待配置预算。
+    """
+    job = {"owner": "subscription"}
+    def accepting():
+        """
+        保持调度器运行，用于验证避让期限。
+        """
+        return True
+    with priority.interactive_execution():
+        with priority.interactive_execution(enabled=False):
+            task = asyncio.create_task(defer_coroutine(job, accepting))
+        await asyncio.sleep(0.02)
+        assert job["deferred"] and not task.done()
+    await asyncio.wait_for(task, 1)
+    assert job["deferred"] is False
+    execution_settings["AGENT_BACKGROUND_DEFER_SECONDS"] = 0.01
+    with priority.interactive_execution():
+        with priority.interactive_execution(enabled=False):
+            await asyncio.wait_for(defer_coroutine(job, accepting), 1)
+            priority.wait_for_interactive_idle(0.01, accepting)
+            for exempt in ({"owner": "outbox"}, {"manual": True}):
+                await asyncio.wait_for(defer_coroutine(exempt, accepting), 1)
+            with pytest.raises(asyncio.CancelledError):
+                await defer_coroutine(job, lambda: False)
+    assert not job["deferred"]
+
+
+@pytest.mark.asyncio
+async def test_cancelling_deferred_job_clears_state() -> None:
+    """
+    等待交互期间取消任务会复原展示状态，不会继续调用业务函数。
+    """
+    job = {"owner": "plugin:test"}
+    with priority.interactive_execution():
+        with priority.interactive_execution(enabled=False):
+            task = asyncio.create_task(defer_coroutine(job, lambda: True))
+        await asyncio.sleep(0)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+    assert job["deferred"] is False
+
+
+def test_sync_deferred_job_does_not_start_after_shutdown() -> None:
+    """
+    关闭期间尚未进入业务函数的同步任务不得继续执行。
+    """
+    with pytest.raises(RuntimeError, match="尚未开始"):
+        invoke_background({}, lambda: False, lambda: pytest.fail("任务不应执行"))
+
+
+def test_sync_background_wait_is_bounded_and_resets_inherited_priority() -> None:
+    """
+    同步任务按预算避让，并在真正执行时撤销意外继承的交互归属。
+    """
+    job = {"owner": "site"}
+    with priority.interactive_execution():
+        assert invoke_background(job, lambda: True, priority.is_interactive) is False
+        with priority.interactive_execution(enabled=False):
+            started = time.monotonic()
+            assert invoke_background(job, lambda: True, lambda: "completed") == "completed"
+            assert time.monotonic() - started >= 0.05
+    assert job["deferred"] is False
+
+
+@pytest.mark.asyncio
+async def test_sync_entry_on_loop_does_not_block_for_defer() -> None:
+    """
+    兼容的主循环同步入口不能再同步等待交互结束。
+    """
+    with priority.interactive_execution():
+        assert invoke_background({}, lambda: True, priority.is_interactive) is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("scheduled", [False, True])
+async def test_agent_process_binds_interactive_only_for_foreground(monkeypatch, scheduled) -> None:
+    """
+    真实 Agent 入口在交互轮次绑定优先级，定时任务即使带回传渠道也不升级。
+    """
+    agent = MoviePilotAgent("priority-test", user_id="1", channel="Web", source="web-agent")
+    observed = []
+
+    async def process(_message, **_kwargs):
+        """
+        代替模型推理记录实际调用上下文。
+        """
+        observed.append(priority.is_interactive())
+        return "done"
+
+    monkeypatch.setattr(agent, "_process", process)
+    scope = TerminalScope("1", "run", "scheduled") if scheduled else None
+    assert await agent.process("message", terminal_scope=scope) == "done"
+    assert observed == [not scheduled]
+    assert not priority.is_interactive()
+    assert await agent.cleanup()
+
+
+@pytest.mark.asyncio
+async def test_model_request_timeout_and_disabled_timeout(monkeypatch) -> None:
+    """
+    模型请求超时包含内部重试，取消真正传入 handler；0 保留无限等待语义。
+    """
+    from app.agent.middleware import config as config_module
+
+    middleware = RuntimeConfigMiddleware()
+    monkeypatch.setattr(middleware, "modify_request", lambda request: request)
+    monkeypatch.setattr(config_module, "get_runtime_setting", lambda *_args: 0.01)
+    cancelled = asyncio.Event()
+
+    async def handler(_request):
+        """
+        持有模型请求直到超时，并记录取消传播。
+        """
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cancelled.set()
+
+    with pytest.raises(TimeoutError):
+        await middleware.awrap_model_call(None, handler)
+    assert cancelled.is_set()
+    monkeypatch.setattr(config_module, "get_runtime_setting", lambda *_args: 0)
+    assert await middleware.awrap_model_call(None, AsyncMock(return_value="done")) == "done"
+
+
+def test_timeout_notice_requires_checking_existing_write_status() -> None:
+    """
+    超时提示必须说明操作仍可能收尾，避免用户直接重复执行写入。
+    """
+    message = MoviePilotAgent._friendly_execution_error_message(TimeoutError())
+    assert "超时" in message
+    assert "先确认状态" in message
+
+
+@pytest.mark.asyncio
+async def test_chat_admission_keeps_foreground_capacity_when_background_is_full() -> None:
+    """
+    聊天写入的上层准入同样隔离，避免请求到达保留 DB worker 前就被后台挤出。
+    """
+    entered = asyncio.Event()
+    release = asyncio.Event()
+
+    class Executor:
+        """
+        用异步屏障隔离数据库 I/O，观察聊天服务自己的准入。
+        """
+
+        async def run(self, operation):
+            """
+            后台持有容量，交互直接完成；不访问真实仓储。
+            """
+            if not priority.is_interactive():
+                entered.set()
+                await release.wait()
+
+    service = AgentChatPersistenceService(
+        repository=lambda _session: None, async_executor=Executor(),
+        sync_transaction=lambda operation: None, capacity=1,
+    )
+    background = asyncio.create_task(service._run_write("background", lambda _repository: None))
+    try:
+        await entered.wait()
+        with priority.interactive_execution():
+            await service._run_write("interactive", lambda _repository: None)
+        assert not background.done()
+        assert service._pending_writes == 1
+        assert service._pending_interactive_writes == 0
+    finally:
+        release.set()
+        await background
+        await service.shutdown()
+    assert service._pending_writes == 0
+
+
+@pytest.mark.asyncio
+async def test_scheduler_applies_default_timeout_and_finishes_timed_out_job(monkeypatch) -> None:
+    """
+    默认超时只填补协程任务的空声明，超时取消后由原 owner 发布失败终态。
+    """
+    owner = scheduler_execution.SchedulerExecutionOwner()
+    owner._registry = SimpleNamespace(assign_generation=Mock())
+    owner._lifecycle_state = "running"
+    owner._finish_job = AsyncMock()
+    owner._handle_job_error = Mock()
+    monkeypatch.setattr(scheduler_execution, "get_runtime_setting", lambda *_args: 7)
+    cancelled = asyncio.Event()
+
+    async def task():
+        """
+        模拟需要收到取消才能结束的自动任务。
+        """
+        assert not priority.is_interactive()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            cancelled.set()
+
+    job = {"func": task, "owner": "site", "timeout_seconds": None}
+    owner._assign_job_generation("job", job)
+    assert job["timeout_seconds"] == 7
+    job["timeout_seconds"] = 0.01
+    owner._assign_job_generation("job", job)
+    assert job["timeout_seconds"] == 0.01
+    with priority.interactive_execution():
+        await owner._run_coro_job(task, "job", job)
+        assert priority.is_interactive()
+    assert cancelled.is_set()
+    assert owner._finish_job.await_args.kwargs["success"] is False
+    assert "超时" in owner._finish_job.await_args.kwargs["error"]
+    monkeypatch.setattr(scheduler_execution, "get_runtime_setting", lambda *_args: 0)
+    unlimited = {"func": task}
+    owner._assign_job_generation("unlimited", unlimited)
+    assert unlimited["timeout_seconds"] is None
--- a/tests/test_runtime_execution.py
+++ b/tests/test_runtime_execution.py
@@ -18,6 +18,7 @@ from app.runtime.execution import (
    await_task_to_terminal,
    run_in_threadpool_to_completion,
 )
+from app.runtime.priority import background_thread_limiter

 PROJECT_ROOT = Path(__file__).resolve().parents[1]

@@ -157,7 +158,8 @@ async def test_threadpool_capacity_is_held_until_cancelled_call_finishes() -> No
        await asyncio.sleep(0)

        assert first.done() is False
-        assert limiter.borrowed_tokens == 1
+        assert background_thread_limiter().borrowed_tokens == 1
+        assert limiter.borrowed_tokens == 0

        second = asyncio.create_task(
            run_in_threadpool_to_completion(blocking_call, second_started)
--- a/tests/test_scheduler_cache_expiry.py
+++ b/tests/test_scheduler_cache_expiry.py
@@ -130,6 +130,7 @@ def test_clear_cache_is_manual_only(monkeypatch):
    缓存清理任务应仅手动执行，不注册到调度器自动运行。
    """
    background_scheduler = _BackgroundSchedulerStub()
+    executor_groups = {}
    services = Mock()
    monkeypatch.setattr(scheduler_catalog, "get_plugin_manager", lambda: Mock())
    monkeypatch.setattr(
@@ -140,7 +141,7 @@ def test_clear_cache_is_manual_only(monkeypatch):
    monkeypatch.setattr(
        scheduler_catalog,
        "BackgroundScheduler",
-        lambda **kwargs: background_scheduler,
+        lambda **kwargs: executor_groups.update(kwargs["executors"]) or background_scheduler,
    )
    monkeypatch.setattr(Scheduler, "stop", lambda self: None)
    monkeypatch.setattr(Scheduler, "init_workflow_jobs", lambda self: None)
@@ -151,7 +152,7 @@ def test_clear_cache_is_manual_only(monkeypatch):
        "get_scheduler_runtime_config",
        lambda: SchedulerRuntimeConfig(
            False, "Asia/Shanghai", 1, False, "", 0, None, False, 24,
-            "rss", 30, False, 0, 0, False, None, False, None,
+            "rss", 30, False, 0, 0, True, 1, False, None,
        ),
    )

@@ -177,6 +178,10 @@ def test_clear_cache_is_manual_only(monkeypatch):
    outbox_job = next(job for job in background_scheduler.jobs if job["id"] == "outbox_dispatch")
    assert outbox_job["name"] == "重试未完成的后台处理"
    assert background_scheduler.started is True
+    assert set(executor_groups) == {"default", "agent"}
+    heartbeat = next(job for job in background_scheduler.jobs if job["id"] == "agent_heartbeat")
+    assert heartbeat["executor"] == "agent"
+    assert scheduler._jobs["agent_heartbeat"]["execution_group"] == "agent"


 def test_user_auth_refreshes_plugin_routes_after_runtime_reinitialization(monkeypatch):