init

supply/__init__.py

@@ -6,9 +6,9 @@
   ├─────────────────┤
   │  supply.infra   │  基础设施（DB / HTTP / 日志 / 工具）
   ├─────────────────┤
-  │  supply.domain  │  领域抽象（Agent / Tool / Memory / Task）
+  │  supply.core    │  框架内核（Agent / Tool / Memory / Orchestrator）
   ├─────────────────┤
-  │  supply.server  │  应用服务（Quart / XXL / TaskScheduler）
+  │  supply.server  │  应用服务（Quart / XXL）
   └─────────────────┘
 
 快速开始:
@@ -21,6 +21,7 @@
 
 # 配置
 from supply.config import (
+    GlobalConfig,
     DatabaseConfig,
     AliyunLogConfig,
     AliyunOssConfig,
@@ -40,11 +41,10 @@ from supply.infra import (
     LogCategory,
     LogStatus,
     AsyncHttpClient,
-    TaskScheduleResponse,
 )
 
 # 领域
-from supply.domain import (
+from supply.core import (
     Agent,
     AgentContext,
     AgentResult,
@@ -58,33 +58,25 @@ from supply.domain import (
     Memory,
     ConversationMemory,
     VectorMemory,
-    Task,
-    TaskResult,
-    TaskHandler,
-    TaskRegistry,
-    TaskConfig,
-    TaskStatus,
 )
 
 # 服务
-from supply.server import create_app, TaskScheduler
+from supply.server import create_app
 
 __all__ = [
     # Config
+    "GlobalConfig",
     "DatabaseConfig", "AliyunLogConfig", "AliyunOssConfig",
     "PgConfig", "MilvusConfig", "RedisConfig",
     # Infra
     "DatabaseManager", "create_backend", "MySQLBackend",
     "LogService", "LogRecord", "LogLevel", "LogCategory", "LogStatus",
-    "AsyncHttpClient", "TaskScheduleResponse",
-    # Domain - Agent
+    "AsyncHttpClient",
+    # Core - Agent
     "Agent", "AgentContext", "AgentResult", "AgentStatus",
     "AgentOrchestrator", "ChainOrchestrator", "ParallelOrchestrator",
     "Tool", "ToolResult", "ToolRegistry",
     "Memory", "ConversationMemory", "VectorMemory",
-    # Domain - Task
-    "Task", "TaskResult", "TaskHandler", "TaskRegistry",
-    "TaskConfig", "TaskStatus",
     # Server
-    "create_app", "TaskScheduler",
+    "create_app",
 ]

supply/config/__init__.py

@@ -13,6 +13,8 @@ from supply.config.database import (
     RedisConfig,
 )
 from supply.config.aliyun import AliyunLogConfig, AliyunOssConfig
+from supply.config.xxljob import XxlJobConfig
+from supply.config.global_config import GlobalConfig
 
 __all__ = [
     "DatabaseConfig",
@@ -27,4 +29,6 @@ __all__ = [
     "RedisConfig",
     "AliyunLogConfig",
     "AliyunOssConfig",
+    "XxlJobConfig",
+    "GlobalConfig",
 ]

supply/config/global_config.py

@@ -0,0 +1,79 @@
+"""全局配置聚合根 —— 唯一的配置入口，启动时一次创建、四处注入"""
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from supply.config.database import (
+    AigcDatabaseConfig,
+    LongVideoDatabaseConfig,
+    LongArticlesDatabaseConfig,
+    PiaoquanCrawlerDatabaseConfig,
+    GrowthDatabaseConfig,
+    ContentDeconstructionSupplyConfig,
+)
+from supply.config.aliyun import AliyunLogConfig, AliyunOssConfig
+from supply.config.xxljob import XxlJobConfig
+
+
+class GlobalConfig(BaseSettings):
+    """应用全局配置 —— 所有子配置的聚合根
+
+    启动时一次性创建，注入到各组件：
+        global_config = GlobalConfig()
+        db = DatabaseManager(create_backend("aigc", global_config.aigc_db))
+        log_service = LogService(global_config.aliyun_log)
+        executor = XxlJobExecutor(global_config.xxl_job)
+
+    每个子配置保持独立的 env_prefix，环境变量按各自前缀覆盖。
+    """
+
+    # ============ 应用基础配置 ============
+    app_name: str = Field(
+        default="long-articles-agentic-supply",
+        description="应用名称，用于 XXL-JOB 执行器注册和日志标识",
+    )
+    environment: str = Field(
+        default="development",
+        description="运行环境: development / pre / production",
+    )
+    debug: bool = Field(default=False, description="调试模式")
+
+    # ============ 数据库配置 ============
+    aigc_db: AigcDatabaseConfig = Field(default_factory=AigcDatabaseConfig)
+    long_video_db: LongVideoDatabaseConfig = Field(
+        default_factory=LongVideoDatabaseConfig
+    )
+    long_articles_db: LongArticlesDatabaseConfig = Field(
+        default_factory=LongArticlesDatabaseConfig
+    )
+    piaoquan_crawler_db: PiaoquanCrawlerDatabaseConfig = Field(
+        default_factory=PiaoquanCrawlerDatabaseConfig
+    )
+    growth_db: GrowthDatabaseConfig = Field(default_factory=GrowthDatabaseConfig)
+    content_decode_db: ContentDeconstructionSupplyConfig = Field(
+        default_factory=ContentDeconstructionSupplyConfig
+    )
+
+    # ============ 阿里云服务配置 ============
+    aliyun_log: AliyunLogConfig = Field(default_factory=AliyunLogConfig)
+    aliyun_oss: AliyunOssConfig = Field(default_factory=AliyunOssConfig)
+
+    # ============ XXL-JOB 配置 ============
+    xxl_job: XxlJobConfig = Field(default_factory=XxlJobConfig)
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    # ============ 组合导出 ============
+
+    def active_db_configs(self) -> dict[str, object]:
+        """返回业务数据库配置映射 {name: config}，供 DatabaseManager 使用"""
+        return {
+            "aigc": self.aigc_db,
+            "long_articles": self.long_articles_db,
+            "piaoquan_crawler": self.piaoquan_crawler_db,
+        }

supply/config/xxljob.py

@@ -0,0 +1,109 @@
+"""XXL-JOB 执行器配置 —— 对标 Java XxlJobConfig.java"""
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class XxlJobConfig(BaseSettings):
+    """XXL-JOB 执行器连接配置
+
+    对应 Java 侧 @Value("${xxl.job.*}") 注入的全部字段。
+    通过环境变量 XXL_<FIELD> 或 .env 文件覆盖默认值。
+
+    环境变量映射（prefix=XXL_）:
+        XXL_ADMIN_ADDRESSES     → admin_addresses
+        XXL_ACCESS_TOKEN        → access_token
+        XXL_APP_NAME            → app_name
+        XXL_ADDRESS             → address
+        XXL_IP                  → ip
+        XXL_PORT                → port
+        XXL_LOG_PATH            → log_path
+        XXL_LOG_RETENTION_DAYS  → log_retention_days
+        XXL_REGISTRY_INTERVAL   → registry_interval
+    """
+
+    # Admin 调度中心地址（多个逗号分隔）
+    admin_addresses: str = Field(
+        default="http://xxl-job-internal.piaoquantv.com/xxl-job-admin",
+        description="XXL-JOB Admin 调度中心地址，多个逗号分隔",
+        min_length=1,
+    )
+
+    # 访问令牌
+    access_token: str = Field(
+        default="",
+        description="与 Admin 端 xxl.job.accessToken 一致的访问令牌",
+    )
+
+    # 执行器标识
+    app_name: str = Field(
+        default="long-articles-agentic-supply",
+        description="执行器 AppName，Admin 通过此名称识别执行器",
+        min_length=1,
+    )
+
+    # 执行器地址（留空自动探测）
+    address: str = Field(
+        default="",
+        description="执行器注册地址，留空则自动拼装 http://{ip}:{port}",
+    )
+
+    # 执行器 IP
+    ip: str = Field(
+        default="",
+        description="执行器 IP，留空则自动探测本机 IP",
+    )
+
+    # 执行器端口
+    port: int = Field(
+        default=7800,
+        description="执行器 HTTP 端口，Admin 通过此端口回调 /run /beat 等",
+        gt=0,
+        lt=65536,
+    )
+
+    # 日志路径
+    log_path: str = Field(
+        default="/datalog/weblog/xxl-job/jobhandler",
+        description="执行器运行日志本地存储路径",
+        min_length=1,
+    )
+
+    # 日志保留天数
+    log_retention_days: int = Field(
+        default=7,
+        description="本地日志文件保留天数",
+        ge=1,
+    )
+
+    # 心跳间隔
+    registry_interval: int = Field(
+        default=30,
+        description="向 Admin 注册/续约心跳间隔（秒）",
+        ge=5,
+    )
+
+    model_config = SettingsConfigDict(
+        env_prefix="XXL_",
+        env_file=".env",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    def to_dict(self) -> dict:
+        """导出为字典，过滤空值字段"""
+        result = {
+            "admin_addresses": self.admin_addresses,
+            "app_name": self.app_name,
+            "port": self.port,
+            "log_path": self.log_path,
+            "log_retention_days": self.log_retention_days,
+            "registry_interval": self.registry_interval,
+        }
+        if self.access_token:
+            result["access_token"] = self.access_token
+        if self.address:
+            result["address"] = self.address
+        if self.ip:
+            result["ip"] = self.ip
+        return result

supply/core/__init__.py

@@ -1,7 +1,7 @@
-"""核心层 —— 生命周期管理
+"""核心层 —— 框架内核 + 生命周期管理
 
-配置和数据库已下沉到 config/ 和 infra/ 层。
-此处保留常见类型的 re-export 方便使用。
+Agent/Tool/Memory/Orchestrator 契约定义在此层。
+业务 Agent 实现放在独立的业务模块中，通过继承这些基类接入。
 """
 
 # 配置
@@ -29,6 +29,24 @@ from supply.infra.observability import LogService, LogRecord, LogLevel, LogCateg
 # 生命周期
 from supply.core.bootstrap import AppContext
 
+# Agent 框架内核
+from supply.core.agent import (
+    Agent,
+    AgentContext,
+    AgentResult,
+    AgentStatus,
+    AgentOrchestrator,
+    ChainOrchestrator,
+    ParallelOrchestrator,
+    Tool,
+    ToolResult,
+    ToolRegistry,
+    Memory,
+    Message,
+    ConversationMemory,
+    VectorMemory,
+)
+
 __all__ = [
     "DatabaseConfig",
     "AliyunLogConfig",
@@ -50,4 +68,19 @@ __all__ = [
     "LogCategory",
     "LogStatus",
     "AppContext",
+    # Agent 框架
+    "Agent",
+    "AgentContext",
+    "AgentResult",
+    "AgentStatus",
+    "AgentOrchestrator",
+    "ChainOrchestrator",
+    "ParallelOrchestrator",
+    "Tool",
+    "ToolResult",
+    "ToolRegistry",
+    "Memory",
+    "Message",
+    "ConversationMemory",
+    "VectorMemory",
 ]

supply/domain/agent/__init__.py → supply/core/agent/__init__.py

@@ -1,9 +1,9 @@
 """Agent 领域层 —— Agent/Tool/Memory/Orchestrator 核心抽象"""
 
-from supply.domain.agent.base import Agent, AgentContext, AgentResult, AgentStatus
-from supply.domain.agent.tool import Tool, ToolResult, ToolRegistry
-from supply.domain.agent.memory import Memory, Message, ConversationMemory, VectorMemory
-from supply.domain.agent.orchestrator import (
+from supply.core.agent.base import Agent, AgentContext, AgentResult, AgentStatus
+from supply.core.agent.tool import Tool, ToolResult, ToolRegistry
+from supply.core.agent.memory import Memory, Message, ConversationMemory, VectorMemory
+from supply.core.agent.orchestrator import (
     AgentOrchestrator,
     ChainOrchestrator,
     ParallelOrchestrator,

supply/domain/agent/base.py → supply/core/agent/base.py

@@ -11,8 +11,8 @@ from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, Dict, List, Optional
 
-from supply.domain.agent.tool import Tool
-from supply.domain.agent.memory import Memory
+from supply.core.agent.tool import Tool
+from supply.core.agent.memory import Memory
 
 
 # ==================== Agent 状态 ====================

supply/domain/agent/orchestrator.py → supply/core/agent/orchestrator.py

@@ -12,7 +12,7 @@ import logging
 from abc import ABC, abstractmethod
 from typing import Any, Dict, List, Optional
 
-from supply.domain.agent.base import Agent, AgentContext, AgentResult, AgentStatus
+from supply.core.agent.base import Agent, AgentContext, AgentResult, AgentStatus
 
 logger = logging.getLogger(__name__)
 

supply/domain/__init__.py

@@ -1,69 +0,0 @@
-"""领域层 —— Agent、Task、Workflow 等核心业务抽象
-
-Domain 层独立于 Infra 层，定义接口（Port），不引入具体实现。
-Infra 层提供 Adapter 实现这些接口，通过 DI 容器注入。
-"""
-
-from supply.domain.agent import (
-    Agent,
-    AgentContext,
-    AgentResult,
-    AgentStatus,
-    Tool,
-    ToolResult,
-    ToolRegistry,
-    Memory,
-    Message,
-    ConversationMemory,
-    VectorMemory,
-    AgentOrchestrator,
-    ChainOrchestrator,
-    ParallelOrchestrator,
-)
-from supply.domain.task import (
-    Task,
-    TaskResult,
-    TaskHandler,
-    TaskRegistry,
-    TaskConfig,
-    TaskStatus,
-    TaskConstants,
-    get_default_task_config,
-    TaskError,
-    TaskValidationError,
-    TaskTimeoutError,
-    TaskConcurrencyError,
-    TaskLockError,
-)
-
-__all__ = [
-    # Agent
-    "Agent",
-    "AgentContext",
-    "AgentResult",
-    "AgentStatus",
-    "Tool",
-    "ToolResult",
-    "ToolRegistry",
-    "Memory",
-    "Message",
-    "ConversationMemory",
-    "VectorMemory",
-    "AgentOrchestrator",
-    "ChainOrchestrator",
-    "ParallelOrchestrator",
-    # Task
-    "Task",
-    "TaskResult",
-    "TaskHandler",
-    "TaskRegistry",
-    "TaskConfig",
-    "TaskStatus",
-    "TaskConstants",
-    "get_default_task_config",
-    "TaskError",
-    "TaskValidationError",
-    "TaskTimeoutError",
-    "TaskConcurrencyError",
-    "TaskLockError",
-]

supply/domain/task/__init__.py

@@ -1,27 +0,0 @@
-"""Task 领域层 —— 任务抽象、处理器基类、注册表、异常"""
-
-from supply.domain.task.base import Task, TaskResult, TaskHandler, TaskRegistry
-from supply.domain.task.config import TaskConfig, TaskStatus, TaskConstants, get_default_task_config
-from supply.domain.task.exceptions import (
-    TaskError,
-    TaskValidationError,
-    TaskTimeoutError,
-    TaskConcurrencyError,
-    TaskLockError,
-)
-
-__all__ = [
-    "Task",
-    "TaskResult",
-    "TaskHandler",
-    "TaskRegistry",
-    "TaskConfig",
-    "TaskStatus",
-    "TaskConstants",
-    "get_default_task_config",
-    "TaskError",
-    "TaskValidationError",
-    "TaskTimeoutError",
-    "TaskConcurrencyError",
-    "TaskLockError",
-]

supply/domain/task/base.py

@@ -1,156 +0,0 @@
-"""Task 领域 —— 任务抽象、处理器基类、注册表
-
-所有任务概念集中于此，替代旧的分散在 jobs/task_*.py 的全局注册表模式。
-"""
-
-from __future__ import annotations
-
-from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-from typing import Any, Callable, Dict, List, Optional
-
-from supply.domain.task.config import TaskConfig, TaskStatus, get_default_task_config
-from supply.domain.task.exceptions import (
-    TaskError,
-    TaskValidationError,
-    TaskTimeoutError,
-    TaskConcurrencyError,
-    TaskLockError,
-)
-
-
-# ==================== Task / TaskResult ====================
-
-
-@dataclass
-class Task:
-    """任务数据模型"""
-
-    task_name: str
-    params: Dict[str, Any] = field(default_factory=dict)
-    trace_id: str = ""
-    config: Optional[TaskConfig] = None
-
-
-@dataclass
-class TaskResult:
-    """任务执行结果"""
-
-    success: bool
-    data: Any = None
-    error: Optional[str] = None
-    trace_id: str = ""
-
-    @classmethod
-    def ok(cls, data: Any = None, trace_id: str = "") -> "TaskResult":
-        return cls(success=True, data=data, trace_id=trace_id)
-
-    @classmethod
-    def fail(cls, error: str, trace_id: str = "") -> "TaskResult":
-        return cls(success=False, error=error, trace_id=trace_id)
-
-
-# ==================== TaskRegistry（实例级） ====================
-
-
-class TaskRegistry:
-    """任务注册表 —— 实例级，替代全局 _TASK_HANDLER_REGISTRY
-
-    使用方式：
-        registry = TaskRegistry()
-
-        @registry.register("content_generation")
-        class ContentGenHandler(TaskHandler):
-            ...
-
-        handler = registry.get("content_generation")
-        result = await handler.execute(task)
-    """
-
-    def __init__(self):
-        self._handlers: Dict[str, type] = {}
-
-    def register(self, task_name: str) -> Callable:
-        """装饰器：将 TaskHandler 子类注册到注册表
-
-        使用方式：
-            @registry.register("my_task")
-            class MyHandler(TaskHandler):
-                ...
-        """
-
-        def decorator(handler_cls: type) -> type:
-            if not issubclass(handler_cls, TaskHandler):
-                raise TypeError(
-                    f"@register 只能用于 TaskHandler 子类，收到 {handler_cls}"
-                )
-            if task_name in self._handlers:
-                raise ValueError(f"任务 '{task_name}' 已注册")
-            self._handlers[task_name] = handler_cls
-            handler_cls.task_name = task_name
-            return handler_cls
-
-        return decorator
-
-    def get(self, task_name: str) -> Optional[type]:
-        return self._handlers.get(task_name)
-
-    def list_names(self) -> List[str]:
-        return list(self._handlers.keys())
-
-    def __len__(self) -> int:
-        return len(self._handlers)
-
-    def __contains__(self, task_name: str) -> bool:
-        return task_name in self._handlers
-
-
-# ==================== TaskHandler 抽象基类 ====================
-
-
-class TaskHandler(ABC):
-    """任务处理器抽象基类
-
-    业务方继承此类实现具体任务逻辑：
-
-        class ContentGenHandler(TaskHandler):
-            task_name = "content_generation"
-
-            async def validate(self, task: Task) -> bool:
-                return "topic" in task.params
-
-            async def execute(self, task: Task) -> TaskResult:
-                ...
-
-    task_name 类属性由 @TaskRegistry.register 装饰器自动设置。
-    """
-
-    task_name: str = ""
-
-    def __init__(
-        self,
-        log_service: Any = None,
-        db_client: Any = None,
-    ):
-        self.log_service = log_service
-        self.db_client = db_client
-
-    @abstractmethod
-    async def execute(self, task: Task) -> TaskResult:
-        """执行任务主逻辑"""
-        ...
-
-    async def validate(self, task: Task) -> bool:
-        """校验任务参数，默认通过。子类可按需覆写"""
-        return True
-
-    async def on_success(self, task: Task) -> None:
-        """任务成功回调（可选覆写）"""
-        pass
-
-    async def on_failure(self, task: Task, error: Exception) -> None:
-        """任务失败回调（可选覆写）"""
-        pass
-
-    def __repr__(self) -> str:
-        return f"<TaskHandler name={self.task_name}>"

supply/domain/task/config.py

@@ -1,40 +0,0 @@
-"""Task 配置与状态常量 —— 从 supply.jobs.task_config 迁移"""
-
-from dataclasses import dataclass
-
-
-@dataclass
-class TaskConfig:
-    """单个任务的运行时配置"""
-
-    timeout: int  # 超时时间（秒）
-    max_concurrent: int = 5  # 最大并发数
-    retry_times: int = 0  # 重试次数
-    retryable: bool = True  # 是否可重试
-    alert_on_failure: bool = True  # 失败时是否告警
-
-
-class TaskStatus:
-    """任务状态常量"""
-
-    INIT = 0
-    PROCESSING = 1
-    SUCCESS = 2
-    FAILED = 99
-
-
-class TaskConstants:
-    """任务系统常量"""
-
-    DEFAULT_TIMEOUT = 1800
-    DEFAULT_MAX_CONCURRENT = 5
-    DEFAULT_RETRY_TIMES = 0
-
-
-def get_default_task_config() -> TaskConfig:
-    """获取默认任务配置"""
-    return TaskConfig(
-        timeout=TaskConstants.DEFAULT_TIMEOUT,
-        max_concurrent=TaskConstants.DEFAULT_MAX_CONCURRENT,
-        retry_times=TaskConstants.DEFAULT_RETRY_TIMES,
-    )

supply/domain/task/exceptions.py

@@ -1,44 +0,0 @@
-"""Task 异常体系 —— 从 supply.jobs.task_utils 迁移"""
-
-
-class TaskError(Exception):
-    """任务异常基类"""
-
-    def __init__(self, message: str, retryable: bool = False):
-        super().__init__(message)
-        self.retryable = retryable
-
-
-class TaskValidationError(TaskError):
-    """任务参数校验失败"""
-
-    def __init__(self, message: str):
-        super().__init__(message, retryable=False)
-
-
-class TaskTimeoutError(TaskError):
-    """任务执行超时"""
-
-    def __init__(self, task_name: str, timeout: int):
-        super().__init__(
-            f"任务 '{task_name}' 执行超时（{timeout}s）", retryable=True
-        )
-
-
-class TaskConcurrencyError(TaskError):
-    """任务并发限制"""
-
-    def __init__(self, task_name: str, max_concurrent: int):
-        super().__init__(
-            f"任务 '{task_name}' 并发数已达上限（{max_concurrent}）",
-            retryable=True,
-        )
-
-
-class TaskLockError(TaskError):
-    """任务锁获取失败（被其他实例持有）"""
-
-    def __init__(self, task_name: str):
-        super().__init__(
-            f"任务 '{task_name}' 锁已被其他实例持有", retryable=True
-        )

supply/handlers/__init__.py

@@ -0,0 +1,13 @@
+"""业务 XXL-JOB Handler 包
+
+Handler 通过 discover_and_register() 在应用启动时自动发现。
+每个 handler 模块用 @xxl_job 装饰器自注册，无需手动 import。
+
+用法:
+    from supply.handlers import discover_and_register
+    discover_and_register()
+"""
+
+from supply.handlers._discovery import discover_and_register
+
+__all__ = ["discover_and_register"]

supply/handlers/_discovery.py

@@ -0,0 +1,52 @@
+"""XXL-JOB handler 自动发现 —— pkgutil + importlib 扫描导入
+
+扫描 supply.handlers 包下的所有非私有模块，逐一 import，
+触发 @xxl_job 装饰器将 handler 注册到 _XXL_REGISTRY。
+"""
+
+import importlib
+import logging
+import pkgutil
+
+logger = logging.getLogger(__name__)
+
+
+def discover_and_register(package: str = "supply.handlers") -> None:
+    """自动发现并导入 package 下的所有 handler 模块
+
+    跳过 _ 前缀的私有模块和子包。每个模块被 import 时，
+    @xxl_job 装饰器自动执行，将 handler 写入共享的 _XXL_REGISTRY。
+
+    幂等调用：importlib 使用 sys.modules 缓存，重复调用是 no-op。
+
+    Args:
+        package: 点分隔的 Python 包路径，默认 "supply.handlers"
+    """
+    try:
+        pkg = importlib.import_module(package)
+    except ImportError:
+        logger.warning(
+            "XXL-JOB handler 包 %r 不存在，无 handler 注册", package
+        )
+        return
+
+    registered = 0
+    for finder, name, is_pkg in pkgutil.iter_modules(
+        pkg.__path__, pkg.__name__ + "."
+    ):
+        if name.split(".")[-1].startswith("_"):
+            continue
+
+        try:
+            importlib.import_module(name)
+            registered += 1
+            logger.debug("XXL-JOB 发现 handler 模块: %s", name)
+        except Exception:
+            logger.exception("XXL-JOB handler 模块 %r 导入失败，跳过", name)
+
+    if registered:
+        logger.info(
+            "XXL-JOB 自动发现完成: 已导入 %d 个 handler 模块", registered
+        )
+    else:
+        logger.info("XXL-JOB 自动发现: 在 %r 中未找到 handler 模块", package)

supply/jobs/test_handler.py → supply/handlers/test_hello_world.py

@@ -2,7 +2,7 @@
 
 import logging
 
-from supply.jobs.xxl_decorator import xxl_job
+from supply.infra.xxl_jobs.xxl_decorator import xxl_job
 
 logger = logging.getLogger(__name__)
 

supply/infra/__init__.py

@@ -23,7 +23,6 @@ from supply.infra.observability import (
 )
 from supply.infra.shared import (
     AsyncHttpClient,
-    TaskScheduleResponse,
     run_tasks_with_async_worker_group,
     run_tasks_with_asyncio_task_group,
     safe_json_parse,
@@ -33,6 +32,12 @@ from supply.infra.shared import (
     timestamp_to_str,
     days_remaining_in_month,
 )
+from supply.infra.xxl_jobs import (
+    XxlJobConfig,
+    xxl_job,
+    xxl_registry,
+    XxlJobExecutor,
+)
 
 __all__ = [
     # 数据库
@@ -56,7 +61,6 @@ __all__ = [
     "SLS_INDEX_CONFIG",
     # 共享工具
     "AsyncHttpClient",
-    "TaskScheduleResponse",
     "run_tasks_with_async_worker_group",
     "run_tasks_with_asyncio_task_group",
     "safe_json_parse",
@@ -65,4 +69,9 @@ __all__ = [
     "yield_batch",
     "timestamp_to_str",
     "days_remaining_in_month",
+    # 任务调度
+    "XxlJobConfig",
+    "xxl_job",
+    "xxl_registry",
+    "XxlJobExecutor",
 ]

supply/infra/shared/__init__.py

@@ -1,5 +1,4 @@
 from .http_client import AsyncHttpClient
-from .response import TaskScheduleResponse
 from .async_tasks import (
     run_tasks_with_async_worker_group,
     run_tasks_with_asyncio_task_group,

supply/infra/shared/response.py

@@ -1,18 +0,0 @@
-from typing import Any, Dict
-
-
-class TaskScheduleResponse:
-    """任务调度专用响应格式"""
-
-    @classmethod
-    def fail(cls, error_code: str, error_message: str) -> Dict[str, Any]:
-        return {"code": error_code, "status": "error", "message": error_message}
-
-    @classmethod
-    def success(cls, task_name: str, data: Any) -> Dict[str, Any]:
-        return {
-            "code": 0,
-            "status": "task execute successfully",
-            "data": data,
-            "task_name": task_name,
-        }

supply/infra/xxl_jobs/__init__.py

@@ -0,0 +1,10 @@
+from supply.config.xxljob import XxlJobConfig
+from supply.infra.xxl_jobs.xxl_decorator import xxl_job, xxl_registry
+from supply.infra.xxl_jobs.xxl_executor import XxlJobExecutor
+
+__all__ = [
+    "XxlJobConfig",
+    "xxl_job",
+    "xxl_registry",
+    "XxlJobExecutor",
+]

supply/jobs/xxl_decorator.py → supply/infra/xxl_jobs/xxl_decorator.py

@@ -1,7 +1,7 @@
 """XXL-JOB 任务注册装饰器 —— 对标 Java @XxlJob 注解
 
 使用方式：
-    from supply.jobs.xxl_decorator import xxl_job, xxl_registry
+    from supply.infra.xxl_jobs.xxl_decorator import xxl_job, xxl_registry
 
     @xxl_job("myJobHandler")
     async def my_job(param: str) -> dict:

supply/jobs/xxl_executor.py → supply/infra/xxl_jobs/xxl_executor.py

@@ -14,8 +14,8 @@ from typing import Any, Dict, Optional
 
 import aiohttp
 
-from supply.jobs.xxl_config import XxlJobConfig
-from supply.jobs.xxl_decorator import xxl_registry
+from supply.config.xxljob import XxlJobConfig
+from supply.infra.xxl_jobs.xxl_decorator import xxl_registry
 
 logger = logging.getLogger(__name__)
 

supply/jobs/__init__.py

@@ -1,32 +0,0 @@
-from supply.jobs.task_config import TaskConfig, TaskStatus, TaskConstants, get_default_task_config
-from supply.jobs.task_utils import (
-    TaskError,
-    TaskValidationError,
-    TaskTimeoutError,
-    TaskConcurrencyError,
-    TaskLockError,
-    TaskUtils,
-)
-from supply.jobs.task_handler import TaskHandler, TaskRegistry
-from supply.jobs.xxl_config import XxlJobConfig
-from supply.jobs.xxl_decorator import xxl_job, xxl_registry
-from supply.jobs.xxl_executor import XxlJobExecutor
-
-__all__ = [
-    "TaskConfig",
-    "TaskStatus",
-    "TaskConstants",
-    "get_default_task_config",
-    "TaskError",
-    "TaskValidationError",
-    "TaskTimeoutError",
-    "TaskConcurrencyError",
-    "TaskLockError",
-    "TaskUtils",
-    "TaskHandler",
-    "TaskRegistry",
-    "XxlJobConfig",
-    "xxl_job",
-    "xxl_registry",
-    "XxlJobExecutor",
-]

supply/jobs/task_config.py

@@ -1,12 +0,0 @@
-"""向后兼容 shim —— 委托到 supply.domain.task.config
-
-新代码请直接使用:
-    from supply.domain.task import TaskConfig, TaskStatus, TaskConstants, get_default_task_config
-"""
-
-from supply.domain.task.config import (  # noqa: F401
-    TaskConfig,
-    TaskStatus,
-    TaskConstants,
-    get_default_task_config,
-)

supply/jobs/task_handler.py

@@ -1,15 +0,0 @@
-"""向后兼容 shim —— 委托到 supply.domain.task.base
-
-新代码请直接使用:
-    from supply.domain.task import TaskHandler, TaskRegistry
-
-注意：原有的模块级 _TASK_HANDLER_REGISTRY 已废弃。
-新代码应使用实例化的 TaskRegistry。
-"""
-
-from supply.domain.task.base import (  # noqa: F401
-    Task,
-    TaskResult,
-    TaskHandler,
-    TaskRegistry,
-)

supply/jobs/task_utils.py

@@ -1,41 +0,0 @@
-"""向后兼容 shim —— 委托到 supply.domain.task.exceptions
-
-新代码请直接使用:
-    from supply.domain.task import TaskError, TaskValidationError, ...
-    TaskUtils 工具方法后续迁移到 domain/task/utils.py
-"""
-
-import traceback
-from typing import Any
-
-from supply.domain.task.exceptions import (  # noqa: F401
-    TaskError,
-    TaskValidationError,
-    TaskTimeoutError,
-    TaskConcurrencyError,
-    TaskLockError,
-)
-
-
-class TaskUtils:
-    """任务工具方法（向后兼容）"""
-
-    @staticmethod
-    def validate_table_name(table_name: str) -> str:
-        cleaned = table_name.strip().replace(" ", "_")
-        if not cleaned or "--" in cleaned or ";" in cleaned.lower():
-            raise TaskValidationError(f"非法的表名: {table_name}")
-        return cleaned
-
-    @staticmethod
-    def validate_task_name(task_name: Any) -> str:
-        task_name = str(task_name).strip()
-        if not task_name:
-            raise TaskValidationError("task_name 不能为空")
-        return task_name
-
-    @staticmethod
-    def format_error_detail(e: Exception) -> str:
-        return "".join(
-            traceback.format_exception(type(e), e, e.__traceback__)
-        )[-1000:]

supply/jobs/xxl_config.py

@@ -1,45 +0,0 @@
-"""XXL-JOB 执行器配置 —— 对标 Java XxlJobConfig.java"""
-
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-class XxlJobConfig(BaseSettings):
-    """XXL-JOB 执行器连接配置
-
-    对应 Java 侧 @Value("${xxl.job.*}") 注入的全部字段。
-    可通过环境变量 XXL_* 或 .env 文件覆盖默认值。
-    """
-
-    # Admin 调度中心地址（多个逗号分隔）
-    admin_addresses: str = "http://xxl-job-internal.piaoquantv.com/xxl-job-admin"
-
-    # 访问令牌（与 Admin 端 xxl.job.accessToken 一致）
-    access_token: str = ""
-
-    # 执行器 AppName（自动注册到 Admin 的标识）
-    app_name: str = "long-articles-agentic-supply"
-
-    # 执行器地址（留空则自动探测本机 IP）
-    address: str = ""
-
-    # 执行器 IP（留空则自动探测）
-    ip: str = ""
-
-    # 执行器端口（Quart 服务端口，与 Admin 通信用）
-    port: int = 7800
-
-    # 执行器运行日志存储路径
-    log_path: str = "/datalog/weblog/xxl-job/jobhandler"
-
-    # 执行器日志保留天数
-    log_retention_days: int = 7
-
-    # 注册心跳间隔（秒）
-    registry_interval: int = 30
-
-    model_config = SettingsConfigDict(
-        env_prefix="XXL_",
-        env_file=".env",
-        case_sensitive=False,
-        extra="ignore",
-    )

supply/server/__init__.py

@@ -1,12 +1,10 @@
 from supply.server.app import create_app
 from supply.server.endpoints import create_health_bp
 from supply.server.routes import server_routes, xxl_routes
-from supply.server.scheduler import TaskScheduler
 
 __all__ = [
     "create_app",
     "create_health_bp",
     "server_routes",
     "xxl_routes",
-    "TaskScheduler",
 ]

supply/server/app.py

@@ -1,4 +1,4 @@
-"""Quart + Hypercorn 应用工厂 —— 内建 DB + 日志服务 + XXL-JOB 生命周期"""
+"""Quart + Hypercorn 应用工厂 —— 全局配置驱动，DB + 日志 + XXL-JOB 生命周期内建"""
 
 import logging
 
@@ -6,50 +6,45 @@ from quart import Quart
 from quart_cors import cors
 
 from supply.core.bootstrap import AppContext
-from supply.config import (
-    AigcDatabaseConfig,
-    LongArticlesDatabaseConfig,
-    PiaoquanCrawlerDatabaseConfig,
-    AliyunLogConfig,
-)
+from supply.config import GlobalConfig
 from supply.infra.database.manager import DatabaseManager, create_backend
 from supply.infra.observability.logging import LogService
-import supply.jobs.test_handler  # noqa: F401 — 触发 @xxl_job 装饰器注册
-from supply.jobs.xxl_config import XxlJobConfig
-from supply.jobs.xxl_executor import XxlJobExecutor
+from supply.infra.xxl_jobs import XxlJobExecutor
+from supply.handlers import discover_and_register
 from supply.server.xxl_endpoints import set_xxl_executor
 
 logger = logging.getLogger(__name__)
 
-_db_configs = {
-    "aigc": AigcDatabaseConfig(),
-    "long_articles": LongArticlesDatabaseConfig(),
-    "piaoquan_crawler": PiaoquanCrawlerDatabaseConfig(),
-}
 
+def create_app(global_config: GlobalConfig = None) -> Quart:
+    """创建 Quart 应用
 
-def _build_backends() -> dict:
-    """将配置字典转换为后端实例字典"""
-    return {
-        name: create_backend(name, config)
-        for name, config in _db_configs.items()
-    }
+    统一从 GlobalConfig 获取所有配置，消除散落的 Config() 调用。
+    可选传入 custom_config 用于测试覆盖。
 
+    Args:
+        global_config: 可选的自定义配置，默认从环境变量加载
+    """
+    if global_config is None:
+        global_config = GlobalConfig()
 
-def create_app() -> Quart:
-    """创建 Quart 应用，内建 DB 连接池、日志服务和 XXL-JOB 执行器的生命周期管理"""
     logging.basicConfig(level=logging.INFO)
 
-    log_service = LogService(AliyunLogConfig())
-    db = DatabaseManager(_build_backends())
-    xxl_config = XxlJobConfig()
-    xxl_executor = XxlJobExecutor(xxl_config)
+    # 初始化各层依赖 —— 全部从 global_config 取值
+    log_service = LogService(global_config.aliyun_log)
+    db = DatabaseManager({
+        name: create_backend(name, config)
+        for name, config in global_config.active_db_configs().items()
+    })
+    discover_and_register()
+    xxl_executor = XxlJobExecutor(global_config.xxl_job)
 
     set_xxl_executor(xxl_executor)
 
     app = Quart(__name__)
     app = cors(app, allow_origin="*")
 
+    # 生命周期
     ctx = AppContext()
     ctx.on_startup(log_service.start)
     ctx.on_startup(db.init_pools)

supply/server/scheduler.py

@@ -1,402 +0,0 @@
-"""通用任务调度器 —— CAS 锁、并发控制、超时检测、后台执行
-
-重构要点（v2）:
-  - 不再继承 TaskHandler（scheduler 不是 handler，是编排层）
-  - 接收 TaskRegistry 实例，消除全局 _TASK_HANDLER_REGISTRY
-  - 使用结构化 LogRecord 替代自由 dict
-"""
-
-import asyncio
-import json
-import time
-from datetime import datetime, timedelta
-from typing import Any, Callable, Dict, List, Optional
-
-from supply.domain.task.config import TaskConfig, TaskStatus, get_default_task_config
-from supply.domain.task.base import TaskRegistry
-from supply.domain.task.exceptions import (
-    TaskConcurrencyError,
-    TaskValidationError,
-)
-from supply.infra import TaskScheduleResponse
-
-AlertCallback = Callable[[str, dict], Any]
-
-
-def _validate_table_name(table_name: str) -> str:
-    cleaned = table_name.strip().replace(" ", "_")
-    if not cleaned or "--" in cleaned or ";" in cleaned.lower():
-        raise TaskValidationError(f"非法的表名: {table_name}")
-    return cleaned
-
-
-def _validate_task_name(task_name: Any) -> str:
-    task_name = str(task_name).strip()
-    if not task_name:
-        raise TaskValidationError("task_name 不能为空")
-    return task_name
-
-
-def _format_error(e: Exception) -> str:
-    import traceback
-
-    return "".join(
-        traceback.format_exception(type(e), e, e.__traceback__)
-    )[-1000:]
-
-
-class TaskScheduler:
-    """通用任务调度器 —— CAS 锁、并发控制、超时检测、后台执行
-
-    使用方式:
-        registry = TaskRegistry()
-        # ... 注册 handler ...
-
-        scheduler = TaskScheduler(
-            data={"task_name": "my_crawl", "date_string": "2026-06-30"},
-            registry=registry,
-            log_service=log,
-            db_client=db,
-            trace_id="trace-123",
-            task_configs={"my_crawl": TaskConfig(timeout=600, max_concurrent=3)},
-            table_name="my_task_table",
-            alert_callback=my_alert_func,
-        )
-        result = await scheduler.deal()
-    """
-
-    def __init__(
-        self,
-        data: dict,
-        registry: TaskRegistry,
-        *,
-        log_service: Any = None,
-        db_client: Any = None,
-        trace_id: str = "",
-        task_configs: Optional[Dict[str, TaskConfig]] = None,
-        table_name: str = "task_manager",
-        alert_callback: Optional[AlertCallback] = None,
-    ):
-        self.data = data
-        self.registry = registry
-        self.log_service = log_service
-        self.db_client = db_client
-        self.trace_id = trace_id
-        self.table = _validate_table_name(table_name)
-        self.task_configs = task_configs or {}
-        self._alert_callback = alert_callback
-
-    # ==================== 日志 ====================
-
-    async def _log(self, event: str, **kwargs) -> None:
-        """结构化日志推送"""
-        if self.log_service is None:
-            return
-        try:
-            from supply.infra.observability.log_schema import (
-                LogRecord,
-                LogLevel,
-                LogCategory,
-            )
-
-            record = LogRecord(
-                level=LogLevel.INFO,
-                category=LogCategory.TASK,
-                event=event,
-                trace_id=self.trace_id,
-                task_name=self.data.get("task_name", ""),
-                extra=kwargs,
-            )
-            await self.log_service.log(record)
-        except Exception:
-            pass
-
-    async def _alert(self, title: str, detail: dict) -> None:
-        if self._alert_callback:
-            try:
-                await self._alert_callback(title, detail)
-            except Exception:
-                pass
-
-    def get_task_config(self, task_name: str) -> TaskConfig:
-        return self.task_configs.get(task_name, get_default_task_config())
-
-    # ==================== 数据库操作 ====================
-
-    async def _insert_or_ignore_task(self, task_name: str, date_str: str) -> None:
-        query = f"""
-            INSERT IGNORE INTO {self.table}
-            (date_string, task_name, start_timestamp, task_status, trace_id, data)
-            VALUES (%s, %s, %s, %s, %s, %s)
-        """
-        await self.db_client.async_save(
-            query=query,
-            params=(
-                date_str,
-                task_name,
-                int(time.time()),
-                TaskStatus.INIT,
-                self.trace_id,
-                json.dumps(self.data, ensure_ascii=False),
-            ),
-        )
-
-    async def _try_lock_task(self) -> bool:
-        query = f"""
-            UPDATE {self.table}
-            SET task_status = %s
-            WHERE trace_id = %s AND task_status = %s
-        """
-        result = await self.db_client.async_save(
-            query=query,
-            params=(TaskStatus.PROCESSING, self.trace_id, TaskStatus.INIT),
-        )
-        return bool(result)
-
-    async def _release_task(self, status: int) -> None:
-        query = f"""
-            UPDATE {self.table}
-            SET task_status = %s, finish_timestamp = %s
-            WHERE trace_id = %s AND task_status = %s
-        """
-        await self.db_client.async_save(
-            query=query,
-            params=(status, int(time.time()), self.trace_id, TaskStatus.PROCESSING),
-        )
-
-    async def _get_processing_tasks(self, task_name: str) -> List[Dict[str, Any]]:
-        query = f"""
-            SELECT trace_id, start_timestamp, data
-            FROM {self.table}
-            WHERE task_status = %s AND task_name = %s
-        """
-        rows = await self.db_client.async_fetch(
-            query=query,
-            params=(TaskStatus.PROCESSING, task_name),
-        )
-        return rows or []
-
-    # ==================== 并发/超时检查 ====================
-
-    async def _check_task_concurrency_and_timeout(self, task_name: str) -> None:
-        processing_tasks = await self._get_processing_tasks(task_name)
-        if not processing_tasks:
-            return
-
-        config = self.get_task_config(task_name)
-        current_time = int(time.time())
-
-        # 超时检测：自动释放超时任务
-        timeout_tasks = [
-            t
-            for t in processing_tasks
-            if current_time - t["start_timestamp"] > config.timeout
-        ]
-        if timeout_tasks:
-            await self._log(
-                "task_timeout_detected",
-                timeout_count=len(timeout_tasks),
-                timeout_traces=[t["trace_id"] for t in timeout_tasks],
-            )
-            await self._alert(
-                title=f"Task Timeout Alert: {task_name}",
-                detail={
-                    "task_name": task_name,
-                    "timeout_count": len(timeout_tasks),
-                    "timeout_threshold": config.timeout,
-                    "timeout_tasks": [
-                        {
-                            "trace_id": t["trace_id"],
-                            "running_time": current_time - t["start_timestamp"],
-                        }
-                        for t in timeout_tasks
-                    ],
-                },
-            )
-            for t in timeout_tasks:
-                await self._force_release_task(t["trace_id"], TaskStatus.FAILED)
-
-        # 并发限制
-        active_tasks = [
-            t
-            for t in processing_tasks
-            if current_time - t["start_timestamp"] <= config.timeout
-        ]
-        if len(active_tasks) >= config.max_concurrent:
-            await self._log(
-                "task_concurrency_limit",
-                current_count=len(active_tasks),
-                max_concurrent=config.max_concurrent,
-            )
-            await self._alert(
-                title=f"Task Concurrency Limit: {task_name}",
-                detail={
-                    "task_name": task_name,
-                    "current_count": len(active_tasks),
-                    "max_concurrent": config.max_concurrent,
-                    "active_tasks": [t["trace_id"] for t in active_tasks],
-                },
-            )
-            raise TaskConcurrencyError(
-                f"Task {task_name} has reached max concurrency limit "
-                f"({len(active_tasks)}/{config.max_concurrent})",
-                task_name=task_name,
-            )
-
-    # ==================== 任务执行 ====================
-
-    async def _run_with_guard(
-        self,
-        task_name: str,
-        date_str: str,
-        handler_callable: Callable,
-    ) -> dict:
-        """带保护的任务执行: 并发检查 → CAS 锁 → 后台执行 → 释放"""
-        # 1. 并发/超时检查
-        try:
-            await self._check_task_concurrency_and_timeout(task_name)
-        except TaskConcurrencyError as e:
-            return TaskScheduleResponse.fail("5005", str(e))
-
-        # 2. 创建任务记录并 CAS 获取锁
-        await self._insert_or_ignore_task(task_name, date_str)
-        if not await self._try_lock_task():
-            return TaskScheduleResponse.fail("5001", "Task is already processing")
-
-        # 3. 后台执行
-        async def _task_wrapper():
-            status = TaskStatus.FAILED
-            config = self.get_task_config(task_name)
-            start_time = time.time()
-
-            try:
-                await self._log("task_started")
-                result = await handler_callable()
-                status = result if isinstance(result, int) else TaskStatus.SUCCESS
-                duration = time.time() - start_time
-                await self._log(
-                    "task_completed",
-                    status=status,
-                    duration_ms=int(duration * 1000),
-                )
-            except Exception as e:
-                duration = time.time() - start_time
-                error_detail = _format_error(e)
-                await self._log(
-                    "task_failed",
-                    error=error_detail,
-                    duration_ms=int(duration * 1000),
-                )
-                if config.alert_on_failure:
-                    await self._alert(
-                        title=f"Task Failed: {task_name}",
-                        detail={
-                            "task_name": task_name,
-                            "trace_id": self.trace_id,
-                            "error": error_detail,
-                            "duration": duration,
-                        },
-                    )
-            finally:
-                await self._release_task(status)
-
-        asyncio.create_task(_task_wrapper(), name=f"{task_name}_{self.trace_id}")
-
-        return TaskScheduleResponse.success(
-            task_name=task_name,
-            data={
-                "code": 0,
-                "message": "Task started successfully",
-                "trace_id": self.trace_id,
-            },
-        )
-
-    # ==================== 任务管理接口 ====================
-
-    async def get_task_status(self, trace_id: Optional[str] = None) -> Optional[Dict[str, Any]]:
-        tid = trace_id or self.trace_id
-        query = f"SELECT * FROM {self.table} WHERE trace_id = %s"
-        return await self.db_client.async_fetch_one(query, params=(tid,))
-
-    async def cancel_task(self, trace_id: Optional[str] = None) -> bool:
-        tid = trace_id or self.trace_id
-        query = f"""
-            UPDATE {self.table}
-            SET task_status = %s, finish_timestamp = %s
-            WHERE trace_id = %s AND task_status IN (%s, %s)
-        """
-        result = await self.db_client.async_save(
-            query,
-            params=(
-                TaskStatus.FAILED,
-                int(time.time()),
-                tid,
-                TaskStatus.INIT,
-                TaskStatus.PROCESSING,
-            ),
-        )
-        if result:
-            await self._log("task_cancelled", trace_id=tid)
-        return bool(result)
-
-    async def retry_task(self, trace_id: Optional[str] = None) -> bool:
-        tid = trace_id or self.trace_id
-        query = f"""
-            UPDATE {self.table}
-            SET task_status = %s, start_timestamp = %s, finish_timestamp = NULL
-            WHERE trace_id = %s
-        """
-        result = await self.db_client.async_save(
-            query, params=(TaskStatus.INIT, int(time.time()), tid)
-        )
-        if result:
-            await self._log("task_retried", trace_id=tid)
-        return bool(result)
-
-    async def _force_release_task(self, trace_id: str, status: int) -> None:
-        query = f"""
-            UPDATE {self.table}
-            SET task_status = %s, finish_timestamp = %s
-            WHERE trace_id = %s
-        """
-        await self.db_client.async_save(
-            query, params=(status, int(time.time()), trace_id)
-        )
-        await self._log("task_force_released", trace_id=trace_id, status=status)
-
-    # ==================== 主入口 ====================
-
-    async def deal(self) -> dict:
-        """任务调度主入口"""
-        task_name = self.data.get("task_name")
-        if not task_name:
-            return TaskScheduleResponse.fail("4003", "task_name is required")
-
-        try:
-            task_name = _validate_task_name(task_name)
-        except TaskValidationError as e:
-            return TaskScheduleResponse.fail("4003", str(e))
-
-        date_str = self.data.get("date_string") or (
-            datetime.utcnow() + timedelta(hours=8)
-        ).strftime("%Y-%m-%d")
-
-        # 从 TaskRegistry 查找 handler
-        handler_cls = self.registry.get(task_name)
-        if handler_cls is None:
-            return TaskScheduleResponse.fail(
-                "4001",
-                f"Unknown task: {task_name}. "
-                f"Available tasks: {', '.join(self.registry.list_names())}",
-            )
-
-        # 实例化 handler
-        handler = handler_cls(log_service=self.log_service, db_client=self.db_client)
-
-        return await self._run_with_guard(
-            task_name,
-            date_str,
-            lambda: handler.execute(
-                type("Task", (), {"task_name": task_name, "params": self.data, "trace_id": self.trace_id})()
-            ),
-        )