Merge pull request #7190 from JontyLee/fix/scheduler-job-timeout-budget

fix(scheduler): 让声明为不限时的协程任务预算真正生效

app/application/scheduling.py

@@ -54,7 +54,8 @@ class JobSpec:
     :param func: 调度触发时调用的业务函数
     :param owner: 任务生命周期所属的业务组件标识
     :param overlap: 前一次仍在执行时的重入策略，默认跳过新触发
-    :param timeout_seconds: 声明的执行超时秒数，None 表示不声明超时
+    :param timeout_seconds: 执行超时预算：正数为总时长上限，0 表示显式不限时，
+        None 表示未声明并由全局默认补齐
     :param manual: 是否允许通过管理界面手动触发
     :param recovery: 进程重启后的恢复策略，默认等待下一次计划触发
     :param kwargs: 每次调用业务函数时使用的关键字参数

app/scheduler/catalog.py

@@ -69,6 +69,8 @@ def _subscription_search_job_specs(
             "恢复订阅搜索队列",
             services.resume_subscribe_search,
             "subscription",
+            # 单轮耗时由队列自身限额决定，取消又杀不掉同步消费者，因此不叠加总时长预算。
+            timeout_seconds=0,
             recovery=JobRecoveryPolicy.DURABLE_QUEUE,
         ),
     )

app/scheduler/execution.py

@@ -50,10 +50,16 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
 
     def _assign_job_generation(self, job_id: str, job: dict[str, Any]) -> None:
         """
-        把注册 generation 写入可变运行时状态。
+        把注册 generation 写入可变运行时状态，并补齐协程任务的执行预算。
+
+        声明为 0 的任务保留不限时：托管协程消费者无法靠取消真正停止，给它总时长上限
+        只会把正常长轮次记成失败，真实收尾仍然要等消费者自己结束。
         """
         self._registry.assign_generation(job_id, job)
-        if inspect.iscoroutinefunction(job.get("func")) and job.get("timeout_seconds") is None:
+        declared = job.get("timeout_seconds")
+        if declared == 0:
+            job["timeout_seconds"] = None
+        elif inspect.iscoroutinefunction(job.get("func")) and declared is None:
             timeout = int(get_runtime_setting("SCHEDULER_JOB_TIMEOUT", 3600))
             job["timeout_seconds"] = timeout if timeout > 0 else None
 
@@ -129,14 +135,26 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
         return job
 
     @staticmethod
-    def _handle_job_error(job_id: str, job: dict[str, Any], error: Exception) -> None:
+    def _handle_job_error(
+        job_id: str,
+        job: dict[str, Any],
+        error: Exception,
+        detail: Optional[str] = None,
+    ) -> None:
         """
         记录定时任务执行异常并发送系统错误事件。
+
+        :param job_id: 调度任务标识
+        :param job: 任务运行时状态
+        :param error: 捕获到的异常，用于取异常类型与堆栈
+        :param detail: 面向日志、通知和事件的错误文本；省略时取异常文本，
+            用于补足 asyncio 超时这类 str() 为空的标准异常
         """
-        logger.error(f"定时任务 {job.get('name')} 执行失败：{str(error)} - {traceback.format_exc()}")
+        message = detail or str(error)
+        logger.error(f"定时任务 {job.get('name')} 执行失败：{message} - {traceback.format_exc()}")
         _message_helper_factory().put(
             title=f"{job.get('name')} 执行失败",
-            message=str(error),
+            message=message,
             role="system",
             level="error",
         )
@@ -146,7 +164,7 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
                 "type": "scheduler",
                 "scheduler_id": job_id,
                 "scheduler_name": job.get("name"),
-                "error": str(error),
+                "error": message,
                 "traceback": traceback.format_exc(),
             },
         )
@@ -196,7 +214,7 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
         except asyncio.TimeoutError as err:
             success = False
             error = f"任务执行超时（{job.get('timeout_seconds')} 秒）"
-            self._handle_job_error(job_id=job_id, job=job, error=err)
+            self._handle_job_error(job_id=job_id, job=job, error=err, detail=error)
         except asyncio.CancelledError:
             success = False
             error = "任务已取消"

docs/agent-execution.md

@@ -52,7 +52,8 @@ APScheduler 3 的标准线程池没有任意数值 `priority` 的调度能力。
 - `AGENT_LLM_TIMEOUT`：单次根 Agent 模型请求总预算，默认 180 秒，覆盖流式与非流式生成
   以及该请求的供应商重试；`0` 表示不限时。工具执行时间由原 `LLM_TOOL_TIMEOUT` 管理。
 - `SCHEDULER_JOB_TIMEOUT`：协程调度任务的默认执行预算，默认 3600 秒，等待避让不计入执行时间；
-  `0` 表示不限时。显式 `JobSpec.timeout_seconds` 优先于此默认值。
+  `0` 表示不限时。显式 `JobSpec.timeout_seconds` 优先于此默认值：正数是总时长上限，`0`
+  表示该任务显式不限时，缺省才继承此默认值。
 
 超时通过协程取消收口，不能杀死同步线程或撤销已经发生的外部副作用。
 数据库和线程调用仍等待实际操作收尾后释放容量，因而收尾时间可能超过超时预算。

