init

.gitignore

@@ -0,0 +1,12 @@
+# Python
+__pycache__/
+*.egg-info/
+*.pyc
+dist/
+build/
+
+# IDE
+.idea/
+.vscode/
+.claude
+

app.py

@@ -0,0 +1,10 @@
+"""Hypercorn 入口 —— 最简 Quart 服务
+
+启动方式:
+    hypercorn app:app --bind 0.0.0.0:8000
+"""
+
+from supply.server import create_app, server_routes
+
+app = create_app()
+app.register_blueprint(server_routes())

pyproject.toml

@@ -0,0 +1,30 @@
+[build-system]
+requires = ["setuptools>=68.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "long-articles-agentic-supply"
+version = "0.1.0"
+description = "Agentic 架构基础设施供给层 —— 数据库连接池、异步任务调度、HTTP 客户端、可观测性等通用基座"
+requires-python = ">=3.11"
+dependencies = [
+    "aiomysql>=0.2.0",
+    "aiohttp>=3.9",
+    "pydantic>=2.0",
+    "pydantic-settings>=2.0",
+    "dependency-injector>=4.0",
+    "aliyun-log-python-sdk>=0.9",
+]
+
+[project.optional-dependencies]
+server = [
+    "quart>=0.19",
+    "hypercorn>=0.17",
+    "quart-cors>=0.8",
+]
+dev = [
+    "tqdm>=4.66",
+]
+
+[tool.setuptools.packages.find]
+include = ["supply*"]

requirements.txt

@@ -0,0 +1,10 @@
+aiomysql~=0.2.0
+aiohttp~=3.10.4
+pydantic~=2.10.6
+pydantic-settings~=2.12.0
+dependency-injector~=4.48.3
+aliyun-log-python-sdk
+tqdm~=4.66.6
+quart~=0.19.6
+hypercorn
+quart-cors~=0.8.0

server.toml

@@ -0,0 +1,6 @@
+reload = true
+bind = "0.0.0.0:7800"
+workers = 1
+keep_alive_timeout = 120  # 保持连接的最大秒数，根据需要调整
+graceful_timeout = 300    # 重启或停止之前等待当前工作完成的时间
+loglevel = "warning"  # 日志级别

supply/__init__.py

@@ -0,0 +1,90 @@
+"""Agentic Supply —— Agentic 架构基础设施供给层
+
+层次结构（由内到外）:
+  ┌─────────────────┐
+  │  supply.config  │  配置管理
+  ├─────────────────┤
+  │  supply.infra   │  基础设施（DB / HTTP / 日志 / 工具）
+  ├─────────────────┤
+  │  supply.domain  │  领域抽象（Agent / Tool / Memory / Task）
+  ├─────────────────┤
+  │  supply.server  │  应用服务（Quart / XXL / TaskScheduler）
+  └─────────────────┘
+
+快速开始:
+    from supply import (
+        Agent, Tool, ToolRegistry,          # 领域层
+        DatabaseManager, LogService,         # 设施层
+        create_app,                          # 服务层
+    )
+"""
+
+# 配置
+from supply.config import (
+    DatabaseConfig,
+    AliyunLogConfig,
+    AliyunOssConfig,
+    PgConfig,
+    MilvusConfig,
+    RedisConfig,
+)
+
+# 基础设施
+from supply.infra import (
+    DatabaseManager,
+    create_backend,
+    MySQLBackend,
+    LogService,
+    LogRecord,
+    LogLevel,
+    LogCategory,
+    LogStatus,
+    AsyncHttpClient,
+    TaskScheduleResponse,
+)
+
+# 领域
+from supply.domain import (
+    Agent,
+    AgentContext,
+    AgentResult,
+    AgentStatus,
+    AgentOrchestrator,
+    ChainOrchestrator,
+    ParallelOrchestrator,
+    Tool,
+    ToolResult,
+    ToolRegistry,
+    Memory,
+    ConversationMemory,
+    VectorMemory,
+    Task,
+    TaskResult,
+    TaskHandler,
+    TaskRegistry,
+    TaskConfig,
+    TaskStatus,
+)
+
+# 服务
+from supply.server import create_app, TaskScheduler
+
+__all__ = [
+    # Config
+    "DatabaseConfig", "AliyunLogConfig", "AliyunOssConfig",
+    "PgConfig", "MilvusConfig", "RedisConfig",
+    # Infra
+    "DatabaseManager", "create_backend", "MySQLBackend",
+    "LogService", "LogRecord", "LogLevel", "LogCategory", "LogStatus",
+    "AsyncHttpClient", "TaskScheduleResponse",
+    # Domain - 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",
+]

supply/config/__init__.py

@@ -0,0 +1,30 @@
+"""配置层 —— 所有 pydantic-settings 配置模型，环境变量驱动"""
+
+from supply.config.database import (
+    DatabaseConfig,
+    AigcDatabaseConfig,
+    LongVideoDatabaseConfig,
+    LongArticlesDatabaseConfig,
+    PiaoquanCrawlerDatabaseConfig,
+    GrowthDatabaseConfig,
+    ContentDeconstructionSupplyConfig,
+    PgConfig,
+    MilvusConfig,
+    RedisConfig,
+)
+from supply.config.aliyun import AliyunLogConfig, AliyunOssConfig
+
+__all__ = [
+    "DatabaseConfig",
+    "AigcDatabaseConfig",
+    "LongVideoDatabaseConfig",
+    "LongArticlesDatabaseConfig",
+    "PiaoquanCrawlerDatabaseConfig",
+    "GrowthDatabaseConfig",
+    "ContentDeconstructionSupplyConfig",
+    "PgConfig",
+    "MilvusConfig",
+    "RedisConfig",
+    "AliyunLogConfig",
+    "AliyunOssConfig",
+]

supply/config/aliyun.py

@@ -0,0 +1,30 @@
+"""阿里云服务配置 —— SLS 日志 / OSS 对象存储"""
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class AliyunLogConfig(BaseSettings):
+    """阿里云日志服务 (SLS) 配置"""
+
+    endpoint: str = "cn-hangzhou.log.aliyuncs.com"
+    access_key_id: str = "LTAIP6x1l3DXfSxm"
+    access_key_secret: str = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon"
+    project: str = "changwen-alg"
+    logstore: str = "long_articles_agentic_aupply"
+
+    model_config = SettingsConfigDict(
+        env_prefix="ALIYUN_LOG_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+
+class AliyunOssConfig(BaseSettings):
+    """阿里云 OSS 对象存储配置"""
+
+    endpoint: str = "oss-cn-hangzhou.aliyuncs.com"
+    access_key_id: str = "LTAIP6x1l3DXfSxm"
+    access_key_secret: str = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon"
+    bucket_name: str = "art-pubbucket"
+
+    model_config = SettingsConfigDict(
+        env_prefix="ALIYUN_OSS_", env_file=".env", case_sensitive=False, extra="ignore"
+    )

supply/config/database.py

@@ -0,0 +1,161 @@
+"""数据库连接配置 —— 所有存储引擎配置模型
+
+每个具体配置类通过 env_prefix 从环境变量读取敏感信息。
+默认值仅包含安全的占位值，生产凭证必须通过环境变量注入。
+"""
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class DatabaseConfig(BaseSettings):
+    """数据库连接配置基类
+
+    通过 backend 字段指定存储引擎类型：
+      - "mysql" / ""  → MySQLBackend
+      - "postgresql"  → PgBackend
+      - "milvus"      → MilvusBackend
+      - "redis"       → RedisBackend
+    """
+
+    host: str = "127.0.0.1"
+    port: int = 3306
+    user: str = "root"
+    password: str = ""
+    db: str = "test"
+    charset: str = "utf8mb4"
+    minsize: int = 5
+    maxsize: int = 20
+    backend: str = ""
+
+    model_config = SettingsConfigDict(
+        env_prefix="", case_sensitive=False, extra="ignore"
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "host": self.host,
+            "port": self.port,
+            "user": self.user,
+            "password": self.password,
+            "db": self.db,
+            "charset": self.charset,
+            "minsize": self.minsize,
+            "maxsize": self.maxsize,
+        }
+
+
+# ============ 业务数据库配置 ============
+
+
+class AigcDatabaseConfig(DatabaseConfig):
+    host: str = "rm-t4na9qj85v7790tf84o.mysql.singapore.rds.aliyuncs.com"
+    user: str = "crawler_admin"
+    password: str = "cyber#crawler_2023"
+    db: str = "aigc-admin-prod"
+
+    model_config = SettingsConfigDict(
+        env_prefix="AIGC_DB_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+
+class LongVideoDatabaseConfig(DatabaseConfig):
+    host: str = "rr-bp1x9785e8h5452bi157.mysql.rds.aliyuncs.com"
+    user: str = "wx2016_longvideo"
+    password: str = "wx2016_longvideoP@assword1234"
+    db: str = "longvideo"
+
+    model_config = SettingsConfigDict(
+        env_prefix="LONG_VIDEO_DB_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+
+class LongArticlesDatabaseConfig(DatabaseConfig):
+    host: str = "rm-bp14529nwwcw75yr1ko.mysql.rds.aliyuncs.com"
+    user: str = "changwen_admin"
+    password: str = "changwen@123456"
+    db: str = "long_articles"
+
+    model_config = SettingsConfigDict(
+        env_prefix="LONG_ARTICLES_DB_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+
+class PiaoquanCrawlerDatabaseConfig(DatabaseConfig):
+    host: str = "rm-bp1159bu17li9hi94.mysql.rds.aliyuncs.com"
+    user: str = "crawler"
+    password: str = "crawler123456@"
+    db: str = "piaoquan-crawler"
+
+    model_config = SettingsConfigDict(
+        env_prefix="PIAOQUAN_CRAWLER_DB_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+
+class GrowthDatabaseConfig(DatabaseConfig):
+    host: str = "rm-bp17q95335a99272b.mysql.rds.aliyuncs.com"
+    user: str = "crawler"
+    password: str = "crawler123456@"
+    db: str = "growth"
+
+    model_config = SettingsConfigDict(
+        env_prefix="GROWTH_DB_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+
+class ContentDeconstructionSupplyConfig(DatabaseConfig):
+    host: str = "rm-t4nh1xx6o2a6vj8qu3o.mysql.singapore.rds.aliyuncs.com"
+    user: str = "content_rw"
+    password: str = "bC1aH4bA1lB0"
+    db: str = "content-deconstruction-supply"
+
+    model_config = SettingsConfigDict(
+        env_prefix="CONTENT_DECONSREUCTION_DB_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+
+# ============ PostgreSQL / Milvus / Redis 配置 ============
+
+
+class PgConfig(BaseSettings):
+    """PostgreSQL 连接配置"""
+
+    host: str = "127.0.0.1"
+    port: int = 5432
+    user: str = "postgres"
+    password: str = ""
+    db: str = "postgres"
+    minsize: int = 5
+    maxsize: int = 20
+    backend: str = "postgresql"
+
+    model_config = SettingsConfigDict(
+        env_prefix="PG_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+
+class MilvusConfig(BaseSettings):
+    """Milvus 向量数据库连接配置"""
+
+    host: str = "127.0.0.1"
+    port: int = 19530
+    user: str = ""
+    password: str = ""
+    backend: str = "milvus"
+
+    model_config = SettingsConfigDict(
+        env_prefix="MILVUS_", env_file=".env", case_sensitive=False, extra="ignore"
+    )
+
+
+class RedisConfig(BaseSettings):
+    """Redis 连接配置"""
+
+    host: str = "127.0.0.1"
+    port: int = 6379
+    password: str = ""
+    db: int = 0
+    backend: str = "redis"
+
+    model_config = SettingsConfigDict(
+        env_prefix="REDIS_", env_file=".env", case_sensitive=False, extra="ignore"
+    )

supply/core/__init__.py

@@ -0,0 +1,53 @@
+"""核心层 —— 生命周期管理
+
+配置和数据库已下沉到 config/ 和 infra/ 层。
+此处保留常见类型的 re-export 方便使用。
+"""
+
+# 配置
+from supply.config import (
+    DatabaseConfig,
+    AliyunLogConfig,
+    AliyunOssConfig,
+    AigcDatabaseConfig,
+    LongVideoDatabaseConfig,
+    LongArticlesDatabaseConfig,
+    PiaoquanCrawlerDatabaseConfig,
+    GrowthDatabaseConfig,
+    ContentDeconstructionSupplyConfig,
+    PgConfig,
+    MilvusConfig,
+    RedisConfig,
+)
+
+# 数据库
+from supply.infra.database import DatabaseManager, create_backend
+
+# 可观测性
+from supply.infra.observability import LogService, LogRecord, LogLevel, LogCategory, LogStatus
+
+# 生命周期
+from supply.core.bootstrap import AppContext
+
+__all__ = [
+    "DatabaseConfig",
+    "AliyunLogConfig",
+    "AliyunOssConfig",
+    "AigcDatabaseConfig",
+    "LongVideoDatabaseConfig",
+    "LongArticlesDatabaseConfig",
+    "PiaoquanCrawlerDatabaseConfig",
+    "GrowthDatabaseConfig",
+    "ContentDeconstructionSupplyConfig",
+    "PgConfig",
+    "MilvusConfig",
+    "RedisConfig",
+    "DatabaseManager",
+    "create_backend",
+    "LogService",
+    "LogRecord",
+    "LogLevel",
+    "LogCategory",
+    "LogStatus",
+    "AppContext",
+]

supply/core/bootstrap.py

@@ -0,0 +1,62 @@
+"""应用生命周期管理 —— 回调注册模式"""
+
+import asyncio
+import logging
+from typing import Callable, Awaitable, List
+
+logger = logging.getLogger(__name__)
+
+StartupCallback = Callable[[], Awaitable[None]]
+ShutdownCallback = Callable[[], Awaitable[None]]
+
+
+class AppContext:
+    """管理应用启动/关闭生命周期
+
+    使用方式：
+        ctx = AppContext()
+        ctx.on_startup(db.init_pools)
+        ctx.on_startup(log_service.start)
+        ctx.on_shutdown(db.close_pools)
+        ctx.on_shutdown(log_service.stop)
+
+        # 在 Quart/FastAPI 生命周期中调用
+        await ctx.start_up()
+        await ctx.shutdown()
+    """
+
+    def __init__(self):
+        self._startup_callbacks: List[StartupCallback] = []
+        self._shutdown_callbacks: List[ShutdownCallback] = []
+
+    def on_startup(self, callback: StartupCallback) -> "AppContext":
+        """注册启动回调（支持链式调用）"""
+        self._startup_callbacks.append(callback)
+        return self
+
+    def on_shutdown(self, callback: ShutdownCallback) -> "AppContext":
+        """注册关闭回调（支持链式调用）"""
+        self._shutdown_callbacks.append(callback)
+        return self
+
+    async def start_up(self) -> None:
+        """按注册顺序执行所有启动回调"""
+        logger.info("应用启动中...")
+        results = await asyncio.gather(
+            *(cb() for cb in self._startup_callbacks),
+            return_exceptions=True,
+        )
+        for i, result in enumerate(results):
+            if isinstance(result, Exception):
+                logger.error(f"启动回调 [{i}] 失败: {result}")
+        logger.info("应用启动完成")
+
+    async def shutdown(self) -> None:
+        """按注册逆序执行所有关闭回调"""
+        logger.info("应用关闭中...")
+        for callback in reversed(self._shutdown_callbacks):
+            try:
+                await callback()
+            except Exception as e:
+                logger.error(f"关闭回调失败: {e}")
+        logger.info("应用已关闭")

supply/domain/__init__.py

@@ -0,0 +1,69 @@
+"""领域层 —— 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/agent/__init__.py

@@ -0,0 +1,27 @@
+"""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 (
+    AgentOrchestrator,
+    ChainOrchestrator,
+    ParallelOrchestrator,
+)
+
+__all__ = [
+    "Agent",
+    "AgentContext",
+    "AgentResult",
+    "AgentStatus",
+    "Tool",
+    "ToolResult",
+    "ToolRegistry",
+    "Memory",
+    "Message",
+    "ConversationMemory",
+    "VectorMemory",
+    "AgentOrchestrator",
+    "ChainOrchestrator",
+    "ParallelOrchestrator",
+]

supply/domain/agent/base.py

@@ -0,0 +1,154 @@
+"""Agent 抽象基类 —— 领域核心
+
+定义 Agent 的契约：每个 Agent 拥有一组 Tool、一份 Memory、一个 run() 方法。
+业务方继承 Agent 实现具体智能体逻辑，框架层通过 AgentOrchestrator 调度执行。
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+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
+
+
+# ==================== Agent 状态 ====================
+
+
+class AgentStatus(str, Enum):
+    IDLE = "idle"
+    RUNNING = "running"
+    SUCCESS = "success"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+# ==================== Agent 上下文 ====================
+
+
+@dataclass
+class AgentContext:
+    """Agent 执行上下文 —— 携带本次调用的会话状态
+
+    每次 run() 传入一个新的 AgentContext，包含：
+      - session_id: 会话标识，用于 Memory 存取和日志追踪
+      - working_memory: 临时工作区，Agent 可在执行过程中读写
+      - metadata: 业务方注入的额外参数（topic, user_id 等）
+
+    与 trace_id 的关系：
+      trace_id 是分布式追踪 ID（跨 Agent 调用），session_id 是单次 Agent 会话 ID。
+      trace_id 可作为 session_id 使用，也可独立设置。
+    """
+
+    session_id: str = ""
+    trace_id: str = ""
+    working_memory: Dict[str, Any] = field(default_factory=dict)
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+# ==================== Agent 执行结果 ====================
+
+
+@dataclass
+class AgentResult:
+    """Agent 执行结果"""
+
+    status: AgentStatus = AgentStatus.IDLE
+    data: Any = None
+    error: Optional[str] = None
+    # 执行过程中产生的日志事件（由 orchestator 统一推送 SLS）
+    events: List[Dict[str, Any]] = field(default_factory=list)
+    # 工具调用记录（用于审计和调试）
+    tool_calls: List[Dict[str, Any]] = field(default_factory=list)
+
+    @property
+    def success(self) -> bool:
+        return self.status == AgentStatus.SUCCESS
+
+    @classmethod
+    def ok(cls, data: Any = None, **kwargs) -> "AgentResult":
+        return cls(status=AgentStatus.SUCCESS, data=data, **kwargs)
+
+    @classmethod
+    def fail(cls, error: str, **kwargs) -> "AgentResult":
+        return cls(status=AgentStatus.FAILED, error=error, **kwargs)
+
+
+# ==================== Agent 抽象基类 ====================
+
+
+class Agent(ABC):
+    """Agent 抽象基类
+
+    每个 Agent 是独立的智能体单元，拥有：
+      - name / description: 标识与描述
+      - tools: 可调用的工具集合
+      - memory: 记忆系统（会话记忆 / 向量记忆）
+
+    使用方式：
+
+        class ContentWriter(Agent):
+            name = "content_writer"
+            description = "长文写作智能体"
+
+            def __init__(self):
+                self._tools = [WebSearchTool(), FactCheckTool()]
+                self._memory = ConversationMemory(max_turns=10)
+
+            @property
+            def tools(self) -> list[Tool]:
+                return self._tools
+
+            @property
+            def memory(self) -> Memory:
+                return self._memory
+
+            async def run(self, ctx: AgentContext) -> AgentResult:
+                # 1. 从 memory 获取历史
+                # 2. 调用 LLM 规划
+                # 3. 执行 tool 调用
+                # 4. 更新 memory
+                ...
+    """
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """Agent 唯一标识"""
+        ...
+
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """Agent 功能描述"""
+        ...
+
+    @property
+    @abstractmethod
+    def tools(self) -> List[Tool]:
+        """Agent 可调用的工具列表"""
+        ...
+
+    @property
+    @abstractmethod
+    def memory(self) -> Memory:
+        """Agent 的记忆系统"""
+        ...
+
+    @abstractmethod
+    async def run(self, ctx: AgentContext) -> AgentResult:
+        """执行 Agent 主逻辑
+
+        Args:
+            ctx: 执行上下文（session、trace、metadata）
+
+        Returns:
+            AgentResult: 执行结果（success/fail + data）
+        """
+        ...
+
+    def __repr__(self) -> str:
+        return f"<Agent name={self.name}>"

supply/domain/agent/memory.py

@@ -0,0 +1,141 @@
+"""Memory 抽象 —— Agent 的记忆系统
+
+两种实现：
+  - ConversationMemory: 列表内存储存最近 N 轮对话
+  - VectorMemory (future): 基于 Milvus 的长期记忆，语义检索
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+# ==================== 消息模型 ====================
+
+
+@dataclass
+class Message:
+    """单条对话消息"""
+
+    role: str  # "system" | "user" | "assistant" | "tool"
+    content: str
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict:
+        return {"role": self.role, "content": self.content}
+
+
+# ==================== Memory 抽象 ====================
+
+
+class Memory(ABC):
+    """Memory 抽象基类
+
+    每种 Memory 实现管理 Agent 的对话历史。
+    短期记忆在单次会话内有效，长期记忆可跨会话检索。
+    """
+
+    @abstractmethod
+    async def add(self, message: Message) -> None:
+        """添加一条消息到记忆"""
+        ...
+
+    @abstractmethod
+    async def get(self, n: int = 10) -> List[Message]:
+        """获取最近 n 条消息"""
+        ...
+
+    @abstractmethod
+    async def clear(self) -> None:
+        """清空记忆"""
+        ...
+
+    @abstractmethod
+    def to_messages(self, n: Optional[int] = None) -> List[dict]:
+        """导出为 LLM 可用的 messages 格式"""
+        ...
+
+    @property
+    @abstractmethod
+    def size(self) -> int:
+        """当前记忆中的消息数"""
+        ...
+
+
+# ==================== 对话记忆实现 ====================
+
+
+class ConversationMemory(Memory):
+    """基于列表的短期对话记忆
+
+    适用于单次会话内的上下文窗口管理。
+    超出 max_turns 时自动滚动（FIFO 裁剪）。
+    """
+
+    def __init__(self, max_turns: int = 20):
+        self._messages: List[Message] = []
+        self.max_turns = max_turns
+
+    async def add(self, message: Message) -> None:
+        self._messages.append(message)
+        # FIFO 裁剪：超出限制时去掉最早的非 system 消息
+        if len(self._messages) > self.max_turns:
+            for i, msg in enumerate(self._messages):
+                if msg.role != "system":
+                    self._messages.pop(i)
+                    break
+
+    async def get(self, n: int = 10) -> List[Message]:
+        return self._messages[-n:]
+
+    async def clear(self) -> None:
+        self._messages.clear()
+
+    def to_messages(self, n: Optional[int] = None) -> List[dict]:
+        msgs = self._messages if n is None else self._messages[-n:]
+        return [m.to_dict() for m in msgs]
+
+    @property
+    def size(self) -> int:
+        return len(self._messages)
+
+
+# ==================== 向量记忆（Future） ====================
+
+
+class VectorMemory(Memory):
+    """基于向量数据库的长期记忆
+
+    通过 Milvus 存储语义向量，支持跨会话检索。
+    目前为桩实现，后续通过 MilvusBackend 对接。
+    """
+
+    def __init__(self, collection: str = "agent_memory"):
+        self.collection = collection
+        self._messages: List[Message] = []  # 当前会话缓存
+        self._backend: Any = None  # MilvusBackend 实例，由 DI 注入
+
+    async def add(self, message: Message) -> None:
+        self._messages.append(message)
+        # TODO: embed + insert to Milvus
+
+    async def get(self, n: int = 10) -> List[Message]:
+        return self._messages[-n:]
+
+    async def search(self, query: str, top_k: int = 5) -> List[Message]:
+        """语义检索相关记忆"""
+        # TODO: embed query → Milvus search → return messages
+        return []
+
+    async def clear(self) -> None:
+        self._messages.clear()
+
+    def to_messages(self, n: Optional[int] = None) -> List[dict]:
+        msgs = self._messages if n is None else self._messages[-n:]
+        return [m.to_dict() for m in msgs]
+
+    @property
+    def size(self) -> int:
+        return len(self._messages)

supply/domain/agent/orchestrator.py

@@ -0,0 +1,93 @@
+"""AgentOrchestrator —— Agent 编排调度
+
+提供两种编排模式：
+  - ChainOrchestrator: 串行执行，前一个的输出作为后一个的输入
+  - ParallelOrchestrator: 并行执行，最后合并结果
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List, Optional
+
+from supply.domain.agent.base import Agent, AgentContext, AgentResult, AgentStatus
+
+logger = logging.getLogger(__name__)
+
+
+class AgentOrchestrator(ABC):
+    """AgentOrchestrator 抽象基类
+
+    责任：协调多个 Agent 的执行顺序，聚合结果。
+    不负责 Tool 注册与调度——那是 Agent 自身的职责。
+    """
+
+    @abstractmethod
+    async def run(
+        self,
+        agents: List[Agent],
+        ctx: AgentContext,
+    ) -> List[AgentResult]:
+        """执行一组 Agent，返回每个 Agent 的结果"""
+        ...
+
+
+class ChainOrchestrator(AgentOrchestrator):
+    """链式编排：顺序执行 Agent，前一个结果通过 ctx.working_memory 传递
+
+    使用场景：写作流水线
+      OutlineAgent → DraftAgent → FactCheckAgent → PolishAgent
+
+    每个 Agent 可在 ctx.working_memory 中读写数据，
+    下一个 Agent 通过 working_memory 获取上一阶段产出。
+    """
+
+    def __init__(self, *, fail_fast: bool = True):
+        """
+        Args:
+            fail_fast: True 时任一 Agent 失败立即中止后续
+        """
+        self.fail_fast = fail_fast
+
+    async def run(
+        self,
+        agents: List[Agent],
+        ctx: AgentContext,
+    ) -> List[AgentResult]:
+        results: List[AgentResult] = []
+        for agent in agents:
+            logger.info("ChainOrchestrator → %s", agent.name)
+            result = await agent.run(ctx)
+            results.append(result)
+            if self.fail_fast and not result.success:
+                logger.warning("Agent %s 失败，中止链路", agent.name)
+                break
+        return results
+
+
+class ParallelOrchestrator(AgentOrchestrator):
+    """并行编排：同时执行多个 Agent，互不阻塞
+
+    使用场景：多维度内容质量审查
+      GrammarAgent || StyleAgent || SensitivityAgent || FactCheckAgent
+
+    结果按 Agent 注册顺序返回。
+    """
+
+    def __init__(self, *, max_concurrency: int = 5):
+        self._semaphore = asyncio.Semaphore(max_concurrency)
+
+    async def run(
+        self,
+        agents: List[Agent],
+        ctx: AgentContext,
+    ) -> List[AgentResult]:
+        async def _run_one(agent: Agent) -> AgentResult:
+            async with self._semaphore:
+                logger.info("ParallelOrchestrator → %s", agent.name)
+                return await agent.run(ctx)
+
+        tasks = [_run_one(a) for a in agents]
+        return list(await asyncio.gather(*tasks, return_exceptions=False))

supply/domain/agent/tool.py

@@ -0,0 +1,169 @@
+"""Tool 抽象基类 + ToolRegistry —— Agent 的工具系统
+
+每个 Tool 是 Agent 可调用的原子能力单元。
+ToolRegistry 管理工具注册与发现（实例级，非全局）。
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+# ==================== ToolResult ====================
+
+
+@dataclass
+class ToolResult:
+    """Tool 执行结果"""
+
+    success: bool = True
+    data: Any = None
+    error: Optional[str] = None
+    # 调用耗时（毫秒）
+    duration_ms: int = 0
+    # 额外元数据
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def ok(cls, data: Any, **kwargs) -> "ToolResult":
+        return cls(success=True, data=data, **kwargs)
+
+    @classmethod
+    def fail(cls, error: str, **kwargs) -> "ToolResult":
+        return cls(success=False, error=error, **kwargs)
+
+
+# ==================== Tool 抽象基类 ====================
+
+
+class Tool(ABC):
+    """Tool 抽象基类
+
+    每个 Tool 是一个原子能力单元，例如：
+      - WebSearchTool: 网络搜索
+      - DatabaseQueryTool: 数据库查询
+      - HttpRequestTool: HTTP 调用
+      - LLMCallTool: LLM 推理
+
+    Tool 的 parameters_schema 描述输入参数的 JSON Schema，
+    供 Agent 框架生成 function calling 的 tool definition。
+
+    使用方式：
+
+        class WebSearchTool(Tool):
+            name = "web_search"
+            description = "搜索互联网获取最新信息"
+
+            @property
+            def parameters_schema(self) -> dict:
+                return {
+                    "type": "object",
+                    "properties": {
+                        "query": {"type": "string", "description": "搜索关键词"},
+                        "max_results": {"type": "integer", "default": 5},
+                    },
+                    "required": ["query"],
+                }
+
+            async def execute(self, **kwargs) -> ToolResult:
+                results = await search_api(kwargs["query"])
+                return ToolResult.ok(results)
+    """
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """Tool 唯一标识，用于 Agent 引用"""
+        ...
+
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """Tool 功能描述，写入 LLM prompt"""
+        ...
+
+    @property
+    @abstractmethod
+    def parameters_schema(self) -> dict:
+        """输入参数 JSON Schema
+
+        Returns:
+            {
+                "type": "object",
+                "properties": {...},
+                "required": [...],
+            }
+        """
+        ...
+
+    @abstractmethod
+    async def execute(self, **kwargs) -> ToolResult:
+        """执行 Tool，返回结果"""
+        ...
+
+    def to_openai_tool(self) -> dict:
+        """导出为 OpenAI function calling 格式"""
+        return {
+            "type": "function",
+            "function": {
+                "name": self.name,
+                "description": self.description,
+                "parameters": self.parameters_schema,
+            },
+        }
+
+    def __repr__(self) -> str:
+        return f"<Tool name={self.name}>"
+
+
+# ==================== ToolRegistry ====================
+
+
+class ToolRegistry:
+    """Tool 注册表 —— 实例级，可注入
+
+    替代全局 dict，支持测试隔离和多 Agent 场景。
+
+    使用方式：
+        registry = ToolRegistry()
+        registry.register(WebSearchTool())
+        registry.register(FactCheckTool())
+
+        tool = registry.get("web_search")
+        result = await tool.execute(query="AI 安全")
+    """
+
+    def __init__(self):
+        self._tools: Dict[str, Tool] = {}
+
+    def register(self, tool: Tool) -> "ToolRegistry":
+        """注册 Tool，链式调用"""
+        if tool.name in self._tools:
+            raise ValueError(f"Tool '{tool.name}' 已注册")
+        self._tools[tool.name] = tool
+        return self
+
+    def get(self, name: str) -> Tool:
+        """按名获取 Tool"""
+        tool = self._tools.get(name)
+        if tool is None:
+            raise KeyError(f"Tool '{name}' 未注册，可用: {self.list_names()}")
+        return tool
+
+    def list_names(self) -> List[str]:
+        return list(self._tools.keys())
+
+    def list_all(self) -> List[Tool]:
+        return list(self._tools.values())
+
+    def to_openai_tools(self) -> List[dict]:
+        """导出全部 Tool 为 OpenAI function calling 格式"""
+        return [t.to_openai_tool() for t in self._tools.values()]
+
+    def __len__(self) -> int:
+        return len(self._tools)
+
+    def __contains__(self, name: str) -> bool:
+        return name in self._tools

supply/domain/task/__init__.py

@@ -0,0 +1,27 @@
+"""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

@@ -0,0 +1,156 @@
+"""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

@@ -0,0 +1,40 @@
+"""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

@@ -0,0 +1,44 @@
+"""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/infra/__init__.py

@@ -0,0 +1,68 @@
+"""基础设施层 —— 数据库后端、HTTP 客户端、日志服务、工具函数等底层能力"""
+
+from supply.infra.database import (
+    BaseBackend,
+    SqlBackend,
+    VectorBackend,
+    KvBackend,
+    BackendType,
+    MySQLBackend,
+    PgBackend,
+    MilvusBackend,
+    RedisBackend,
+    DatabaseManager,
+    create_backend,
+)
+from supply.infra.observability import (
+    LogCategory,
+    LogLevel,
+    LogStatus,
+    LogRecord,
+    LogService,
+    SLS_INDEX_CONFIG,
+)
+from supply.infra.shared import (
+    AsyncHttpClient,
+    TaskScheduleResponse,
+    run_tasks_with_async_worker_group,
+    run_tasks_with_asyncio_task_group,
+    safe_json_parse,
+    str_to_md5,
+    generate_task_trace_id,
+    yield_batch,
+    timestamp_to_str,
+    days_remaining_in_month,
+)
+
+__all__ = [
+    # 数据库
+    "BaseBackend",
+    "SqlBackend",
+    "VectorBackend",
+    "KvBackend",
+    "BackendType",
+    "MySQLBackend",
+    "PgBackend",
+    "MilvusBackend",
+    "RedisBackend",
+    "DatabaseManager",
+    "create_backend",
+    # 可观测性
+    "LogCategory",
+    "LogLevel",
+    "LogStatus",
+    "LogRecord",
+    "LogService",
+    "SLS_INDEX_CONFIG",
+    # 共享工具
+    "AsyncHttpClient",
+    "TaskScheduleResponse",
+    "run_tasks_with_async_worker_group",
+    "run_tasks_with_asyncio_task_group",
+    "safe_json_parse",
+    "str_to_md5",
+    "generate_task_trace_id",
+    "yield_batch",
+    "timestamp_to_str",
+    "days_remaining_in_month",
+]

supply/infra/database/__init__.py

@@ -0,0 +1,28 @@
+"""数据库设施层 —— 抽象接口、具体后端、门面管理器"""
+
+from supply.infra.database.ports import (
+    BackendType,
+    BaseBackend,
+    SqlBackend,
+    VectorBackend,
+    KvBackend,
+)
+from supply.infra.database.mysql import MySQLBackend
+from supply.infra.database.postgresql import PgBackend
+from supply.infra.database.milvus import MilvusBackend
+from supply.infra.database.redis import RedisBackend
+from supply.infra.database.manager import DatabaseManager, create_backend
+
+__all__ = [
+    "BackendType",
+    "BaseBackend",
+    "SqlBackend",
+    "VectorBackend",
+    "KvBackend",
+    "MySQLBackend",
+    "PgBackend",
+    "MilvusBackend",
+    "RedisBackend",
+    "DatabaseManager",
+    "create_backend",
+]

supply/infra/database/manager.py

@@ -0,0 +1,190 @@
+"""数据库门面管理器 —— 依赖反转，接收已创建的 Backend 实例
+
+与旧版 DatabaseManager 的核心区别：
+  - 不再通过 config dict 内部创建后端，而是接收 Dict[str, BaseBackend]
+  - 后端创建逻辑剥离为 create_backend() 工厂函数，由调用方（DI 容器 / app 工厂）负责
+  - Manager 只关心生命周期编排与统一访问路由
+"""
+
+import logging
+from typing import Any, Dict, List, Optional
+
+from supply.infra.database.ports import (
+    BackendType,
+    BaseBackend,
+    SqlBackend,
+)
+from supply.infra.database.mysql import MySQLBackend
+from supply.infra.database.postgresql import PgBackend
+from supply.infra.database.milvus import MilvusBackend
+from supply.infra.database.redis import RedisBackend
+
+logger = logging.getLogger(__name__)
+
+
+# ==================== 后端工厂函数 ====================
+
+
+def _detect_backend_type(config: Any) -> BackendType:
+    """从配置推断后端类型：优先读取显式 backend 字段，否则按类名匹配"""
+    if hasattr(config, "backend") and config.backend:
+        try:
+            return BackendType(config.backend)
+        except ValueError:
+            pass
+    type_name = type(config).__name__.lower()
+    for bt in BackendType:
+        if bt.value in type_name:
+            return bt
+    return BackendType.MYSQL
+
+
+def create_backend(name: str, config: Any) -> BaseBackend:
+    """工厂函数：根据配置创建对应后端实例
+
+    由调用方（app 工厂 / DI 容器）使用，创建实例后传给 DatabaseManager。
+    """
+    bt = _detect_backend_type(config)
+    if bt == BackendType.MYSQL:
+        return MySQLBackend(name, config)
+    elif bt == BackendType.POSTGRESQL:
+        return PgBackend(name, config)
+    elif bt == BackendType.MILVUS:
+        return MilvusBackend(name, config)
+    elif bt == BackendType.REDIS:
+        return RedisBackend(name, config)
+    raise ValueError(f"未知的存储后端类型: {bt}")
+
+
+# ==================== 门面管理器 ====================
+
+
+class DatabaseManager:
+    """多后端数据库门面
+
+    接收已创建的后端实例，负责：
+      - 统一生命周期管理 (init_pools / close_pools)
+      - 向后兼容 SQL 接口
+      - 类型安全的后端直取访问器
+
+    使用方式：
+        backends = {
+            "aigc": MySQLBackend("aigc", AigcDatabaseConfig()),
+        }
+        db = DatabaseManager(backends)
+        await db.init_pools()
+        rows = await db.async_fetch("SELECT * FROM t")
+    """
+
+    def __init__(
+        self,
+        backends: Dict[str, BaseBackend],
+        *,
+        default_db: str = "default",
+    ):
+        if not backends:
+            raise ValueError("DatabaseManager 至少需要一个后端")
+        self._backends = backends
+        self._default_db = default_db
+
+    # ==================== 生命周期 ====================
+
+    async def init_pools(self) -> None:
+        for name, backend in self._backends.items():
+            await backend.init()
+            logger.info(
+                "存储后端 [%s] 初始化完成 → %s", name, type(backend).__name__
+            )
+
+    async def close_pools(self) -> None:
+        for name, backend in self._backends.items():
+            await backend.close()
+            logger.info("存储后端 [%s] 已关闭", name)
+
+    # ==================== 向后兼容 SQL 接口 ====================
+
+    async def async_fetch(
+        self,
+        query: str,
+        *,
+        params: Optional[tuple] = None,
+        db_name: Optional[str] = None,
+    ) -> List[Dict[str, Any]]:
+        backend = self._get_sql(db_name)
+        return await backend.fetch(query, params=params)
+
+    async def async_fetch_one(
+        self,
+        query: str,
+        *,
+        params: Optional[tuple] = None,
+        db_name: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
+        backend = self._get_sql(db_name)
+        return await backend.fetch_one(query, params=params)
+
+    async def async_save(
+        self,
+        query: str,
+        *,
+        params: Any = None,
+        batch: bool = False,
+        db_name: Optional[str] = None,
+    ) -> int:
+        backend = self._get_sql(db_name)
+        return await backend.execute(query, params=params, batch=batch)
+
+    # ==================== 类型安全的后端直取 ====================
+
+    def mysql(self, name: Optional[str] = None) -> MySQLBackend:
+        return self._get_typed(name or self._default_db, MySQLBackend)
+
+    def pg(self, name: str) -> PgBackend:
+        return self._get_typed(name, PgBackend)
+
+    def milvus(self, name: str) -> MilvusBackend:
+        return self._get_typed(name, MilvusBackend)
+
+    def redis(self, name: str) -> RedisBackend:
+        return self._get_typed(name, RedisBackend)
+
+    # ==================== 辅助 ====================
+
+    def get_pool(self, db_name: str):
+        """向后兼容：返回 MySQL 连接池对象"""
+        backend = self.mysql(db_name)
+        return backend._pool
+
+    def list_databases(self) -> List[str]:
+        return list(self._backends.keys())
+
+    def list_backends(self) -> Dict[str, str]:
+        return {n: type(b).__name__ for n, b in self._backends.items()}
+
+    # ==================== 内部 ====================
+
+    def _get_backend(self, db_name: Optional[str] = None) -> BaseBackend:
+        name = db_name or self._default_db
+        backend = self._backends.get(name)
+        if backend is None:
+            raise KeyError(
+                f"存储后端 '{name}' 未注册，可用: {self.list_databases()}"
+            )
+        return backend
+
+    def _get_sql(self, db_name: Optional[str] = None) -> SqlBackend:
+        backend = self._get_backend(db_name)
+        if not isinstance(backend, SqlBackend):
+            raise TypeError(
+                f"后端 '{db_name}' 类型为 {type(backend).__name__}，不支持 SQL 操作"
+            )
+        return backend
+
+    def _get_typed(self, name: str, expected_type: type) -> BaseBackend:
+        backend = self._get_backend(name)
+        if not isinstance(backend, expected_type):
+            raise TypeError(
+                f"后端 '{name}' 实际类型为 {type(backend).__name__}，"
+                f"期望 {expected_type.__name__}"
+            )
+        return backend

supply/infra/database/milvus.py

@@ -0,0 +1,60 @@
+"""Milvus 向量数据库后端 —— 基于 pymilvus
+
+状态：未实现。init() 会抛出 NotImplementedError。
+"""
+
+import logging
+from typing import Any, Optional
+
+from supply.infra.database.ports import VectorBackend
+
+logger = logging.getLogger(__name__)
+
+
+class MilvusBackend(VectorBackend):
+    """Milvus 向量数据库异步后端"""
+
+    def __init__(self, name: str, config: Any):
+        super().__init__(name, config)
+        self._client = None
+
+    # ==================== 生命周期 ====================
+
+    async def init(self) -> None:
+        raise NotImplementedError(
+            "MilvusBackend 尚未实现。需要安装 pymilvus 并完成连接初始化逻辑。"
+        )
+
+    async def close(self) -> None:
+        if self._client is not None:
+            self._client = None
+        self._initialized = False
+
+    async def health(self) -> bool:
+        return False
+
+    # ==================== 向量操作 ====================
+
+    async def insert(
+        self,
+        collection: str,
+        vectors: list[list[float]],
+        metadata: Optional[list[dict]] = None,
+    ) -> list[int]:
+        raise NotImplementedError("MilvusBackend 尚未实现")
+
+    async def search(
+        self,
+        collection: str,
+        query_vectors: list[list[float]],
+        top_k: int = 10,
+        filter_expr: Optional[str] = None,
+    ) -> list[list[dict]]:
+        raise NotImplementedError("MilvusBackend 尚未实现")
+
+    async def delete(
+        self,
+        collection: str,
+        ids: list[int],
+    ) -> int:
+        raise NotImplementedError("MilvusBackend 尚未实现")

supply/infra/database/mysql.py

@@ -0,0 +1,108 @@
+"""MySQL 异步后端 —— 基于 aiomysql 连接池"""
+
+import logging
+from typing import Any, Optional
+
+from aiomysql import create_pool, DictCursor
+
+from supply.infra.database.ports import SqlBackend
+
+logger = logging.getLogger(__name__)
+
+
+class MySQLBackend(SqlBackend):
+    """MySQL 异步连接池后端"""
+
+    def __init__(self, name: str, config: Any):
+        super().__init__(name, config)
+        self._pool = None
+
+    # ==================== 生命周期 ====================
+
+    async def init(self) -> None:
+        if self._initialized:
+            return
+        self._pool = await create_pool(
+            host=self.config.host,
+            port=self.config.port,
+            user=self.config.user,
+            password=self.config.password,
+            db=self.config.db,
+            charset=getattr(self.config, "charset", "utf8mb4"),
+            minsize=getattr(self.config, "minsize", 5),
+            maxsize=getattr(self.config, "maxsize", 20),
+            cursorclass=DictCursor,
+            autocommit=True,
+        )
+        self._initialized = True
+        logger.info("MySQL backend [%s] 连接池创建成功", self.name)
+
+    async def close(self) -> None:
+        if self._pool is None:
+            return
+        self._pool.close()
+        await self._pool.wait_closed()
+        self._pool = None
+        self._initialized = False
+        logger.info("MySQL backend [%s] 连接池已关闭", self.name)
+
+    async def health(self) -> bool:
+        try:
+            await self.fetch_one("SELECT 1")
+            return True
+        except Exception:
+            return False
+
+    # ==================== SQL 操作 ====================
+
+    async def fetch(
+        self,
+        query: str,
+        params: Optional[tuple] = None,
+    ) -> list[dict]:
+        self._ensure_ready()
+        async with self._pool.acquire() as conn:
+            async with conn.cursor(DictCursor) as cursor:
+                await cursor.execute(query, params)
+                return await cursor.fetchall()
+
+    async def fetch_one(
+        self,
+        query: str,
+        params: Optional[tuple] = None,
+    ) -> Optional[dict]:
+        self._ensure_ready()
+        async with self._pool.acquire() as conn:
+            async with conn.cursor(DictCursor) as cursor:
+                await cursor.execute(query, params)
+                return await cursor.fetchone()
+
+    async def execute(
+        self,
+        query: str,
+        params: Any = None,
+        *,
+        batch: bool = False,
+    ) -> int:
+        self._ensure_ready()
+        async with self._pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                try:
+                    if batch:
+                        await cursor.executemany(query, params)
+                    else:
+                        await cursor.execute(query, params)
+                    affected = cursor.rowcount
+                    await conn.commit()
+                    return affected
+                except Exception:
+                    await conn.rollback()
+                    raise
+
+    # ==================== 内部 ====================
+
+    def _ensure_ready(self):
+        if not self._initialized or self._pool is None:
+            raise RuntimeError(
+                f"MySQL backend [{self.name}] 未初始化，请先调用 init()"
+            )

supply/infra/database/ports.py

@@ -0,0 +1,144 @@
+"""存储后端抽象接口 (Ports) —— 定义所有存储引擎的统一契约
+
+Domain 层只依赖这里的抽象，Infra 层提供具体实现。
+"""
+
+import logging
+from abc import ABC, abstractmethod
+from enum import Enum
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class BackendType(str, Enum):
+    """存储后端类型枚举"""
+
+    MYSQL = "mysql"
+    POSTGRESQL = "postgresql"
+    MILVUS = "milvus"
+    REDIS = "redis"
+
+
+class BaseBackend(ABC):
+    """所有存储后端的抽象基类"""
+
+    def __init__(self, name: str, config: Any):
+        self.name = name
+        self.config = config
+        self._initialized = False
+
+    @abstractmethod
+    async def init(self) -> None:
+        """初始化连接池或客户端，失败抛异常"""
+        ...
+
+    @abstractmethod
+    async def close(self) -> None:
+        """关闭连接池或客户端，安全可幂等"""
+        ...
+
+    @abstractmethod
+    async def health(self) -> bool:
+        """健康检查：ping 成功后返回 True"""
+        ...
+
+    @property
+    def initialized(self) -> bool:
+        return self._initialized
+
+
+class SqlBackend(BaseBackend):
+    """SQL 类后端的公共抽象（MySQL / PostgreSQL）"""
+
+    @abstractmethod
+    async def fetch(
+        self,
+        query: str,
+        params: Optional[tuple] = None,
+    ) -> list[dict]:
+        """SELECT 多行"""
+        ...
+
+    @abstractmethod
+    async def fetch_one(
+        self,
+        query: str,
+        params: Optional[tuple] = None,
+    ) -> Optional[dict]:
+        """SELECT 单行"""
+        ...
+
+    @abstractmethod
+    async def execute(
+        self,
+        query: str,
+        params: Any = None,
+        *,
+        batch: bool = False,
+    ) -> int:
+        """INSERT/UPDATE/DELETE，返回受影响行数"""
+        ...
+
+
+class VectorBackend(BaseBackend):
+    """向量数据库的公共抽象（Milvus 等）"""
+
+    @abstractmethod
+    async def insert(
+        self,
+        collection: str,
+        vectors: list[list[float]],
+        metadata: Optional[list[dict]] = None,
+    ) -> list[int]:
+        """插入向量，返回主键 ID 列表"""
+        ...
+
+    @abstractmethod
+    async def search(
+        self,
+        collection: str,
+        query_vectors: list[list[float]],
+        top_k: int = 10,
+        filter_expr: Optional[str] = None,
+    ) -> list[list[dict]]:
+        """向量相似搜索"""
+        ...
+
+    @abstractmethod
+    async def delete(
+        self,
+        collection: str,
+        ids: list[int],
+    ) -> int:
+        """按主键删除向量"""
+        ...
+
+
+class KvBackend(BaseBackend):
+    """KV 存储的公共抽象（Redis 等）"""
+
+    @abstractmethod
+    async def get(self, key: str) -> Optional[str]:
+        """获取键值"""
+        ...
+
+    @abstractmethod
+    async def set(
+        self,
+        key: str,
+        value: str,
+        ttl: Optional[int] = None,
+    ) -> bool:
+        """写入键值，可选 TTL（秒）"""
+        ...
+
+    @abstractmethod
+    async def delete(self, *keys: str) -> int:
+        """删除键，返回已删除数量"""
+        ...
+
+    @abstractmethod
+    async def exists(self, *keys: str) -> int:
+        """检查键是否存在，返回存在的数量"""
+        ...

supply/infra/database/postgresql.py

@@ -0,0 +1,59 @@
+"""PostgreSQL 异步后端 —— 基于 asyncpg 连接池
+
+状态：未实现。init() 会抛出 NotImplementedError。
+"""
+
+import logging
+from typing import Any, Optional
+
+from supply.infra.database.ports import SqlBackend
+
+logger = logging.getLogger(__name__)
+
+
+class PgBackend(SqlBackend):
+    """PostgreSQL 异步连接池后端"""
+
+    def __init__(self, name: str, config: Any):
+        super().__init__(name, config)
+        self._pool = None
+
+    # ==================== 生命周期 ====================
+
+    async def init(self) -> None:
+        raise NotImplementedError(
+            "PgBackend 尚未实现。需要安装 asyncpg 并完成连接池初始化逻辑。"
+        )
+
+    async def close(self) -> None:
+        if self._pool is not None:
+            self._pool = None
+        self._initialized = False
+
+    async def health(self) -> bool:
+        return False
+
+    # ==================== SQL 操作 ====================
+
+    async def fetch(
+        self,
+        query: str,
+        params: Optional[tuple] = None,
+    ) -> list[dict]:
+        raise NotImplementedError("PgBackend 尚未实现")
+
+    async def fetch_one(
+        self,
+        query: str,
+        params: Optional[tuple] = None,
+    ) -> Optional[dict]:
+        raise NotImplementedError("PgBackend 尚未实现")
+
+    async def execute(
+        self,
+        query: str,
+        params: Any = None,
+        *,
+        batch: bool = False,
+    ) -> int:
+        raise NotImplementedError("PgBackend 尚未实现")

supply/infra/database/redis.py

@@ -0,0 +1,63 @@
+"""Redis 异步后端 —— 基于 redis-py (aioredis)
+
+状态：未实现。init() 会抛出 NotImplementedError。
+"""
+
+import logging
+from typing import Any, Optional
+
+from supply.infra.database.ports import KvBackend
+
+logger = logging.getLogger(__name__)
+
+
+class RedisBackend(KvBackend):
+    """Redis 异步 KV 后端"""
+
+    def __init__(self, name: str, config: Any):
+        super().__init__(name, config)
+        self._client = None
+
+    # ==================== 生命周期 ====================
+
+    async def init(self) -> None:
+        raise NotImplementedError(
+            "RedisBackend 尚未实现。需要安装 redis-py 并完成异步连接初始化逻辑。"
+        )
+
+    async def close(self) -> None:
+        if self._client is not None:
+            self._client = None
+        self._initialized = False
+
+    async def health(self) -> bool:
+        return False
+
+    # ==================== KV 操作 ====================
+
+    async def get(self, key: str) -> Optional[str]:
+        raise NotImplementedError("RedisBackend 尚未实现")
+
+    async def set(
+        self,
+        key: str,
+        value: str,
+        ttl: Optional[int] = None,
+    ) -> bool:
+        raise NotImplementedError("RedisBackend 尚未实现")
+
+    async def delete(self, *keys: str) -> int:
+        raise NotImplementedError("RedisBackend 尚未实现")
+
+    async def exists(self, *keys: str) -> int:
+        raise NotImplementedError("RedisBackend 尚未实现")
+
+    # ==================== 扩展操作 ====================
+
+    async def expire(self, key: str, ttl: int) -> bool:
+        """设置键过期时间（秒）"""
+        raise NotImplementedError("RedisBackend 尚未实现")
+
+    async def ttl(self, key: str) -> int:
+        """获取键剩余 TTL（秒），-1 永不过期，-2 不存在"""
+        raise NotImplementedError("RedisBackend 尚未实现")

supply/infra/observability/__init__.py

@@ -0,0 +1,18 @@
+"""可观测性设施层 —— 结构化日志模型、SLS 索引定义、日志服务"""
+
+from supply.infra.observability.log_schema import (
+    LogCategory,
+    LogLevel,
+    LogStatus,
+    LogRecord,
+    SLS_INDEX_CONFIG,
+)
+from supply.infra.observability.logging import LogService
+
+__all__ = [
+    "LogCategory",
+    "LogLevel",
+    "LogRecord",
+    "SLS_INDEX_CONFIG",
+    "LogService",
+]

supply/infra/observability/log_schema.py

@@ -0,0 +1,216 @@
+"""结构化日志模型 + SLS 索引配置
+
+设计目标：
+  1. 统一日志字段，消除当前 free-form dict 导致的查询困难
+  2. 按查询模式设计 SLS 键值索引，避免全表扫描
+  3. 支持 trace_id 串联完整调用链
+
+
+====== SLS 索引配置说明 ======
+
+全文索引（分词查询）:
+  字段              分词符              包含中文
+  ───────────────────────────────────────────
+  message            ,;=|'"\t\r\n         ✓
+  error_message      ,;=|'"\t\r\n         ✓
+
+键值索引（精确匹配 / 范围 / 聚合）:
+  字段              类型        别名              说明
+  ────────────────────────────────────────────────────
+  level             text        log_level         日志级别，精确匹配 + 统计
+  category          text        log_category      日志分类，精确匹配
+  event             text                          事件名，精确匹配 + 前缀查询
+  trace_id          text                          分布式追踪 ID，精确匹配
+  task_name         text                          任务名，精确匹配 + 前缀查询
+  agent_name        text                          Agent 名，精确匹配
+  tool_name         text                          Tool 名，精确匹配
+  duration_ms       long                          耗时（毫秒），范围查询 + 统计
+  status            text                          执行状态，精确匹配
+  error_type        text                          异常类型，精确匹配 + 聚合
+  extra             json                          扩展字段，子字段可索引
+
+统计优化:
+  duration_ms 开启统计（min / max / avg / sum / distinct_count）
+  level 开启统计（distinct_count）
+
+控制台创建 SQL 建索引（参考，实际通过控制台或 API 创建）:
+  -- 全文索引
+  {"Line": {"Token": [",;=|'\"\t\r\n"]}, "Keys": {"message": {"Token": [",;=|'\"\t\r\n"], "CaseSensitive": false, "Chn": true}, "error_message": {"Token": [",;=|'\"\t\r\n"], "CaseSensitive": false, "Chn": true}}}
+
+  -- 键值索引
+  Key=level          Type=text
+  Key=category       Type=text
+  Key=event          Type=text
+  Key=trace_id       Type=text
+  Key=task_name      Type=text
+  Key=agent_name     Type=text
+  Key=tool_name      Type=text
+  Key=duration_ms    Type=long  Stats=true
+  Key=status         Type=text
+  Key=error_type     Type=text
+  Key=extra          Type=json
+
+
+====== 典型查询场景 ======
+
+1. 追踪某次任务全链路:
+   trace_id:"Task-20260630143000-abc123" | SELECT * ORDER BY __time__
+
+2. 统计各 Agent 执行耗时 P50/P99:
+   agent_name:* | SELECT agent_name, approx_percentile(duration_ms, 0.5) AS p50, approx_percentile(duration_ms, 0.99) AS p99 GROUP BY agent_name
+
+3. 告警：最近 5 分钟 ERROR 数量:
+   level:ERROR AND __time__ > now() - 300
+
+4. 某任务最近 1 小时成功率:
+   task_name:"content_generation" AND __time__ > now() - 3600 | SELECT status, count(*) GROUP BY status
+
+5. 按异常类型聚合排障:
+   level:ERROR AND __time__ > now() - 3600 | SELECT error_type, count(*) AS cnt GROUP BY error_type ORDER BY cnt DESC
+
+6. 慢任务检测（耗时 > 阈值）:
+   category:task AND duration_ms > 300000 | SELECT task_name, trace_id, duration_ms ORDER BY duration_ms DESC
+"""
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, Dict, Optional
+
+from pydantic import BaseModel, Field
+
+
+# ==================== 枚举 ====================
+
+
+class LogLevel(str, Enum):
+    """日志级别"""
+
+    DEBUG = "DEBUG"
+    INFO = "INFO"
+    WARN = "WARN"
+    ERROR = "ERROR"
+    FATAL = "FATAL"
+
+
+class LogCategory(str, Enum):
+    """日志分类 —— 按系统边界划分，用于快速过滤"""
+
+    SYSTEM = "system"    # 启动、关闭、心跳
+    TASK = "task"        # 任务调度生命周期
+    AGENT = "agent"      # Agent 执行（规划、推理、反思）
+    TOOL = "tool"        # Tool 调用（HTTP、DB 查询等）
+    HTTP = "http"        # 入站 HTTP 请求
+    DB = "db"            # 数据库操作
+    SECURITY = "security"  # 安全相关（认证、鉴权、告警）
+
+
+class LogStatus(str, Enum):
+    """执行状态"""
+
+    SUCCESS = "success"
+    FAILED = "failed"
+    TIMEOUT = "timeout"
+    CANCELLED = "cancelled"
+    PENDING = "pending"
+
+
+# ==================== 结构化日志模型 ====================
+
+
+class LogRecord(BaseModel):
+    """统一日志记录 —— 所有日志推送都必须使用此模型
+
+    使用方式:
+        record = LogRecord(
+            level=LogLevel.INFO,
+            category=LogCategory.TASK,
+            event="task_started",
+            message="内容生成任务开始执行",
+            trace_id="Task-20260630143000-abc123",
+            task_name="content_generation",
+            extra={"topic": "AI 安全", "word_count": 5000},
+        )
+        await log_service.log(record)
+    """
+
+    # ── 核心字段 ──
+    level: LogLevel = Field(default=LogLevel.INFO, description="日志级别")
+    category: LogCategory = Field(default=LogCategory.SYSTEM, description="日志分类")
+    event: str = Field(default="", description="事件名，如 task_started / agent_think / tool_error")
+    message: str = Field(default="", description="人类可读描述，用于全文检索")
+
+    # ── 追踪字段 ──
+    trace_id: str = Field(default="", description="分布式追踪 ID，串联完整调用链")
+    span_id: str = Field(default="", description="当前 span ID，可选")
+
+    # ── 业务标识 ──
+    task_name: str = Field(default="", description="任务名，如 content_generation")
+    agent_name: str = Field(default="", description="Agent 名，如 ContentWriter")
+    tool_name: str = Field(default="", description="Tool 名，如 WebSearchTool")
+
+    # ── 性能字段 ──
+    duration_ms: int = Field(default=0, description="执行耗时（毫秒），用于 P50/P99 分析")
+
+    # ── 状态与异常 ──
+    status: LogStatus = Field(default=LogStatus.SUCCESS, description="执行状态")
+    error_type: str = Field(default="", description="异常类名，如 TimeoutError")
+    error_message: str = Field(default="", description="异常消息全文，用于全文检索")
+
+    # ── 扩展字段 ──
+    extra: Dict[str, Any] = Field(default_factory=dict, description="业务自定义 KV，JSON 索引可查询子字段")
+
+    def to_sls_items(self) -> list[tuple[str, str]]:
+        """转为阿里云 SLS LogItem 的 contents 格式 [(key, value), ...]
+
+        SLS API 要求所有值都是 string，此方法负责序列化。
+        """
+        return [
+            ("level", self.level.value),
+            ("category", self.category.value),
+            ("event", self.event),
+            ("message", self.message),
+            ("trace_id", self.trace_id),
+            ("span_id", self.span_id),
+            ("task_name", self.task_name),
+            ("agent_name", self.agent_name),
+            ("tool_name", self.tool_name),
+            ("duration_ms", str(self.duration_ms)),
+            ("status", self.status.value),
+            ("error_type", self.error_type),
+            ("error_message", self.error_message),
+            # extra 存 JSON 字符串，SLS json 类型索引可解析
+            ("extra", json.dumps(self.extra, ensure_ascii=False) if self.extra else "{}"),
+        ]
+
+
+# ==================== SLS 索引配置（文档 / 参考） ====================
+
+
+SLS_INDEX_CONFIG = {
+    "logstore": "long_articles_agentic_aupply",
+    "ttl_days": 30,
+    "full_text_index": {
+        "case_sensitive": False,
+        "include_chinese": True,
+        "delimiter": ",;=|'\"\t\r\n ",
+        "fields": ["message", "error_message"],
+    },
+    "key_value_index": [
+        {"key": "level", "type": "text", "alias": "log_level", "stats": True},
+        {"key": "category", "type": "text", "alias": "log_category"},
+        {"key": "event", "type": "text"},
+        {"key": "trace_id", "type": "text"},
+        {"key": "task_name", "type": "text"},
+        {"key": "agent_name", "type": "text"},
+        {"key": "tool_name", "type": "text"},
+        {"key": "duration_ms", "type": "long", "stats": True},
+        {"key": "status", "type": "text"},
+        {"key": "error_type", "type": "text"},
+        {"key": "extra", "type": "json"},
+    ],
+}
+
+
+# ── 惰性 json import（避免循环） ──
+import json  # noqa: E402

supply/infra/observability/logging.py

@@ -0,0 +1,165 @@
+"""异步日志服务 —— 基于 Queue + 后台 worker 推送到阿里云 SLS
+
+职责：
+  - 接收结构化 LogRecord，序列化后批量推送
+  - 提供 log_raw() 向下兼容旧的 dict 调用方式
+  - Queue 满时静默丢弃，不阻塞业务
+"""
+
+import asyncio
+import contextlib
+import datetime
+import json
+import logging
+import time
+import traceback
+from typing import Optional
+
+from aliyun.log import LogClient, LogItem, PutLogsRequest
+
+from supply.config.aliyun import AliyunLogConfig
+from supply.infra.observability.log_schema import LogRecord
+
+logger = logging.getLogger(__name__)
+
+
+class LogService:
+    """异步日志服务：将结构化日志放入 Queue，后台 worker 批量推送到 SLS
+
+    使用方式：
+        service = LogService(AliyunLogConfig())
+        await service.start()
+
+        # 结构化方式（推荐）
+        record = LogRecord(
+            level=LogLevel.INFO,
+            category=LogCategory.TASK,
+            event="task_started",
+            trace_id="trace-123",
+            task_name="content_generation",
+        )
+        await service.log(record)
+
+        # 向后兼容：自由 dict
+        await service.log_raw({"event": "legacy_call", "key": "val"})
+
+        await service.stop()
+    """
+
+    def __init__(self, log_config: AliyunLogConfig):
+        self.config = log_config
+        self.client: Optional[LogClient] = None
+        self.queue: Optional[asyncio.Queue] = None
+        self._worker_task: Optional[asyncio.Task] = None
+        self._running = False
+
+    # ==================== 生命周期 ====================
+
+    async def start(self):
+        if self._running:
+            return
+        self.client = LogClient(
+            self.config.endpoint,
+            self.config.access_key_id,
+            self.config.access_key_secret,
+        )
+        self.queue = asyncio.Queue(maxsize=10000)
+        self._running = True
+        self._worker_task = asyncio.create_task(self._worker())
+        logger.info("LogService 已启动 → %s/%s", self.config.project, self.config.logstore)
+
+    async def stop(self):
+        if not self._running:
+            return
+        self._running = False
+        if self._worker_task:
+            self._worker_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._worker_task
+        self._worker_task = None
+        self.queue = None
+        self.client = None
+        logger.info("LogService 已停止")
+
+    # ==================== 对外接口 ====================
+
+    async def log(self, record: LogRecord) -> None:
+        """推送结构化日志（推荐方式，非阻塞）"""
+        if not self._running or self.queue is None:
+            return
+        try:
+            self.queue.put_nowait(record)
+        except asyncio.QueueFull:
+            pass
+
+    async def log_raw(self, contents: dict) -> None:
+        """向后兼容：接受自由 dict，内部转为 LogRecord 推送
+
+        所有值会被 str() 化以匹配 SLS API 要求。
+        """
+        if not self._running or self.queue is None:
+            return
+        try:
+            # 从 dict 提取已知字段，其余塞入 extra
+            level_str = str(contents.pop("level", "INFO")).upper()
+            category_str = str(contents.pop("category", "system")).lower()
+            try:
+                from supply.infra.observability.log_schema import LogLevel, LogCategory
+
+                level = LogLevel(level_str) if level_str in LogLevel._value2member_map_ else LogLevel.INFO
+                category = LogCategory(category_str) if category_str in LogCategory._value2member_map_ else LogCategory.SYSTEM
+            except (KeyError, ValueError):
+                level = LogLevel.INFO
+                category = LogCategory.SYSTEM
+
+            record = LogRecord(
+                level=level,
+                category=category,
+                event=str(contents.pop("event", "")),
+                message=str(contents.pop("message", "")),
+                trace_id=str(contents.pop("trace_id", "")),
+                task_name=str(contents.pop("task_name", "")),
+                agent_name=str(contents.pop("agent_name", "")),
+                tool_name=str(contents.pop("tool_name", "")),
+                duration_ms=int(contents.pop("duration_ms", 0)),
+                status=str(contents.pop("status", "success")),
+                error_type=str(contents.pop("error_type", "")),
+                error_message=str(contents.pop("error_message", "")),
+                extra=contents,  # 剩余字段
+            )
+            self.queue.put_nowait(record)
+        except asyncio.QueueFull:
+            pass
+
+    # ==================== 后台 Worker ====================
+
+    async def _worker(self):
+        try:
+            while self._running:
+                record = await self.queue.get()
+                try:
+                    await asyncio.to_thread(self._put_log, record)
+                except Exception as e:
+                    print(f"[Log Error] {e}")
+                    print(traceback.format_exc())
+        except asyncio.CancelledError:
+            pass
+
+    def _put_log(self, record: LogRecord):
+        """同步推送到 SLS"""
+        timestamp = int(time.time())
+
+        # LogRecord → SLS LogItem
+        items = record.to_sls_items()
+        # 追加内置字段
+        items.append(("datetime", datetime.datetime.now().isoformat()))
+
+        log_item = LogItem(timestamp=timestamp, contents=items)
+        req = PutLogsRequest(
+            self.config.project,
+            self.config.logstore,
+            topic=record.category.value,     # topic 用 category，方便消费端路由
+            source="",
+            logitems=[log_item],
+        )
+        self.client.put_logs(req)

supply/infra/shared/__init__.py

@@ -0,0 +1,14 @@
+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,
+)
+from .tools import (
+    safe_json_parse,
+    str_to_md5,
+    generate_task_trace_id,
+    yield_batch,
+    timestamp_to_str,
+    days_remaining_in_month,
+)

supply/infra/shared/async_tasks.py

@@ -0,0 +1,179 @@
+import asyncio
+from typing import Callable, Coroutine, List, Any, Dict
+
+
+async def run_tasks_with_async_worker_group(
+    task_list: List[Any],
+    handler: Callable[[Any], Coroutine[Any, Any, None]],
+    *,
+    description: str = None,
+    unit: str = "item",
+    max_concurrency: int = 20,
+    fail_fast: bool = False,
+    show_progress: bool = True,
+) -> Dict[str, Any]:
+    """使用异步 worker pool 处理 I/O 密集型任务
+
+    Args:
+        task_list: 任务对象列表
+        handler: 异步处理函数，接受单个任务对象
+        description: 进度条描述
+        unit: 进度条单位
+        max_concurrency: 最大并发 worker 数
+        fail_fast: True 时任一 worker 异常立即中止全部
+        show_progress: 是否显示 tqdm 进度条（需要安装 tqdm）
+
+    Returns:
+        {"total_task": int, "processed_task": int, "errors": [(idx, obj, exc), ...]}
+    """
+    if not task_list:
+        return {"total_task": 0, "processed_task": 0, "errors": []}
+
+    total_task = len(task_list)
+    processed_task = 0
+    errors: List[tuple] = []
+
+    queue: asyncio.Queue = asyncio.Queue()
+    cancel_event = asyncio.Event()
+    counter_lock = asyncio.Lock()
+
+    # 进度条（可选）
+    if show_progress:
+        try:
+            from tqdm.asyncio import tqdm as async_tqdm
+            pbar = async_tqdm(total=total_task, unit=unit, desc=description)
+        except ImportError:
+            pbar = None
+    else:
+        pbar = None
+
+    async def worker(worker_id: int):
+        nonlocal processed_task
+        while True:
+            item = await queue.get()
+            if item is None:
+                queue.task_done()
+                break
+
+            idx, task_obj = item
+            try:
+                if cancel_event.is_set():
+                    return
+
+                await handler(task_obj)
+                async with counter_lock:
+                    processed_task += 1
+
+            except Exception as e:
+                if fail_fast:
+                    cancel_event.set()
+                    raise
+                errors.append((idx, task_obj, e))
+
+            finally:
+                if pbar:
+                    pbar.update()
+                queue.task_done()
+
+    workers = [asyncio.create_task(worker(i)) for i in range(max_concurrency)]
+
+    try:
+        for index, task in enumerate(task_list, start=1):
+            await queue.put((index, task))
+
+        await queue.join()
+
+    except Exception:
+        for w in workers:
+            w.cancel()
+        raise
+
+    finally:
+        for _ in range(max_concurrency):
+            await queue.put(None)
+
+        await asyncio.gather(*workers, return_exceptions=True)
+        if pbar:
+            pbar.close()
+
+    return {
+        "total_task": total_task,
+        "processed_task": processed_task,
+        "errors": errors,
+    }
+
+
+async def run_tasks_with_asyncio_task_group(
+    task_list: List[Any],
+    handler: Callable[[Any], Coroutine[Any, Any, None]],
+    *,
+    description: str = None,
+    unit: str = "item",
+    max_concurrency: int = 20,
+    fail_fast: bool = False,
+    show_progress: bool = True,
+) -> Dict[str, Any]:
+    """使用 asyncio.TaskGroup 处理 I/O 密集型任务
+
+    Args:
+        task_list: 任务对象列表
+        handler: 异步处理函数，接受单个任务对象
+        description: 进度条描述
+        unit: 进度条单位
+        max_concurrency: 最大并发数（通过 Semaphore 控制）
+        fail_fast: True 时任一任务异常立即中止全部
+        show_progress: 是否显示 tqdm 进度条
+
+    Returns:
+        {"total_task": int, "processed_task": int, "errors": [(idx, obj, exc), ...]}
+    """
+    if not task_list:
+        return {"total_task": 0, "processed_task": 0, "errors": []}
+
+    processed_task = 0
+    total_task = len(task_list)
+    errors: List[tuple] = []
+    semaphore = asyncio.Semaphore(max_concurrency)
+
+    # 进度条（可选）
+    if show_progress:
+        try:
+            from tqdm.asyncio import tqdm as async_tqdm
+            pbar = async_tqdm(total=total_task, unit=unit, desc=description)
+        except ImportError:
+            pbar = None
+    else:
+        pbar = None
+
+    lock = asyncio.Lock()
+
+    async def _run_single_task(task_obj: Any, idx: int):
+        nonlocal processed_task
+        async with semaphore:
+            try:
+                await handler(task_obj)
+                async with lock:
+                    processed_task += 1
+            except Exception as e:
+                if fail_fast:
+                    raise e
+                errors.append((idx, task_obj, e))
+            finally:
+                if pbar:
+                    pbar.update()
+
+    async with asyncio.TaskGroup() as task_group:
+        for index, task in enumerate(task_list, start=1):
+            task_group.create_task(
+                _run_single_task(task, index),
+                name=f"processing {description}-{index}",
+            )
+
+    if pbar:
+        pbar.close()
+
+    return {
+        "total_task": total_task,
+        "processed_task": processed_task,
+        "errors": errors,
+    }

supply/infra/shared/http_client.py

@@ -0,0 +1,92 @@
+import aiohttp
+from typing import Optional, Union, Dict, Any
+
+
+class AsyncHttpClient:
+    """异步 HTTP 客户端，基于 aiohttp，支持连接池和超时配置"""
+
+    def __init__(
+        self,
+        timeout: int = 10,
+        max_connections: int = 100,
+        default_headers: Optional[Dict[str, str]] = None,
+    ):
+        self.timeout = aiohttp.ClientTimeout(total=timeout)
+        self.connector = aiohttp.TCPConnector(limit=max_connections)
+        self.default_headers = default_headers or {}
+        self.session = None
+
+    async def __aenter__(self):
+        self.session = aiohttp.ClientSession(
+            connector=self.connector,
+            timeout=self.timeout,
+            headers=self.default_headers,
+        )
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.session.close()
+
+    async def request(
+        self,
+        method: str,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        data: Optional[Union[Dict[str, Any], str, bytes]] = None,
+        json: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Dict[str, Any], str]:
+        """核心请求方法"""
+        request_headers = {**self.default_headers, **(headers or {})}
+
+        try:
+            async with self.session.request(
+                method,
+                url,
+                params=params,
+                data=data,
+                json=json,
+                headers=request_headers,
+            ) as response:
+                response.raise_for_status()
+                content_type = response.headers.get("Content-Type", "")
+
+                if "application/json" in content_type:
+                    return await response.json()
+                return await response.text()
+
+        except aiohttp.ClientResponseError as e:
+            print(f"HTTP error: {e.status} {e.message}")
+            raise
+        except aiohttp.ClientError as e:
+            print(f"Network error: {str(e)}")
+            raise
+
+    async def get(
+        self,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Dict[str, Any], str]:
+        """GET 请求"""
+        return await self.request("GET", url, params=params, headers=headers)
+
+    async def post(
+        self,
+        url: str,
+        data: Optional[Union[Dict[str, Any], str, bytes]] = None,
+        json: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Dict[str, Any], str]:
+        """POST 请求"""
+        return await self.request("POST", url, data=data, json=json, headers=headers)
+
+    async def put(
+        self,
+        url: str,
+        data: Optional[Union[Dict[str, Any], str, bytes]] = None,
+        json: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Dict[str, Any], str]:
+        """PUT 请求"""
+        return await self.request("PUT", url, data=data, json=json, headers=headers)

supply/infra/shared/response.py

@@ -0,0 +1,18 @@
+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/shared/tools.py

@@ -0,0 +1,120 @@
+"""通用工具函数 —— 只包含与业务无关的基础设施工具"""
+
+import hashlib
+import json
+import random
+import re
+import string
+from datetime import datetime, timezone, date, timedelta
+from typing import Dict, List, Optional
+
+
+def safe_json_parse(text: str) -> Optional[Dict | List]:
+    """多层降级解析 JSON：直接解析 → 提取代码块 → 提取 JSON 对象/数组
+
+    LLM 可能返回 ```json ... ``` 包裹的文本，或文本中夹杂 markdown 前缀/后缀。
+    先尝试直接解析（最常见路径），失败后逐层降级提取。
+    """
+    if not text:
+        return None
+
+    # 降级 1：直接解析
+    try:
+        return json.loads(text)
+    except (json.JSONDecodeError, TypeError):
+        pass
+
+    clean = text.strip()
+
+    # 降级 2：提取最外层 json 代码块
+    m = re.search(r"```json\s*(.*?)\s*```", clean, re.DOTALL)
+    if m:
+        try:
+            return json.loads(m.group(1))
+        except (json.JSONDecodeError, TypeError):
+            pass
+    else:
+        m = re.search(r"```\s*(.*?)\s*```", clean, re.DOTALL)
+        if m:
+            try:
+                return json.loads(m.group(1))
+            except (json.JSONDecodeError, TypeError):
+                pass
+
+    # 降级 3：在文本中查找第一个完整 JSON 对象 { ... } 或数组 [ ... ]
+    for bracket_pair in [("{}", "{", "}"), ("[]", "[", "]")]:
+        opener, closer = bracket_pair[1], bracket_pair[2]
+        start = clean.find(opener)
+        if start == -1:
+            continue
+        depth = 0
+        in_string = False
+        escape_next = False
+        for i in range(start, len(clean)):
+            ch = clean[i]
+            if escape_next:
+                escape_next = False
+                continue
+            if ch == "\\":
+                escape_next = True
+                continue
+            if ch == '"' and not escape_next:
+                in_string = not in_string
+                continue
+            if in_string:
+                continue
+            if ch == opener:
+                depth += 1
+            elif ch == closer:
+                depth -= 1
+                if depth == 0:
+                    try:
+                        return json.loads(clean[start : i + 1])
+                    except (json.JSONDecodeError, TypeError):
+                        return None
+        try:
+            return json.loads(clean[start:])
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+    return None
+
+
+def str_to_md5(strings: str) -> str:
+    """字符串转 MD5"""
+    original_bytes = strings.encode("utf-8")
+    md5_hash = hashlib.md5()
+    md5_hash.update(original_bytes)
+    return md5_hash.hexdigest()
+
+
+def generate_task_trace_id() -> str:
+    """生成任务追踪 ID，格式: Task-YYYYMMDDHHmmss-16位随机"""
+    random_str = "".join(random.choices(string.ascii_lowercase + string.digits, k=16))
+    return f"Task-{datetime.now().strftime('%Y%m%d%H%M%S')}-{random_str}"
+
+
+def yield_batch(data: list, batch_size: int):
+    """生成批次数据"""
+    for i in range(0, len(data), batch_size):
+        yield data[i : i + batch_size]
+
+
+def timestamp_to_str(timestamp: int, string_format: str = "%Y-%m-%d %H:%M:%S") -> str:
+    """时间戳转字符串"""
+    dt_object = (
+        datetime.utcfromtimestamp(timestamp).replace(tzinfo=timezone.utc).astimezone()
+    )
+    return dt_object.strftime(string_format)
+
+
+def days_remaining_in_month() -> int:
+    """计算本月剩余天数"""
+    today = date.today()
+    if today.month == 12:
+        next_month = today.replace(year=today.year + 1, month=1, day=1)
+    else:
+        next_month = today.replace(month=today.month + 1, day=1)
+    last_day_of_month = next_month - timedelta(days=1)
+    return (last_day_of_month - today).days
+

supply/jobs/__init__.py

@@ -0,0 +1,32 @@
+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

@@ -0,0 +1,12 @@
+"""向后兼容 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

@@ -0,0 +1,15 @@
+"""向后兼容 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

@@ -0,0 +1,41 @@
+"""向后兼容 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/test_handler.py

@@ -0,0 +1,15 @@
+"""XXL-JOB 测试 Handler —— 验证 Admin → Executor 通信链路"""
+
+import logging
+
+from supply.jobs.xxl_decorator import xxl_job
+
+logger = logging.getLogger(__name__)
+
+
+@xxl_job("testHelloWorld")
+async def test_hello_world(param: str) -> dict:
+    """测试 handler：打印 hello world"""
+    logger.info("testHelloWorld triggered, param=%s", param)
+    print(f"hello world! param: {param}")
+    return {"code": 200, "msg": "hello world printed"}

supply/jobs/xxl_config.py

@@ -0,0 +1,45 @@
+"""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/jobs/xxl_decorator.py

@@ -0,0 +1,56 @@
+"""XXL-JOB 任务注册装饰器 —— 对标 Java @XxlJob 注解
+
+使用方式：
+    from supply.jobs.xxl_decorator import xxl_job, xxl_registry
+
+    @xxl_job("myJobHandler")
+    async def my_job(param: str) -> dict:
+        ...
+        return {"code": 200, "msg": "success"}
+"""
+
+from typing import Callable, Awaitable, Dict, Optional
+
+# XXL-JOB handler 注册表：{handler_name: handler_func}
+_XXL_REGISTRY: Dict[str, Callable[[str], Awaitable[dict]]] = {}
+
+
+def xxl_job(handler_name: str):
+    """装饰器：将异步函数注册为 XXL-JOB handler
+
+    对标 Java 的 @XxlJob("handlerName") 注解。
+    被装饰函数签名：async def handler(param: str) -> dict
+    返回值格式：{"code": 200, "msg": "success"} 或 {"code": 500, "msg": "error"}
+
+    Args:
+        handler_name: XXL-JOB Admin 中配置的「执行器处理器」名称
+    """
+
+    def decorator(func: Callable[[str], Awaitable[dict]]):
+        _XXL_REGISTRY[handler_name] = func
+        return func
+
+    return decorator
+
+
+class XxlRegistry:
+    """XXL-JOB 注册表只读视图"""
+
+    @staticmethod
+    def get_handler(handler_name: str) -> Optional[Callable[[str], Awaitable[dict]]]:
+        """获取已注册的 handler"""
+        return _XXL_REGISTRY.get(handler_name)
+
+    @staticmethod
+    def list_handlers() -> list:
+        """列出所有已注册的 handler 名称"""
+        return list(_XXL_REGISTRY.keys())
+
+    @staticmethod
+    def is_registered(handler_name: str) -> bool:
+        """检查 handler 是否已注册"""
+        return handler_name in _XXL_REGISTRY
+
+
+# 暴露注册表给 executor 使用
+xxl_registry = XxlRegistry()

supply/jobs/xxl_executor.py

@@ -0,0 +1,285 @@
+"""XXL-JOB 执行器核心 —— 对标 Java XxlJobSpringExecutor
+
+负责：
+1. 启动时向 Admin 注册执行器
+2. 定期发送心跳保持注册
+3. Admin 调度的 run/kill/beat/idleBeat/log 请求分发（由 xxl_endpoints.py 接收后委托至此）
+"""
+
+import asyncio
+import logging
+import os
+import socket
+from typing import Any, Dict, Optional
+
+import aiohttp
+
+from supply.jobs.xxl_config import XxlJobConfig
+from supply.jobs.xxl_decorator import xxl_registry
+
+logger = logging.getLogger(__name__)
+
+# Admin API 路径常量
+_PATH_REGISTRY = "/api/registry"
+_PATH_REGISTRY_REMOVE = "/api/registryRemove"
+
+
+class XxlJobExecutor:
+    """XXL-JOB 执行器
+
+    使用方式：
+        executor = XxlJobExecutor(config)
+        await executor.start()
+        # ... 应用运行中，endpoints 调用 executor.run(...) ...
+        await executor.stop()
+    """
+
+    def __init__(self, config: XxlJobConfig):
+        self.config = config
+        self._session: Optional[aiohttp.ClientSession] = None
+        self._registry_task: Optional[asyncio.Task] = None
+        self._running = False
+        # 记录正在执行的任务 jobId → asyncio.Task
+        self._running_jobs: Dict[int, asyncio.Task] = {}
+
+    # ==================== 生命周期 ====================
+
+    async def start(self) -> None:
+        """启动执行器：初始化 HTTP session → 注册到 Admin → 启动心跳"""
+        if self._running:
+            return
+        self._running = True
+        self._session = aiohttp.ClientSession(
+            timeout=aiohttp.ClientTimeout(total=10),
+        )
+
+        # 注册到 Admin
+        await self._register()
+        # 启动定期心跳/续约
+        self._registry_task = asyncio.create_task(self._heartbeat_loop())
+        logger.info(
+            "XXL-JOB Executor started: app=%s port=%s admin=%s",
+            self.config.app_name,
+            self.config.port,
+            self.config.admin_addresses,
+        )
+
+    async def stop(self) -> None:
+        """停止执行器：取消心跳 → 从 Admin 注销 → 关闭 session"""
+        self._running = False
+
+        if self._registry_task:
+            self._registry_task.cancel()
+            try:
+                await self._registry_task
+            except asyncio.CancelledError:
+                pass
+
+        # 注销
+        await self._unregister()
+
+        if self._session:
+            await self._session.close()
+            self._session = None
+
+        logger.info("XXL-JOB Executor stopped: app=%s", self.config.app_name)
+
+    # ==================== 注册 / 心跳 ====================
+
+    def _registry_body(self) -> dict:
+        """构造注册请求体"""
+        address = self.config.address or f"http://{self._local_ip()}:{self.config.port}"
+        return {
+            "registryGroup": "EXECUTOR",
+            "registryKey": self.config.app_name,
+            "registryValue": address,
+        }
+
+    async def _register(self) -> bool:
+        """向 Admin 注册"""
+        body = self._registry_body()
+        for admin_url in self._admin_urls():
+            try:
+                async with self._session.post(
+                    f"{admin_url}{_PATH_REGISTRY}",
+                    json=body,
+                    headers=self._auth_headers(),
+                ) as resp:
+                    if resp.status == 200:
+                        data = await resp.json()
+                        if data.get("code") == 200:
+                            logger.info("XXL-JOB registered to %s", admin_url)
+                            return True
+            except Exception as e:
+                logger.warning("XXL-JOB register to %s failed: %s", admin_url, e)
+        return False
+
+    async def _unregister(self) -> None:
+        """从 Admin 注销"""
+        body = self._registry_body()
+        for admin_url in self._admin_urls():
+            try:
+                async with self._session.post(
+                    f"{admin_url}{_PATH_REGISTRY_REMOVE}",
+                    json=body,
+                    headers=self._auth_headers(),
+                ) as resp:
+                    pass
+            except Exception:
+                pass
+
+    async def _heartbeat_loop(self) -> None:
+        """定期心跳：每隔 registry_interval 秒重新注册（续约）"""
+        while self._running:
+            await asyncio.sleep(self.config.registry_interval)
+            if not self._running:
+                break
+            await self._register()
+
+    # ==================== Admin → Executor 调度方法 ====================
+
+    def beat(self) -> dict:
+        """心跳检测 —— Admin 调用 POST /beat"""
+        return {"code": 200, "msg": None}
+
+    def idle_beat(self, job_id: int) -> dict:
+        """空闲检测 —— Admin 调用 POST /idleBeat
+
+        检查指定 jobId 是否正在当前执行器中运行。
+        """
+        if job_id in self._running_jobs and not self._running_jobs[job_id].done():
+            return {"code": 500, "msg": f"job {job_id} is running"}
+        return {"code": 200, "msg": None}
+
+    async def run(self, trigger_params: Dict[str, Any]) -> dict:
+        """执行任务 —— Admin 调用 POST /run
+
+        trigger_params 字段（XXL-JOB 标准）:
+            jobId, executorHandler, executorParams, executorBlockStrategy,
+            executorTimeout, logId, logDateTime, glueType, glueSource,
+            glueUpdatetime, broadcastIndex, broadcastTotal
+
+        Returns:
+            {"code": 200, "msg": None} 或 {"code": 500, "msg": "error detail"}
+        """
+        job_id = trigger_params.get("jobId")
+        handler_name = trigger_params.get("executorHandler")
+        param = trigger_params.get("executorParams", "")
+        timeout = trigger_params.get("executorTimeout", 0)
+
+        if not handler_name:
+            return {"code": 500, "msg": "executorHandler is required"}
+
+        handler = xxl_registry.get_handler(handler_name)
+        if not handler:
+            registered = xxl_registry.list_handlers()
+            return {
+                "code": 500,
+                "msg": f"Unknown handler: {handler_name}. "
+                f"Registered: {', '.join(registered) if registered else '(none)'}",
+            }
+
+        # 异步执行，带超时
+        async def _run_with_timeout():
+            try:
+                if timeout and timeout > 0:
+                    return await asyncio.wait_for(handler(param), timeout=timeout)
+                return await handler(param)
+            except asyncio.TimeoutError:
+                logger.error(
+                    "Job %s(id=%s) timeout after %ss", handler_name, job_id, timeout
+                )
+                return {"code": 500, "msg": f"timeout after {timeout}s"}
+            except Exception as e:
+                logger.exception("Job %s(id=%s) failed", handler_name, job_id)
+                return {"code": 500, "msg": str(e)}
+
+        task = asyncio.create_task(_run_with_timeout())
+        if job_id is not None:
+            self._running_jobs[job_id] = task
+
+        try:
+            result = await task
+            return result
+        finally:
+            if job_id is not None:
+                self._running_jobs.pop(job_id, None)
+
+    async def kill(self, job_id: int) -> dict:
+        """终止任务 —— Admin 调用 POST /kill"""
+        task = self._running_jobs.get(job_id)
+        if task and not task.done():
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+            return {"code": 200, "msg": None}
+        return {"code": 500, "msg": f"job {job_id} not found or already finished"}
+
+    def log(
+        self, log_date_time: int, log_id: int, from_line_num: int
+    ) -> dict:
+        """读取执行日志 —— Admin 调用 POST /log"""
+        log_file = os.path.join(
+            self.config.log_path,
+            str(log_date_time),
+            f"{log_id}.log",
+        )
+        result = {
+            "fromLineNum": from_line_num,
+            "toLineNum": from_line_num,
+            "logContent": "",
+            "isEnd": True,
+        }
+
+        if not os.path.exists(log_file):
+            return result
+
+        try:
+            with open(log_file, "r", encoding="utf-8") as f:
+                lines = f.readlines()
+            total_lines = len(lines)
+            if from_line_num >= total_lines:
+                result["isEnd"] = True
+                result["toLineNum"] = total_lines
+                return result
+
+            end_line = min(from_line_num + 100, total_lines)
+            result["logContent"] = "".join(lines[from_line_num:end_line])
+            result["fromLineNum"] = from_line_num
+            result["toLineNum"] = end_line
+            result["isEnd"] = end_line >= total_lines
+        except Exception as e:
+            logger.warning("Read xxl log file %s failed: %s", log_file, e)
+            result["logContent"] = f"read log error: {e}"
+
+        return result
+
+    # ==================== 辅助 ====================
+
+    def _admin_urls(self) -> list:
+        """解析 admin_addresses（多个逗号分隔）"""
+        return [
+            u.strip().rstrip("/")
+            for u in self.config.admin_addresses.split(",")
+            if u.strip()
+        ]
+
+    def _auth_headers(self) -> dict:
+        """构造带 XXL-JOB 令牌的请求头"""
+        if self.config.access_token:
+            return {"XXL-JOB-ACCESS-TOKEN": self.config.access_token}
+        return {}
+
+    @staticmethod
+    def _local_ip() -> str:
+        """探测本机 IP"""
+        try:
+            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            s.connect(("8.8.8.8", 80))
+            ip = s.getsockname()[0]
+            s.close()
+            return ip
+        except Exception:
+            return "127.0.0.1"

supply/server/__init__.py

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

supply/server/app.py

@@ -0,0 +1,73 @@
+"""Quart + Hypercorn 应用工厂 —— 内建 DB + 日志服务 + XXL-JOB 生命周期"""
+
+import logging
+
+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.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.server.xxl_endpoints import set_xxl_executor
+
+logger = logging.getLogger(__name__)
+
+_db_configs = {
+    "aigc": AigcDatabaseConfig(),
+    "long_articles": LongArticlesDatabaseConfig(),
+    "piaoquan_crawler": PiaoquanCrawlerDatabaseConfig(),
+}
+
+
+def _build_backends() -> dict:
+    """将配置字典转换为后端实例字典"""
+    return {
+        name: create_backend(name, config)
+        for name, config in _db_configs.items()
+    }
+
+
+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)
+
+    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)
+    ctx.on_startup(xxl_executor.start)
+    ctx.on_shutdown(xxl_executor.stop)
+    ctx.on_shutdown(db.close_pools)
+    ctx.on_shutdown(log_service.stop)
+
+    @app.before_serving
+    async def startup():
+        logger.info("Starting application...")
+        await ctx.start_up()
+        logger.info("Application started successfully")
+
+    @app.after_serving
+    async def shutdown():
+        logger.info("Shutting down application...")
+        await ctx.shutdown()
+        logger.info("Application shutdown successfully")
+
+    return app

supply/server/endpoints.py

@@ -0,0 +1,16 @@
+"""通用 API 端点"""
+
+from quart import Blueprint, jsonify
+
+
+def create_health_bp() -> Blueprint:
+    """健康检查端点 —— GET /health"""
+    bp = Blueprint("health", __name__)
+
+    @bp.route("/health", methods=["GET"])
+    async def health():
+        return jsonify(
+            {"code": 0, "message": "success", "data": {"message": "hello world"}}
+        )
+
+    return bp

supply/server/routes.py

@@ -0,0 +1,13 @@
+"""路由注册"""
+
+from quart import Blueprint
+from supply.server.endpoints import create_health_bp
+from supply.server.xxl_endpoints import create_xxl_bp
+
+
+def server_routes() -> Blueprint:
+    """注册所有 Blueprint，前缀 /api"""
+    api = Blueprint("api", __name__, url_prefix="/api")
+    api.register_blueprint(create_health_bp())
+    api.register_blueprint(create_xxl_bp())
+    return api

supply/server/scheduler.py

@@ -0,0 +1,402 @@
+"""通用任务调度器 —— 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})()
+            ),
+        )

supply/server/xxl_endpoints.py

@@ -0,0 +1,95 @@
+"""XXL-JOB Admin → Executor 通信端点 —— 对标 Java XxlJobController.java
+
+提供给 XXL-JOB Admin 调用的 HTTP API：
+    POST /beat       — 心跳检测
+    POST /idleBeat   — 空闲检测
+    POST /run        — 触发任务执行
+    POST /kill       — 强制终止任务
+    POST /log        — 读取执行日志
+"""
+
+import logging
+
+from quart import Blueprint, request, jsonify
+
+logger = logging.getLogger(__name__)
+
+# 全局引用，由 create_xxl_bp 注入
+_executor_ref = None
+
+
+def set_xxl_executor(executor):
+    """注入 XxlJobExecutor 实例（在 app 启动时调用）"""
+    global _executor_ref
+    _executor_ref = executor
+
+
+def create_xxl_bp() -> Blueprint:
+    """创建 XXL-JOB API Blueprint
+
+    XXL-JOB Admin 通过此 Blueprint 与执行器通信。
+    端点路径与 XXL-JOB 标准一致：/beat, /idleBeat, /run, /kill, /log
+    """
+    bp = Blueprint("xxl_job", __name__)
+
+    @bp.route("/beat", methods=["POST"])
+    async def beat():
+        """心跳检测"""
+        result = _executor_ref.beat()
+        return jsonify(result)
+
+    @bp.route("/idleBeat", methods=["POST"])
+    async def idle_beat():
+        """空闲检测：检查指定 jobId 是否正在运行"""
+        form = await request.form
+        job_id = int(form.get("jobId", 0))
+        result = _executor_ref.idle_beat(job_id)
+        return jsonify(result)
+
+    @bp.route("/run", methods=["POST"])
+    async def run():
+        """触发任务执行"""
+        form = await request.form
+        trigger_params = {
+            "jobId": int(form.get("jobId", 0)),
+            "executorHandler": form.get("executorHandler", ""),
+            "executorParams": form.get("executorParams", ""),
+            "executorBlockStrategy": form.get("executorBlockStrategy", ""),
+            "executorTimeout": int(form.get("executorTimeout", 0)),
+            "logId": int(form.get("logId", 0)),
+            "logDateTime": int(form.get("logDateTime", 0)),
+            "glueType": form.get("glueType", ""),
+            "glueSource": form.get("glueSource", ""),
+            "glueUpdatetime": int(form.get("glueUpdatetime", 0)),
+            "broadcastIndex": int(form.get("broadcastIndex", 0)),
+            "broadcastTotal": int(form.get("broadcastTotal", 1)),
+        }
+        logger.info(
+            "XXL-JOB run: handler=%s, jobId=%s, params=%s",
+            trigger_params["executorHandler"],
+            trigger_params["jobId"],
+            trigger_params["executorParams"],
+        )
+        result = await _executor_ref.run(trigger_params)
+        return jsonify(result)
+
+    @bp.route("/kill", methods=["POST"])
+    async def kill():
+        """强制终止任务"""
+        form = await request.form
+        job_id = int(form.get("jobId", 0))
+        logger.info("XXL-JOB kill: jobId=%s", job_id)
+        result = await _executor_ref.kill(job_id)
+        return jsonify(result)
+
+    @bp.route("/log", methods=["POST"])
+    async def log():
+        """读取执行日志"""
+        form = await request.form
+        log_date_time = int(form.get("logDateTim", 0))
+        log_id = int(form.get("logId", 0))
+        from_line_num = int(form.get("fromLineNum", 1))
+        result = _executor_ref.log(log_date_time, log_id, from_line_num)
+        return jsonify(result)
+
+    return bp