fix: isolate runtime scheduler with portable watchdog (#2219)

* fix: isolate runtime scheduler with portable watchdog

* fix: harden runtime scheduler watchdog shutdown

* fix: prevent scheduled analysis after shutdown

* fix: fence scheduler shutdown and harden watchdog

---------

Co-authored-by: wvyan <13724623+wvyan@users.noreply.github.com>

.env.example

@@ -821,6 +821,8 @@ SCHEDULE_ENABLED=false
 SCHEDULE_TIME=18:00
 # 多时间执行列表（逗号分隔，留空时使用 SCHEDULE_TIME）
 SCHEDULE_TIMES=
+# Web/API runtime scheduler 单次分析硬超时（秒，最小 60 秒）
+DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS=2700
 # 定时模式启动时是否立即执行一次分析（true/false）
 # 若未显式设置，定时模式会沿用 RUN_IMMEDIATELY 的运行时覆盖语义以兼容旧配置
 SCHEDULE_RUN_IMMEDIATELY=true

api/v1/endpoints/system_config.py

@@ -63,7 +63,7 @@ def get_scheduler_status(
 @router.post(
     "/scheduler/run-now",
     summary="Run scheduled analysis now",
-    description="Trigger one scheduled analysis run in the current process.",
+    description="Trigger one isolated analysis run through the current runtime scheduler service.",
 )
 def run_scheduler_now(
     scheduler: RuntimeSchedulerService = Depends(get_runtime_scheduler_service),

apps/dsa-web/src/locales/settingsHelp.ts

@@ -673,10 +673,11 @@ const settingsHelpZhCN: SettingsHelpMap = {
   'settings.system.schedule': {
     title: '定时任务',
     summary: '控制是否启用每日定时分析以及启动时是否立即执行一次。',
-    usage: 'SCHEDULE_TIME 使用 HH:MM 24 小时格式；SCHEDULE_TIMES 可配置逗号分隔的多个 HH:MM 时间点；SCHEDULE_ENABLED 控制 runtime scheduler 是否启用。',
+    usage: 'SCHEDULE_TIME 使用 HH:MM 24 小时格式；SCHEDULE_TIMES 可配置逗号分隔的多个 HH:MM 时间点；SCHEDULE_ENABLED 控制 runtime scheduler 是否启用；DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS 控制单次 Web/API 定时分析的硬超时，默认 2700 秒、最小 60 秒。',
     valueNotes: [
       '已运行的 schedule 模式会在下一轮调度检查中读取新的 SCHEDULE_TIME / SCHEDULE_TIMES 并重建 daily jobs。',
       'WebUI/API/Desktop 长运行进程保存 SCHEDULE_ENABLED、SCHEDULE_TIME 或 SCHEDULE_TIMES 后会按新配置启停或重建 runtime scheduler。',
+      '修改 DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS 后，下一次分析会使用新值，无需重建 daily jobs；超时会终止独立分析进程。',
       '定时任务触发时会读取当前保存的 STOCK_LIST。',
     ],
     impact: ['影响 schedule 模式下自动分析频率、启动行为和通知推送时间。'],
@@ -1860,10 +1861,11 @@ const settingsHelpEnUS: SettingsHelpMap = {
   'settings.system.schedule': {
     title: 'Schedule',
     summary: 'Controls daily scheduled analysis and whether startup runs immediately.',
-    usage: 'SCHEDULE_TIME uses HH:MM 24-hour format. SCHEDULE_TIMES accepts comma-separated HH:MM values. SCHEDULE_ENABLED controls whether the runtime scheduler is enabled.',
+    usage: 'SCHEDULE_TIME uses HH:MM 24-hour format. SCHEDULE_TIMES accepts comma-separated HH:MM values. SCHEDULE_ENABLED controls whether the runtime scheduler is enabled. DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS sets the hard timeout for each Web/API scheduled analysis (default 2700 seconds, minimum 60).',
     valueNotes: [
       'An already-running schedule mode reads new SCHEDULE_TIME / SCHEDULE_TIMES values on the next scheduler check and rebuilds the daily jobs.',
       'Long-running WebUI/API/Desktop processes start, stop, or rebuild the runtime scheduler after saving SCHEDULE_ENABLED, SCHEDULE_TIME, or SCHEDULE_TIMES.',
+      'After DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS changes, the next analysis uses the new value without rebuilding daily jobs; the isolated analysis process is terminated on timeout.',
       'Scheduled runs read the currently saved STOCK_LIST.',
     ],
     impact: ['Affects automatic analysis frequency, startup behavior, and notification timing in schedule mode.'],

apps/dsa-web/src/utils/systemConfigI18n.ts

@@ -165,6 +165,7 @@ const fieldTitleMap: Record<string, string> = {
   REPORT_SHOW_LLM_MODEL: '显示分析模型',
   MAX_WORKERS: '最大并发线程数',
   SCHEDULE_ENABLED: '启用定时任务',
+  DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS: '定时分析硬超时（秒）',
   SCHEDULE_TIMES: '定时执行时间',
   SCHEDULE_RUN_IMMEDIATELY: '启动定时模式后立即执行',
   TRADING_DAY_CHECK_ENABLED: '启用交易日检查',
@@ -334,6 +335,7 @@ const fieldDescriptionMap: Record<string, string> = {
   REPORT_SHOW_LLM_MODEL: '在通知报告底部显示本次分析使用的 LLM 模型名称；关闭后隐藏运行时模型信息。仅影响展示，不会影响 provider/model/Base URL、运行时模型保存、迁移或清理。',
   MAX_WORKERS: '异步任务队列最大并发数。配置保存后，队列空闲时会自动应用；繁忙时延后生效。',
   SCHEDULE_ENABLED: '启用每日定时分析任务。',
+  DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS: 'Web/API 定时分析的最长运行时间；超时后终止独立分析进程，最小 60 秒。',
   SCHEDULE_TIMES: '多个每日定时任务执行时间，使用逗号分隔；Web 设置页会以时间控件维护。',
   SCHEDULE_RUN_IMMEDIATELY: '仅影响 schedule 模式启动时是否立即触发一次任务；保存 Web 设置不会触发执行。',
   TRADING_DAY_CHECK_ENABLED: '启用交易日校验，非交易日自动跳过定时分析。',

apps/dsa-web/tests/system_config_i18n.test.ts

@@ -70,6 +70,7 @@ const requiredLocalizedKeys = [
   'NOTIFICATION_MIN_SEVERITY',
   'NOTIFICATION_DAILY_DIGEST_ENABLED',
   'SCHEDULE_ENABLED',
+  'DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS',
   'SCHEDULE_RUN_IMMEDIATELY',
   'TRADING_DAY_CHECK_ENABLED',
   'WEBUI_HOST',
@@ -118,6 +119,20 @@ describe('systemConfigI18n required key coverage', () => {
     expect(en?.summary).toContain('HMAC');
     expect(en?.notes?.join(' ')).toContain('Do not');
   });
+
+  it('documents the runtime scheduler timeout in both UI languages', () => {
+    const zh = getSettingsHelpContent('settings.system.schedule', undefined, 'zh-CN');
+    const en = getSettingsHelpContent('settings.system.schedule', undefined, 'en');
+    const zhText = JSON.stringify(zh);
+    const enText = JSON.stringify(en);
+
+    expect(zhText).toContain('DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS');
+    expect(zhText).toContain('2700');
+    expect(zhText).toContain('60');
+    expect(enText).toContain('DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS');
+    expect(enText).toContain('2700');
+    expect(enText).toContain('60');
+  });
 });
 
 describe('systemConfigI18n option label localization', () => {

docs/CHANGELOG.md

@@ -32,6 +32,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [改进] AIHubMix 注册与引流链接统一使用 inferera.com，改善中国大陆网络直连体验。
 - [修复] 单股推送模式在未配置通知渠道时仍会落盘本地个股报告；CLI 启动分析若因空股票列表、个股结果全失败或本地报告保存失败而未生成报告，会显式返回失败并记录原因。
 - [修复] 合并推送模式下即使个股汇总报告落盘失败，仍会先发送已有的合并通知；仅启用大盘复盘但最终未生成任何复盘内容时，分析任务会显式返回失败。
+- [修复] Web/API runtime scheduler 使用跨平台独立进程执行分析，并在默认 45 分钟硬超时或服务停止后清理进程树；停止返回后不再派发新的自动任务，避免一次卡死阻断后续调度。
 - [修复] SearXNG 公共实例发现的默认值由启用改为关闭：公共实例普遍存在限流、下线或不返回 JSON 的情况，默认开启会让未配置搜索 key 的用户每次分析多耗 30~60 秒且新闻面最终为空。运行时默认值、配置模板、中英文档与工作流诊断同步调整；显式设为 true 的用户行为不变。
 - [改进] 新闻检索未执行或零命中时，报告中如实标注结论未纳入新闻面证据：零命中与「未配置搜索渠道」使用各自独立的文案，覆盖日报 / dashboard / brief / 个股 / 企业微信与模板渲染的详细与摘要分支、历史报告与分享导出、报告详情 API 与 Web 报告详情页，并按 `zh` / `en` / `ko` 分别本地化。此前该情况下消息面章节直接消失，读者无从区分「确实没有新闻」与「检索静默失败」。披露以本次分析实际收到的消息面证据为准，涵盖实时检索、社交情绪与本地已落库的资讯池三路来源；搜索命中数仅用于在确无证据时说明原因（未配置渠道 / 检索零命中），避免把已用到本地或社交证据的分析误报成「未纳入新闻面证据」。Agent 模式的命中数取自 Agent 实际消费的搜索工具结果，而非分析结束后为持久化情报而补打的查询。
 

docs/full-guide.md

@@ -492,6 +492,7 @@ daily_stock_analysis/
 | `SCHEDULE_ENABLED` | 启用定时任务 | `false` |
 | `SCHEDULE_TIME` | 定时执行时间 | `18:00` |
 | `SCHEDULE_TIMES` | 多个定时执行时间，逗号分隔；为空时使用 `SCHEDULE_TIME` | 空 |
+| `DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS` | Web/API runtime scheduler 单次分析硬超时（秒，最小 60 秒）；超时后终止独立分析进程，不阻塞后续任务 | `2700` |
 | `LOG_DIR` | 日志目录 | `./logs` |
 | `SAVE_CONTEXT_SNAPSHOT` | 保存分析历史 `context_snapshot`；设为 `false` 时新历史不保存 enhanced_context、market_phase_summary、AnalysisContextPack overview 或诊断快照，但不关闭当次 Prompt 低敏摘要 | `true` |
 
@@ -810,6 +811,7 @@ python main.py --schedule --no-run-immediately
 | `SCHEDULE_ENABLED` | 是否启用定时任务 | `false` | `true` |
 | `SCHEDULE_TIME` | 每日执行时间 (HH:MM) | `18:00` | `09:30` |
 | `SCHEDULE_TIMES` | 多个每日执行时间，逗号分隔；为空时使用 `SCHEDULE_TIME` | 空 | `09:20,12:30,15:10,18:00` |
+| `DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS` | Web/API runtime scheduler 单次分析硬超时（秒，最小 60 秒） | `2700` | `3600` |
 | `SCHEDULE_RUN_IMMEDIATELY` | 定时模式启动时是否立即运行一次；未显式设置时沿用 `RUN_IMMEDIATELY` 的运行时覆盖语义 | `true` | `false` |
 | `RUN_IMMEDIATELY` | 非定时模式启动时是否立即运行一次；同时作为未显式设置 `SCHEDULE_RUN_IMMEDIATELY` 时的 legacy 回退 | `true` | `false` |
 | `TRADING_DAY_CHECK_ENABLED` | 交易日检查：非交易日跳过执行；设为 `false` 可强制执行 | `true` | `false` |

docs/full-guide_EN.md

@@ -414,6 +414,7 @@ For the notification baseline, diagnostics, and deployment notes, see [Notificat
 | `SCHEDULE_ENABLED` | Enable scheduled tasks | `false` |
 | `SCHEDULE_TIME` | Scheduled execution time | `18:00` |
 | `SCHEDULE_TIMES` | Multiple scheduled execution times, comma-separated; falls back to `SCHEDULE_TIME` when empty | empty |
+| `DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS` | Hard timeout in seconds for each Web/API runtime scheduler analysis (minimum 60); the isolated analysis process is terminated on timeout so later runs can continue | `2700` |
 | `SCHEDULE_RUN_IMMEDIATELY` | Run once immediately when scheduler mode starts; when unset it keeps following the legacy `RUN_IMMEDIATELY` runtime override | `true` |
 | `RUN_IMMEDIATELY` | Run once immediately for non-scheduler startup; also acts as the legacy fallback when `SCHEDULE_RUN_IMMEDIATELY` is unset | `true` |
 | `LOG_DIR` | Log directory | `./logs` |

src/core/config_registry.py

@@ -3024,6 +3024,31 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
         ],
         "warning_codes": ["local_timezone"],
     },
+    "DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS": {
+        "title": "Runtime Scheduler Timeout",
+        "description": "Hard timeout in seconds for each Web/API scheduled analysis.",
+        "category": "system",
+        "data_type": "integer",
+        "ui_control": "number",
+        "is_sensitive": False,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": "2700",
+        "options": [],
+        "validation": {"min": 60},
+        "display_order": 12,
+        "help_key": "settings.system.schedule",
+        "examples": [
+            "DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS=2700",
+        ],
+        "docs": [
+            {
+                "label": "Full guide: configuration",
+                "href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#其他配置",
+            },
+        ],
+        "warning_codes": [],
+    },
     "HTTP_PROXY": {
         "title": "HTTP Proxy",
         "description": "Optional HTTP proxy endpoint.",

src/scheduler.py

@@ -123,7 +123,9 @@ class Scheduler:
         self._daily_job: Optional[Any] = None
         self._daily_jobs: List[Any] = []
         self._background_tasks: List[Dict[str, Any]] = []
+        self._lifecycle_lock = threading.Lock()
         self._running = False
+        self._stop_requested = False
 
     def set_daily_task(self, task: Callable, run_immediately: bool = True):
         """
@@ -378,20 +380,33 @@ class Scheduler:
                 continue
             self._start_background_task(entry)
 
+    def _dispatch_background_tasks_if_running(self) -> bool:
+        """Dispatch due background tasks before a concurrent stop can return."""
+        with self._lifecycle_lock:
+            if not self._running or self.shutdown_handler.should_shutdown:
+                return False
+            self._run_background_tasks()
+            return True
+
     def run(self):
         """
         运行调度器主循环
 
         阻塞运行，直到收到退出信号
         """
-        self._running = True
+        with self._lifecycle_lock:
+            if self._stop_requested:
+                logger.info("调度器已停止，忽略迟到的启动请求")
+                return
+            self._running = True
         logger.info("调度器开始运行...")
         logger.info(f"下次执行时间: {self._get_next_run_time()}")
 
         while self._running and not self.shutdown_handler.should_shutdown:
             self._refresh_daily_schedule_if_needed()
             self.schedule.run_pending()
-            self._run_background_tasks()
+            if not self._dispatch_background_tasks_if_running():
+                break
             time.sleep(30)  # 每30秒检查一次
 
             # 每小时打印一次心跳
@@ -410,7 +425,9 @@ class Scheduler:
 
     def stop(self):
         """停止调度器"""
-        self._running = False
+        with self._lifecycle_lock:
+            self._stop_requested = True
+            self._running = False
         self._cancel_daily_job()
 
 

src/services/runtime_scheduler.py

@@ -4,10 +4,16 @@
 from __future__ import annotations
 
 import logging
+import multiprocessing
 import os
+import signal
+import subprocess
 import threading
 import _thread
+import time
 from datetime import datetime
+from functools import partial
+from queue import Empty
 from types import SimpleNamespace
 from typing import Any, Callable, Dict, List, Optional, Set
 
@@ -20,6 +26,8 @@ RUNTIME_SCHEDULER_FORCE_ENABLED_ENV = "DSA_RUNTIME_SCHEDULER_FORCE_ENABLED"
 RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV = "DSA_RUNTIME_SCHEDULER_RUN_IMMEDIATELY"
 RUNTIME_SCHEDULER_SUPPRESS_START_ENV = "DSA_RUNTIME_SCHEDULER_SUPPRESS_START"
 RUNTIME_SCHEDULER_ARGS_ENV = "DSA_RUNTIME_SCHEDULER_ARGS"
+RUNTIME_SCHEDULER_TIMEOUT_ENV = "DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS"
+DEFAULT_RUNTIME_SCHEDULER_TIMEOUT_SECONDS = 45 * 60
 _RUNTIME_ANALYSIS_LOCK = threading.Lock()
 SCHEDULE_ARGS_OVERRIDE_KEYS = {
     "no_notify",
@@ -51,6 +59,136 @@ def run_with_global_analysis_lock(
     return True
 
 
+def _run_scheduled_analysis_process(
+    result_queue: Any,
+    stock_codes: Optional[List[str]],
+    schedule_args_overrides: Dict[str, Any],
+) -> None:
+    """Run one analysis in a spawn-safe child process."""
+    if os.name == "posix":
+        try:
+            os.setsid()
+        except OSError:
+            # Being a session leader already is equivalent to success. Any
+            # other failure would make process-tree cleanup unsafe, so fail
+            # before analysis can create descendants.
+            if os.getsid(0) != os.getpid():
+                raise
+    service = RuntimeSchedulerService(schedule_args_overrides=schedule_args_overrides)
+    success = service._run_analysis_locked(stock_codes)
+    result_queue.put({"success": success, "error": service._last_error})
+
+
+def _posix_descendant_process_ids(root_pid: int) -> Set[int]:
+    """Return a best-effort snapshot of descendants before the root exits."""
+    try:
+        completed = subprocess.run(
+            ["ps", "-axo", "pid=,ppid="],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return set()
+
+    children_by_parent: Dict[int, List[int]] = {}
+    for line in completed.stdout.splitlines():
+        parts = line.split()
+        if len(parts) != 2:
+            continue
+        try:
+            pid, parent_pid = (int(value) for value in parts)
+        except ValueError:
+            continue
+        children_by_parent.setdefault(parent_pid, []).append(pid)
+
+    descendants: Set[int] = set()
+    pending = list(children_by_parent.get(root_pid, []))
+    while pending:
+        pid = pending.pop()
+        if pid in descendants:
+            continue
+        descendants.add(pid)
+        pending.extend(children_by_parent.get(pid, []))
+    return descendants
+
+
+def _terminate_analysis_process_tree(process: Any) -> None:
+    """Stop an analysis worker and any descendants it created."""
+    root_alive = process.is_alive()
+    process_id = process.pid
+
+    posix_process_groups: Set[int] = set()
+    if os.name == "posix" and process_id:
+        posix_process_groups.add(process_id)
+        current_process_group = os.getpgrp()
+        if root_alive:
+            for descendant_pid in _posix_descendant_process_ids(process_id):
+                try:
+                    descendant_group = os.getpgid(descendant_pid)
+                except ProcessLookupError:
+                    continue
+                if descendant_group != current_process_group:
+                    posix_process_groups.add(descendant_group)
+
+    try:
+        if os.name == "posix" and process_id:
+            for process_group in posix_process_groups:
+                try:
+                    os.killpg(process_group, signal.SIGTERM)
+                except ProcessLookupError:
+                    continue
+            # The spawned worker calls setsid(), but stop/timeout can win the
+            # race before that happens. In that window killpg(worker_pid, ...)
+            # has no target, so also terminate the multiprocessing handle.
+            if process.is_alive():
+                process.terminate()
+        elif os.name == "nt" and process_id:
+            subprocess.run(
+                ["taskkill", "/PID", str(process_id), "/T", "/F"],
+                check=False,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                timeout=10,
+            )
+        elif root_alive:
+            process.terminate()
+    except (OSError, subprocess.SubprocessError):
+        if root_alive:
+            process.terminate()
+
+    process.join(2)
+    if os.name == "posix" and process_id:
+        remaining_process_groups: Set[int] = set()
+        for process_group in posix_process_groups:
+            try:
+                os.killpg(process_group, 0)
+            except ProcessLookupError:
+                continue
+            except PermissionError:
+                pass
+            remaining_process_groups.add(process_group)
+        if not remaining_process_groups:
+            return
+    elif not process.is_alive():
+        return
+
+    try:
+        if os.name == "posix" and process_id:
+            for process_group in remaining_process_groups:
+                try:
+                    os.killpg(process_group, signal.SIGKILL)
+                except ProcessLookupError:
+                    continue
+        else:
+            process.kill()
+    except (OSError, AttributeError):
+        if process.is_alive():
+            process.terminate()
+    process.join(10)
+
+
 def _agent_event_monitor_interval_seconds(config: Config) -> int:
     """Return the validated Event Monitor polling interval in seconds."""
     interval_minutes = getattr(config, "agent_event_monitor_interval_minutes", 5)
@@ -141,6 +279,10 @@ class RuntimeSchedulerService:
         self._last_error: Optional[str] = None
         self._last_skipped_at: Optional[str] = None
         self._last_skip_reason: Optional[str] = None
+        self._analysis_process_target = _run_scheduled_analysis_process
+        self._analysis_process: Optional[Any] = None
+        self._analysis_process_lock = threading.Lock()
+        self._analysis_generation = 0
 
     def _make_schedule_args(self) -> SimpleNamespace:
         defaults = {
@@ -172,7 +314,7 @@ class RuntimeSchedulerService:
         self._last_skip_reason = "analysis_already_running"
         logger.warning("Runtime scheduler skipped run: analysis already running")
 
-    def _run_analysis_locked(self, stock_codes: Optional[List[str]]) -> None:
+    def _run_analysis_locked(self, stock_codes: Optional[List[str]]) -> bool:
         try:
             config = self._reload_config()
             runner = self._task_runner
@@ -186,9 +328,11 @@ class RuntimeSchedulerService:
                 raise RuntimeError("runtime scheduled analysis reported failure")
             self._last_success_at = datetime.now().isoformat()
             self._last_error = None
+            return True
         except Exception as exc:  # noqa: BLE001 - scheduled runs must not kill API process.
             self._last_error = str(exc)
             logger.exception("Runtime scheduled analysis failed: %s", exc)
+            return False
 
     def _run_analysis_once(self, stock_codes: Optional[List[str]] = None) -> bool:
         if not self._run_lock.acquire(blocking=False):
@@ -200,6 +344,150 @@ class RuntimeSchedulerService:
             self._run_lock.release()
         return True
 
+    def _analysis_timeout_seconds(self) -> int:
+        try:
+            value = os.getenv(
+                RUNTIME_SCHEDULER_TIMEOUT_ENV,
+                str(DEFAULT_RUNTIME_SCHEDULER_TIMEOUT_SECONDS),
+            )
+            return max(60, int(value))
+        except ValueError:
+            logger.warning(
+                "Invalid %s; using %ss",
+                RUNTIME_SCHEDULER_TIMEOUT_ENV,
+                DEFAULT_RUNTIME_SCHEDULER_TIMEOUT_SECONDS,
+            )
+            return DEFAULT_RUNTIME_SCHEDULER_TIMEOUT_SECONDS
+
+    def _run_analysis_with_watchdog(
+        self,
+        stock_codes: Optional[List[str]] = None,
+        *,
+        lock_held: bool = False,
+        generation: Optional[int] = None,
+    ) -> None:
+        if not lock_held and not self._run_lock.acquire(blocking=False):
+            self._record_analysis_busy_skip()
+            return
+        if generation is None:
+            with self._analysis_process_lock:
+                generation = self._analysis_generation
+
+        result_queue = None
+        try:
+            context = multiprocessing.get_context("spawn")
+            result_queue = context.Queue()
+            process = context.Process(
+                target=self._analysis_process_target,
+                args=(result_queue, stock_codes, dict(self._schedule_args_overrides)),
+                name="runtime-scheduled-analysis",
+            )
+            timeout = self._analysis_timeout_seconds()
+            with self._analysis_process_lock:
+                if generation != self._analysis_generation:
+                    return
+                process.start()
+                self._analysis_process = process
+                self._last_run_at = datetime.now().isoformat()
+
+            result = None
+            deadline = time.monotonic() + timeout
+            while result is None:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    break
+                try:
+                    result = result_queue.get(timeout=min(0.2, remaining))
+                except Empty:
+                    if not process.is_alive():
+                        deadline = min(deadline, time.monotonic() + 2)
+
+            if result is None and process.is_alive():
+                logger.error(
+                    "Runtime scheduled analysis exceeded %ss; terminating worker",
+                    timeout,
+                )
+                _terminate_analysis_process_tree(process)
+                with self._analysis_process_lock:
+                    if generation != self._analysis_generation:
+                        return
+                    self._last_error = f"runtime scheduled analysis timed out after {timeout}s"
+                return
+
+            if result is None:
+                exit_code = process.exitcode
+                _terminate_analysis_process_tree(process)
+                with self._analysis_process_lock:
+                    if generation != self._analysis_generation:
+                        return
+                    self._last_error = (
+                        "runtime scheduled analysis worker exited without a result "
+                        f"(exit code {exit_code})"
+                    )
+                return
+
+            process.join(2)
+            if process.is_alive():
+                _terminate_analysis_process_tree(process)
+                with self._analysis_process_lock:
+                    if generation != self._analysis_generation:
+                        return
+                    self._last_error = "runtime scheduled analysis worker did not exit"
+                return
+
+            with self._analysis_process_lock:
+                if generation != self._analysis_generation:
+                    return
+                if result.get("success"):
+                    self._last_success_at = datetime.now().isoformat()
+                    self._last_error = None
+                else:
+                    self._last_error = result.get("error") or "runtime scheduled analysis failed"
+        except Exception as exc:  # noqa: BLE001 - watchdog failures must release the scheduler.
+            with self._analysis_process_lock:
+                if generation == self._analysis_generation:
+                    self._last_error = str(exc)
+            logger.exception("Runtime scheduler watchdog failed: %s", exc)
+        finally:
+            self._run_lock.release()
+            if "process" in locals():
+                with self._analysis_process_lock:
+                    if self._analysis_process is process:
+                        self._analysis_process = None
+            if result_queue is not None:
+                result_queue.cancel_join_thread()
+                result_queue.close()
+
+    def _start_analysis_watchdog(
+        self,
+        stock_codes: Optional[List[str]] = None,
+        *,
+        generation: Optional[int] = None,
+    ) -> bool:
+        with self._analysis_process_lock:
+            current_generation = self._analysis_generation
+            if generation is not None and generation != current_generation:
+                return False
+            generation = current_generation
+        if not self._run_lock.acquire(blocking=False):
+            self._record_analysis_busy_skip()
+            return False
+        worker = threading.Thread(
+            target=lambda: self._run_analysis_with_watchdog(
+                stock_codes,
+                lock_held=True,
+                generation=generation,
+            ),
+            daemon=True,
+            name="runtime-scheduler-watchdog",
+        )
+        try:
+            worker.start()
+        except Exception:
+            self._run_lock.release()
+            raise
+        return True
+
     def _current_times(self) -> List[str]:
         config = self._config_provider()
         return normalize_schedule_times(
@@ -274,6 +562,12 @@ class RuntimeSchedulerService:
                 return
             background_tasks = self._current_background_tasks(config)
             self.stop()
+            with self._analysis_process_lock:
+                generation = self._analysis_generation
+            scheduled_analysis = partial(
+                self._start_analysis_watchdog,
+                generation=generation,
+            )
             times = normalize_schedule_times(
                 getattr(config, "schedule_times", None),
                 fallback_time=getattr(config, "schedule_time", "18:00"),
@@ -285,9 +579,12 @@ class RuntimeSchedulerService:
                 register_signals=False,
             )
             if run_immediately and self._run_immediately_in_background:
-                scheduler.set_daily_task(self._run_analysis_once, run_immediately=False)
+                scheduler.set_daily_task(scheduled_analysis, run_immediately=False)
             else:
-                scheduler.set_daily_task(self._run_analysis_once, run_immediately=run_immediately)
+                scheduler.set_daily_task(
+                    scheduled_analysis,
+                    run_immediately=run_immediately,
+                )
             for entry in background_tasks:
                 scheduler.add_background_task(
                     entry["task"],
@@ -296,7 +593,7 @@ class RuntimeSchedulerService:
                     name=entry.get("name"),
                 )
             if run_immediately and self._run_immediately_in_background:
-                self._run_in_background_thread(self._run_analysis_once)
+                self._run_in_background_thread(scheduled_analysis)
             thread = threading.Thread(
                 target=scheduler.run,
                 daemon=True,
@@ -308,12 +605,18 @@ class RuntimeSchedulerService:
             thread.start()
 
     def stop(self) -> None:
-        scheduler = self._scheduler
-        if scheduler is not None:
-            scheduler.stop()
-        self._scheduler = None
-        self._thread = None
-        self._enabled = False
+        with self._lock:
+            with self._analysis_process_lock:
+                self._analysis_generation += 1
+                process = self._analysis_process
+            scheduler = self._scheduler
+            if scheduler is not None:
+                scheduler.stop()
+            if process is not None:
+                _terminate_analysis_process_tree(process)
+            self._scheduler = None
+            self._thread = None
+            self._enabled = False
 
     def reconcile_from_config(
         self,
@@ -333,30 +636,12 @@ class RuntimeSchedulerService:
             self.stop()
 
     def run_now(self) -> Dict[str, Any]:
-        if not self._run_lock.acquire(blocking=False):
-            self._record_analysis_busy_skip()
+        if not self._start_analysis_watchdog():
             return {
                 "accepted": False,
                 "running": True,
                 "reason": "analysis_already_running",
             }
-
-        def run_and_release() -> None:
-            try:
-                self._run_analysis_locked(None)
-            finally:
-                self._run_lock.release()
-
-        worker = threading.Thread(
-            target=run_and_release,
-            daemon=True,
-            name="runtime-scheduler-run-now",
-        )
-        try:
-            worker.start()
-        except Exception:
-            self._run_lock.release()
-            raise
         return {"accepted": True, "running": True}
 
     def status(self) -> Dict[str, Any]:

tests/test_runtime_scheduler_service.py

@@ -5,8 +5,12 @@ from __future__ import annotations
 
 import json
 import os
+import signal
+import subprocess
 import sys
 import tempfile
+import threading
+import time
 import unittest
 from datetime import datetime
 from pathlib import Path
@@ -18,11 +22,14 @@ from fastapi.testclient import TestClient
 
 from src.services.runtime_scheduler import (
     CLI_SCHEDULER_OWNER_ENV,
+    DEFAULT_RUNTIME_SCHEDULER_TIMEOUT_SECONDS,
     RUNTIME_SCHEDULER_ARGS_ENV,
     RUNTIME_SCHEDULER_FORCE_ENABLED_ENV,
     RUNTIME_SCHEDULER_RUN_IMMEDIATELY_ENV,
     RUNTIME_SCHEDULER_SUPPRESS_START_ENV,
+    RUNTIME_SCHEDULER_TIMEOUT_ENV,
     RuntimeSchedulerService,
+    _run_scheduled_analysis_process,
 )
 
 
@@ -86,7 +93,89 @@ class _SynchronousThread(_NoopThread):
             self.target()
 
 
+_BLOCKING_THREAD_RELEASE = threading.Event()
+
+
+def _blocking_thread_runner(config, args, stock_codes):
+    _BLOCKING_THREAD_RELEASE.wait(timeout=5)
+    return True
+
+
+def _blocking_spawn_runner(result_queue, stock_codes, schedule_args_overrides):
+    time.sleep(10)
+
+
+def _successful_spawn_runner(result_queue, stock_codes, schedule_args_overrides):
+    result_queue.put({"success": True, "error": None})
+
+
+def _large_failure_spawn_runner(result_queue, stock_codes, schedule_args_overrides):
+    result_queue.put({"success": False, "error": "x" * (1024 * 1024)})
+
+
+def _exit_with_live_descendant_spawn_runner(result_queue, stock_codes, schedule_args_overrides):
+    os.setsid()
+    child = subprocess.Popen(
+        [sys.executable, "-c", "import time; time.sleep(30)"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    pid_file = Path(os.environ["DSA_TEST_DESCENDANT_PID_FILE"])
+    pid_file.write_text(f"{os.getpid()} {child.pid}", encoding="utf-8")
+
+
+def _block_with_isolated_descendant_spawn_runner(result_queue, stock_codes, schedule_args_overrides):
+    os.setsid()
+    child = subprocess.Popen(
+        [sys.executable, "-c", "import time; time.sleep(30)"],
+        start_new_session=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    pid_file = Path(os.environ["DSA_TEST_DESCENDANT_PID_FILE"])
+    pid_file.write_text(f"{os.getpid()} {child.pid}", encoding="utf-8")
+    time.sleep(30)
+
+
 class RuntimeSchedulerServiceTestCase(unittest.TestCase):
+    @unittest.skipUnless(os.name == "posix", "POSIX session-isolation regression")
+    def test_worker_fails_before_analysis_when_posix_session_isolation_fails(self) -> None:
+        with patch(
+            "src.services.runtime_scheduler.os.setsid",
+            side_effect=OSError("not permitted"),
+        ), patch(
+            "src.services.runtime_scheduler.os.getsid",
+            return_value=101,
+        ), patch(
+            "src.services.runtime_scheduler.os.getpid",
+            return_value=202,
+        ):
+            with self.assertRaisesRegex(OSError, "not permitted"):
+                _run_scheduled_analysis_process(MagicMock(), None, {})
+
+    def test_analysis_timeout_reads_current_environment_with_safe_bounds(self) -> None:
+        service = RuntimeSchedulerService()
+
+        with patch.dict(os.environ, {}, clear=True):
+            self.assertEqual(
+                service._analysis_timeout_seconds(),
+                DEFAULT_RUNTIME_SCHEDULER_TIMEOUT_SECONDS,
+            )
+
+        for value, expected in (
+            ("0", 60),
+            ("59", 60),
+            ("60", 60),
+            ("600", 600),
+            ("invalid", DEFAULT_RUNTIME_SCHEDULER_TIMEOUT_SECONDS),
+        ):
+            with self.subTest(value=value), patch.dict(
+                os.environ,
+                {RUNTIME_SCHEDULER_TIMEOUT_ENV: value},
+                clear=True,
+            ):
+                self.assertEqual(service._analysis_timeout_seconds(), expected)
+
     def test_run_analysis_args_include_workers(self) -> None:
         config = SimpleNamespace(
             schedule_enabled=True,
@@ -194,32 +283,252 @@ class RuntimeSchedulerServiceTestCase(unittest.TestCase):
             schedule_time="18:00",
             schedule_times=["18:00"],
         )
-        seen_stock_codes = []
-
-        def runner(config_arg, args, stock_codes):
-            seen_stock_codes.append(stock_codes)
-            return True
-
-        service = RuntimeSchedulerService(
-            config_provider=lambda: config,
-            task_runner=runner,
-        )
-        service._reload_config = lambda: config
-
-        with patch(
-            "src.services.runtime_scheduler.threading.Thread",
-            _SynchronousThread,
-        ):
-            result = service.run_now()
+        service = RuntimeSchedulerService(config_provider=lambda: config)
+        service._analysis_process_target = _successful_spawn_runner
+        result = service.run_now()
 
         self.assertTrue(result["accepted"])
-        self.assertEqual(seen_stock_codes, [None])
+        deadline = time.monotonic() + 4
+        while service.status()["last_success_at"] is None and time.monotonic() < deadline:
+            time.sleep(0.05)
         status = service.status()
         self.assertFalse(status["running"])
         self.assertIsNotNone(status["last_run_at"])
         self.assertIsNotNone(status["last_success_at"])
         self.assertIsNone(status["last_error"])
 
+    def test_blocked_scheduled_analysis_times_out_and_allows_next_run(self) -> None:
+        fake_schedule = _FakeScheduleModule()
+        config = SimpleNamespace(
+            schedule_enabled=True,
+            schedule_time="18:00",
+            schedule_times=["18:00"],
+        )
+        service = RuntimeSchedulerService(
+            config_provider=lambda: config,
+            task_runner=_blocking_thread_runner,
+        )
+        service._reload_config = lambda: config
+        service._analysis_process_target = _blocking_spawn_runner
+        service._analysis_timeout_seconds = lambda: 1
+
+        with patch.dict(sys.modules, {"schedule": fake_schedule}), patch(
+            "src.services.runtime_scheduler.threading.Thread",
+            _NoopThread,
+        ):
+            service.reconcile_from_config()
+
+        callback_returned = threading.Event()
+        trigger = threading.Thread(
+            target=lambda: (fake_schedule.run_pending(), callback_returned.set()),
+            daemon=True,
+        )
+        _BLOCKING_THREAD_RELEASE.clear()
+        trigger.start()
+        try:
+            self.assertTrue(
+                callback_returned.wait(timeout=0.5),
+                "the scheduler callback remained blocked by analysis",
+            )
+
+            deadline = time.monotonic() + 4
+            while service.status()["last_error"] is None and time.monotonic() < deadline:
+                time.sleep(0.05)
+
+            status = service.status()
+            self.assertFalse(status["running"])
+            self.assertIn("timed out after 1s", status["last_error"])
+
+            service._analysis_process_target = _successful_spawn_runner
+            self.assertTrue(service.run_now()["accepted"])
+
+            deadline = time.monotonic() + 4
+            while service.status()["last_success_at"] is None and time.monotonic() < deadline:
+                time.sleep(0.05)
+
+            status = service.status()
+            self.assertFalse(status["running"])
+            self.assertIsNotNone(status["last_success_at"])
+            self.assertIsNone(status["last_error"])
+        finally:
+            _BLOCKING_THREAD_RELEASE.set()
+            trigger.join(timeout=5)
+
+    def test_stop_terminates_active_analysis_worker(self) -> None:
+        config = SimpleNamespace(
+            schedule_enabled=True,
+            schedule_time="18:00",
+            schedule_times=["18:00"],
+        )
+        service = RuntimeSchedulerService(config_provider=lambda: config)
+        service._analysis_process_target = _blocking_spawn_runner
+
+        self.assertTrue(service.run_now()["accepted"])
+        deadline = time.monotonic() + 5
+        while service.status()["last_run_at"] is None and time.monotonic() < deadline:
+            time.sleep(0.05)
+        self.assertIsNotNone(service.status()["last_run_at"])
+
+        service.stop()
+
+        deadline = time.monotonic() + 5
+        while service.status()["running"] and time.monotonic() < deadline:
+            time.sleep(0.05)
+        self.assertFalse(service.status()["running"])
+
+    def test_stop_does_not_record_expected_worker_termination_as_error(self) -> None:
+        config = SimpleNamespace(
+            schedule_enabled=True,
+            schedule_time="18:00",
+            schedule_times=["18:00"],
+        )
+        service = RuntimeSchedulerService(config_provider=lambda: config)
+        service._analysis_process_target = _blocking_spawn_runner
+
+        self.assertTrue(service.run_now()["accepted"])
+        deadline = time.monotonic() + 5
+        while service.status()["last_run_at"] is None and time.monotonic() < deadline:
+            time.sleep(0.05)
+        self.assertIsNotNone(service.status()["last_run_at"])
+
+        service.stop()
+
+        deadline = time.monotonic() + 5
+        while service.status()["running"] and time.monotonic() < deadline:
+            time.sleep(0.05)
+        status = service.status()
+        self.assertFalse(status["running"])
+        self.assertIsNone(status["last_error"])
+
+    def test_stale_scheduled_callback_cannot_start_after_stop(self) -> None:
+        fake_schedule = _FakeScheduleModule()
+        config = SimpleNamespace(
+            schedule_enabled=True,
+            schedule_time="18:00",
+            schedule_times=["18:00"],
+        )
+        service = RuntimeSchedulerService(config_provider=lambda: config)
+
+        with patch.dict(sys.modules, {"schedule": fake_schedule}), patch(
+            "src.services.runtime_scheduler.threading.Thread",
+            _NoopThread,
+        ):
+            service.start()
+
+        callback = fake_schedule.get_jobs()[0].job_func
+        service.stop()
+
+        with patch(
+            "src.services.runtime_scheduler.multiprocessing.get_context"
+        ) as get_context:
+            self.assertFalse(callback())
+
+        get_context.assert_not_called()
+        self.assertFalse(service.status()["running"])
+
+    def test_watchdog_reads_large_worker_result_before_joining(self) -> None:
+        config = SimpleNamespace(
+            schedule_enabled=True,
+            schedule_time="18:00",
+            schedule_times=["18:00"],
+        )
+        service = RuntimeSchedulerService(config_provider=lambda: config)
+        service._analysis_process_target = _large_failure_spawn_runner
+        service._analysis_timeout_seconds = lambda: 2
+
+        self.assertTrue(service.run_now()["accepted"])
+        deadline = time.monotonic() + 5
+        while service.status()["last_error"] is None and time.monotonic() < deadline:
+            time.sleep(0.05)
+
+        self.assertTrue(service.status()["last_error"].startswith("x"))
+
+    @unittest.skipUnless(os.name == "posix", "POSIX process-group regression")
+    def test_worker_exit_without_result_terminates_remaining_process_group(self) -> None:
+        config = SimpleNamespace(
+            schedule_enabled=True,
+            schedule_time="18:00",
+            schedule_times=["18:00"],
+        )
+        service = RuntimeSchedulerService(config_provider=lambda: config)
+        service._analysis_process_target = _exit_with_live_descendant_spawn_runner
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            pid_file = Path(temp_dir) / "descendant.pid"
+            process_group_id = None
+            child_pid = None
+            try:
+                with patch.dict(
+                    os.environ,
+                    {"DSA_TEST_DESCENDANT_PID_FILE": str(pid_file)},
+                ):
+                    self.assertTrue(service.run_now()["accepted"])
+                    deadline = time.monotonic() + 5
+                    while not pid_file.exists() and time.monotonic() < deadline:
+                        time.sleep(0.05)
+                    self.assertTrue(pid_file.exists())
+                    process_group_id, child_pid = [
+                        int(value) for value in pid_file.read_text(encoding="utf-8").split()
+                    ]
+
+                    while service.status()["running"] and time.monotonic() < deadline:
+                        time.sleep(0.05)
+
+                self.assertFalse(service.status()["running"])
+                self.assertIn("exited without a result", service.status()["last_error"])
+                with self.assertRaises(ProcessLookupError):
+                    os.kill(child_pid, 0)
+            finally:
+                if process_group_id is not None:
+                    try:
+                        os.killpg(process_group_id, signal.SIGKILL)
+                    except ProcessLookupError:
+                        pass
+
+    @unittest.skipUnless(os.name == "posix", "POSIX process-tree regression")
+    def test_stop_terminates_descendant_in_a_nested_process_group(self) -> None:
+        config = SimpleNamespace(
+            schedule_enabled=True,
+            schedule_time="18:00",
+            schedule_times=["18:00"],
+        )
+        service = RuntimeSchedulerService(config_provider=lambda: config)
+        service._analysis_process_target = _block_with_isolated_descendant_spawn_runner
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            pid_file = Path(temp_dir) / "descendant.pid"
+            process_group_id = None
+            child_pid = None
+            try:
+                with patch.dict(
+                    os.environ,
+                    {"DSA_TEST_DESCENDANT_PID_FILE": str(pid_file)},
+                ):
+                    self.assertTrue(service.run_now()["accepted"])
+                    deadline = time.monotonic() + 5
+                    while not pid_file.exists() and time.monotonic() < deadline:
+                        time.sleep(0.05)
+                    self.assertTrue(pid_file.exists())
+                    process_group_id, child_pid = [
+                        int(value) for value in pid_file.read_text(encoding="utf-8").split()
+                    ]
+
+                    service.stop()
+                    while service.status()["running"] and time.monotonic() < deadline:
+                        time.sleep(0.05)
+
+                self.assertFalse(service.status()["running"])
+                with self.assertRaises(ProcessLookupError):
+                    os.kill(child_pid, 0)
+            finally:
+                for group_id in (child_pid, process_group_id):
+                    if group_id is None:
+                        continue
+                    try:
+                        os.killpg(group_id, signal.SIGKILL)
+                    except ProcessLookupError:
+                        pass
+
     def test_run_now_uses_shared_lock_across_service_instances(self) -> None:
         config = SimpleNamespace(
             schedule_enabled=True,
@@ -279,7 +588,11 @@ class RuntimeSchedulerServiceTestCase(unittest.TestCase):
         )
         service._reload_config = lambda: config
 
-        with patch.dict(sys.modules, {"schedule": fake_schedule}), patch(
+        with patch.object(
+            service,
+            "_start_analysis_watchdog",
+            side_effect=lambda stock_codes=None, **kwargs: calls.append("run") or True,
+        ), patch.dict(sys.modules, {"schedule": fake_schedule}), patch(
             "src.services.runtime_scheduler.threading.Thread",
             _NoopThread,
         ):
@@ -315,7 +628,11 @@ class RuntimeSchedulerServiceTestCase(unittest.TestCase):
         )
         service._reload_config = lambda: config
 
-        with patch.dict(sys.modules, {"schedule": fake_schedule}), patch(
+        with patch.object(
+            service,
+            "_start_analysis_watchdog",
+            side_effect=lambda stock_codes=None, **kwargs: calls.append("run") or True,
+        ), patch.dict(sys.modules, {"schedule": fake_schedule}), patch(
             "src.services.runtime_scheduler.threading.Thread",
             _NoopThread,
         ):

tests/test_scheduler_background.py

@@ -3,6 +3,7 @@
 
 from datetime import datetime
 import sys
+import threading
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -167,6 +168,114 @@ class SchedulerBackgroundTaskTestCase(unittest.TestCase):
         self.assertEqual(scheduler._daily_jobs, [])
         self.assertIsNone(scheduler._daily_job)
 
+    def test_stop_during_pending_jobs_prevents_new_background_dispatch(self):
+        fake_schedule = _FakeScheduleModule()
+        pending_jobs_entered = threading.Event()
+        release_pending_jobs = threading.Event()
+        background_started = threading.Event()
+
+        def run_pending():
+            pending_jobs_entered.set()
+            release_pending_jobs.wait(timeout=1)
+
+        fake_schedule.run_pending = run_pending
+        with patch.dict(sys.modules, {"schedule": fake_schedule}):
+            from src.scheduler import Scheduler
+
+            scheduler = Scheduler(schedule_time="18:00")
+            with patch("src.scheduler.time.time", return_value=0):
+                scheduler.add_background_task(
+                    background_started.set,
+                    interval_seconds=30,
+                    run_immediately=False,
+                    name="test",
+                )
+
+            with patch("src.scheduler.time.time", return_value=30), patch(
+                "src.scheduler.time.sleep",
+                return_value=None,
+            ):
+                scheduler_thread = threading.Thread(target=scheduler.run, daemon=True)
+                scheduler_thread.start()
+                self.assertTrue(pending_jobs_entered.wait(timeout=1))
+
+                scheduler.stop()
+                release_pending_jobs.set()
+                scheduler_thread.join(timeout=1)
+
+        self.assertFalse(scheduler_thread.is_alive())
+        self.assertFalse(background_started.wait(timeout=0.1))
+
+    def test_stop_before_run_enters_prevents_scheduler_restart(self):
+        fake_schedule = _FakeScheduleModule()
+        background_started = threading.Event()
+        with patch.dict(sys.modules, {"schedule": fake_schedule}):
+            from src.scheduler import Scheduler
+
+            scheduler = Scheduler(schedule_time="18:00")
+            with patch("src.scheduler.time.time", return_value=0):
+                scheduler.add_background_task(
+                    background_started.set,
+                    interval_seconds=30,
+                    run_immediately=False,
+                    name="test",
+                )
+
+            scheduler.stop()
+
+            def stop_after_pending_jobs():
+                scheduler.shutdown_handler.shutdown_requested = True
+
+            fake_schedule.run_pending = stop_after_pending_jobs
+            with patch("src.scheduler.time.time", return_value=30), patch(
+                "src.scheduler.time.sleep",
+                return_value=None,
+            ):
+                scheduler_thread = threading.Thread(target=scheduler.run, daemon=True)
+                scheduler_thread.start()
+                scheduler_thread.join(timeout=1)
+
+        self.assertFalse(scheduler_thread.is_alive())
+        self.assertFalse(background_started.wait(timeout=0.1))
+
+    def test_shutdown_during_pending_jobs_prevents_new_background_dispatch(self):
+        fake_schedule = _FakeScheduleModule()
+        pending_jobs_entered = threading.Event()
+        release_pending_jobs = threading.Event()
+        background_started = threading.Event()
+
+        def run_pending():
+            pending_jobs_entered.set()
+            release_pending_jobs.wait(timeout=1)
+
+        fake_schedule.run_pending = run_pending
+        with patch.dict(sys.modules, {"schedule": fake_schedule}):
+            from src.scheduler import Scheduler
+
+            scheduler = Scheduler(schedule_time="18:00", register_signals=False)
+            with patch("src.scheduler.time.time", return_value=0):
+                scheduler.add_background_task(
+                    background_started.set,
+                    interval_seconds=30,
+                    run_immediately=False,
+                    name="test",
+                )
+
+            with patch("src.scheduler.time.time", return_value=30), patch(
+                "src.scheduler.time.sleep",
+                return_value=None,
+            ):
+                scheduler_thread = threading.Thread(target=scheduler.run, daemon=True)
+                scheduler_thread.start()
+                self.assertTrue(pending_jobs_entered.wait(timeout=1))
+
+                scheduler.shutdown_handler.shutdown_requested = True
+                release_pending_jobs.set()
+                scheduler_thread.join(timeout=1)
+
+        self.assertFalse(scheduler_thread.is_alive())
+        self.assertFalse(background_started.wait(timeout=0.1))
+
     def test_scheduler_keeps_existing_daily_job_when_schedule_time_invalid(self):
         fake_schedule = _FakeScheduleModule()
         with patch.dict(sys.modules, {"schedule": fake_schedule}):