diff --git a/app/adapters/web/priority.py b/app/adapters/web/priority.py new file mode 100644 index 000000000..bb7c2cb89 --- /dev/null +++ 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) diff --git a/app/agent/api/executor.py b/app/agent/api/executor.py index 7ad3e7aaf..f8eaa1ff2 100644 --- 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 diff --git a/app/agent/middleware/config.py b/app/agent/middleware/config.py index 7baa4b0b5..eb0e15269 100644 --- 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"] diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index 67cf7a9a0..a7d9395ad 100644 --- 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 ) diff --git a/app/agent/tools/base.py b/app/agent/tools/base.py index f1eea3a17..5f54a9386 100644 --- 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: diff --git a/app/application/messaging/chat.py b/app/application/messaging/chat.py index 60c16a574..b67c1ecb8 100644 --- 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 diff --git a/app/application/scheduling.py b/app/application/scheduling.py index 538ff7515..3497625ad 100644 --- 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, diff --git a/app/application/settings/contract.py b/app/application/settings/contract.py index 02b925efe..8391bf20c 100644 --- 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",) diff --git a/app/application/settings/resources/system_settings_catalog.json b/app/application/settings/resources/system_settings_catalog.json index c799a6d2c..b0f5412b4 100644 --- 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", diff --git a/app/db/worker.py b/app/db/worker.py index 6bda14b33..39a324e96 100644 --- 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: diff --git a/app/factory.py b/app/factory.py index be816f2f2..0b002945b 100644 --- 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") diff --git a/app/runtime/config.py b/app/runtime/config.py index 4a3329196..3db42a6c1 100644 --- 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,留空使用内置机器人;用户个人选择优先') diff --git a/app/runtime/execution.py b/app/runtime/execution.py index e0d7421b6..18e8f1e24 100644 --- 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 diff --git a/app/runtime/priority.py b/app/runtime/priority.py new file mode 100644 index 000000000..b8b620485 --- /dev/null +++ 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)) diff --git a/app/runtime/thread.py b/app/runtime/thread.py index 91e301424..959530c8f 100644 --- 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) diff --git a/app/scheduler/admission.py b/app/scheduler/admission.py new file mode 100644 index 000000000..9b80f9a20 --- /dev/null +++ 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) diff --git a/app/scheduler/catalog.py b/app/scheduler/catalog.py index 9726c48b8..91a525ade 100644 --- 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"}, diff --git a/app/scheduler/execution.py b/app/scheduler/execution.py index 16a14cd7e..62092f34f 100644 --- 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, diff --git a/app/scheduler/reconcile.py b/app/scheduler/reconcile.py index c27b11185..6d512c904 100644 --- 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, diff --git a/app/schemas/dashboard.py b/app/schemas/dashboard.py index 6c7dbf070..b063cd790 100644 --- 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="当前进度文本") diff --git a/docs/agent-execution.md b/docs/agent-execution.md new file mode 100644 index 000000000..2edeae7e9 --- /dev/null +++ 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` 提供,修改后重启生效。 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index a030671f4..232cc0886 100644 --- 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) | diff --git a/docs/development-setup.md b/docs/development-setup.md index b6e8d0fc8..e0b77f0b9 100644 --- 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 许可在 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 585f7c4ce..fa95eca49 100644 --- 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 | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 2e6c6c99c..2f834dbac 100644 --- 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", diff --git a/tests/fixtures/architecture/startup-performance-baseline.json b/tests/fixtures/architecture/startup-performance-baseline.json index 067e0a47c..38d052092 100644 --- 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, diff --git a/tests/test_interactive_priority.py b/tests/test_interactive_priority.py new file mode 100644 index 000000000..4236f0c6e --- /dev/null +++ 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 diff --git a/tests/test_runtime_execution.py b/tests/test_runtime_execution.py index ea6d78d23..2a0940a82 100644 --- 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) diff --git a/tests/test_scheduler_cache_expiry.py b/tests/test_scheduler_cache_expiry.py index 93cdd3041..c27d2fa63 100644 --- 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):