tests/test_scheduler_contracts.py

@@ -1,9 +1,11 @@
 """
-Scheduler 声明、catalog 与 execution state 测试。
+Scheduler 声明、catalog、execution state 与协程执行预算语义测试。
 """
 
 import asyncio
-from unittest.mock import MagicMock
+import threading
+from typing import Any
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -14,6 +16,9 @@ from app.application.scheduling import (
     JobRecoveryPolicy,
     JobSpec,
 )
+from app.scheduler import execution as scheduler_execution
+from app.scheduler.facade import Scheduler
+from app.scheduler.registry import ExecutionRegistry
 
 
 def test_job_spec_exports_complete_legacy_state() -> None:
@@ -99,3 +104,116 @@ def test_manual_job_declares_restart_policy() -> None:
 
     assert state["manual"] is True
     assert state["recovery"] == JobRecoveryPolicy.MANUAL_ONLY.value
+
+
+async def _coro_job() -> None:
+    """
+    协程任务样本，只提供可内省的异步签名。
+    """
+    return None
+
+
+def _resolved_timeout(spec: JobSpec, default_seconds: int) -> Any:
+    """
+    用最小 Scheduler owner 走一遍预算补齐路径，返回运行状态里的最终超时值。
+
+    :param spec: 待解析的任务声明
+    :param default_seconds: 模拟的全局协程默认预算
+    :return: 补齐后的 `timeout_seconds`
+    """
+    scheduler = object.__new__(Scheduler)
+    scheduler._lock = threading.RLock()
+    scheduler._registry = ExecutionRegistry(scheduler._lock)
+    state = spec.to_runtime_state()
+    with patch.object(
+        scheduler_execution,
+        "get_runtime_setting",
+        return_value=default_seconds,
+    ):
+        scheduler._assign_job_generation(spec.job_id, state)
+    return state["timeout_seconds"]
+
+
+def test_declared_zero_timeout_keeps_job_unlimited() -> None:
+    """
+    显式声明 0 秒的协程任务不得继承全局预算：托管消费者取消不掉，套上总时长上限只会
+    把正常长轮次记成超时失败。
+    """
+    spec = JobSpec(
+        "queue",
+        "队列恢复",
+        _coro_job,
+        "subscription",
+        timeout_seconds=0,
+        recovery=JobRecoveryPolicy.DURABLE_QUEUE,
+    )
+
+    assert _resolved_timeout(spec, default_seconds=3600) is None
+
+
+def test_undeclared_coroutine_timeout_inherits_global_default() -> None:
+    """
+    未声明预算的协程任务继承全局默认，同步任务保持不限时。
+    """
+    assert _resolved_timeout(
+        JobSpec("coro", "协程任务", _coro_job, "runtime"),
+        default_seconds=120,
+    ) == 120
+    assert _resolved_timeout(
+        JobSpec("sync", "同步任务", MagicMock(), "runtime"),
+        default_seconds=120,
+    ) is None
+
+
+def test_positive_declared_timeout_wins_over_global_default() -> None:
+    """
+    正数声明是本次执行的总时长上限，不被全局默认覆盖。
+    """
+    assert _resolved_timeout(
+        JobSpec("bounded", "有界任务", _coro_job, "runtime", timeout_seconds=30),
+        default_seconds=3600,
+    ) == 30
+
+
+def test_timeout_error_keeps_descriptive_message() -> None:
+    """
+    `asyncio` 超时异常的文本为空，日志、系统通知和事件都必须带上可读的超时说明。
+    """
+    job = {"name": "恢复订阅搜索队列", "timeout_seconds": 3600}
+    helper = MagicMock()
+    with (
+        patch.object(scheduler_execution.logger, "error") as error_log,
+        patch.object(scheduler_execution, "_message_helper_factory", new=lambda: helper),
+        patch.object(scheduler_execution.eventmanager, "send_event") as send_event,
+    ):
+        Scheduler._handle_job_error(
+            job_id="subscribe_search_queue",
+            job=job,
+            error=asyncio.TimeoutError(),
+            detail="任务执行超时（3600 秒）",
+        )
+
+    expected = "任务执行超时（3600 秒）"
+    assert expected in error_log.call_args.args[0]
+    assert helper.put.call_args.kwargs["message"] == expected
+    assert send_event.call_args.args[1]["error"] == expected
+
+
+def test_job_error_falls_back_to_exception_text() -> None:
+    """
+    未提供说明时沿用异常文本，普通异常的错误链路不变。
+    """
+    helper = MagicMock()
+    with (
+        patch.object(scheduler_execution.logger, "error") as error_log,
+        patch.object(scheduler_execution, "_message_helper_factory", new=lambda: helper),
+        patch.object(scheduler_execution.eventmanager, "send_event"),
+    ):
+        Scheduler._handle_job_error(
+            job_id="job",
+            job={"name": "普通任务"},
+            error=RuntimeError("站点无响应"),
+        )
+
+    assert "站点无响应" in error_log.call_args.args[0]
+    assert helper.put.call_args.kwargs["message"] == "站点无响应"