Agent 任务队列设计：从 Issue 到执行器的调度模型

# 反例：直调模式
agent.run("帮我修复这个 Bug")
# 问题：
# 1. 当前 Agent 忙怎么办？—— 阻塞等待
# 2. 任务优先级怎么处理？—— 没有优先级
# 3. 失败后怎么办？—— 没有重试
# 4. 谁发起了什么任务？—— 没有审计
# 5. 高峰期怎么限流？—— 无法控制

┌─────────────┐    入队    ┌──────────┐    出队    ┌─────────────┐
│  Issue/PR   │ ────────→ │  Task     │ ────────→ │   Agent     │
│  API 请求   │           │  Queue    │           │   执行器    │
│  定时任务   │           │           │           │  (沙箱内)   │
│  人工指派   │           │  持久化   │           │             │
└─────────────┘           └──────┬───┘           └─────────────┘
                                 │
                          ┌──────▼───────┐
                          │  Audit Log   │
                          │  (事件溯源)  │
                          └──────────────┘

CREATE TABLE agent_tasks (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    -- 任务来源
    source          VARCHAR(32)  NOT NULL,  -- 'issue', 'pr', 'api', 'cron', 'manual'
    source_id       VARCHAR(128),           -- GitHub Issue #ID, PR #ID, etc.
    -- 任务内容
    title           VARCHAR(256) NOT NULL,
    task_type       VARCHAR(64)  NOT NULL,  -- 'bugfix', 'feature', 'review', 'test', 'docs'
    payload         JSONB        NOT NULL,  -- Agent 执行所需的上下文包
    priority        SMALLINT     NOT NULL DEFAULT 5,  -- 0=最高, 9=最低
    -- 调度控制
    status          VARCHAR(16)  NOT NULL DEFAULT 'pending',
                    -- pending → queued → executing → completed
                    --                                  → failed → pending(重试)
                    --              → cancelled
                    --              → pending(超时重试)
                    --              → pending(人工接管后重指派)
    max_retries     SMALLINT     NOT NULL DEFAULT 3,
    retry_count     SMALLINT     NOT NULL DEFAULT 0,
    timeout_secs    INTEGER      NOT NULL DEFAULT 300,  -- 单次执行超时
    lock_token      UUID,                  -- 执行器持有，防重复调度
    locked_at       TIMESTAMPTZ,           -- 锁定时间，用于超时检测
    -- 执行与回调
    assigned_to     VARCHAR(64),            -- 执行器 ID 或 Agent 名称
    callback_url    VARCHAR(512),           -- 完成后回调（可选）
    result          JSONB,                  -- 执行结果摘要
    -- 审计
    requested_by    VARCHAR(64)  NOT NULL,  -- 发起人
    approved_by     VARCHAR(64),            -- 审批人（如果需要）
    created_at      TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ  NOT NULL DEFAULT NOW()
);

-- 核心查询索引
CREATE INDEX idx_tasks_status_priority ON agent_tasks (status, priority, created_at)
    WHERE status IN ('pending', 'queued');
CREATE INDEX idx_tasks_locked_at ON agent_tasks (lock_token, locked_at)
    WHERE lock_token IS NOT NULL;
CREATE INDEX idx_tasks_source ON agent_tasks (source, source_id);

-- 去重约束：同一来源同一 ID 只能有一个活跃任务
CREATE UNIQUE INDEX idx_tasks_dedup ON agent_tasks (source, source_id)
    WHERE status NOT IN ('completed', 'cancelled', 'failed');

                  ┌──────────────────────┐
                  │      pending          │ ← 新创建，未入调度
                  └──────────┬───────────┘
                             │ 调度器扫描
                             ▼
                  ┌──────────────────────┐
                  │       queued          │ ← 已入队，等待执行器领取
                  └──────────┬───────────┘
                             │ 执行器领取
                             ▼
                  ┌──────────────────────┐
                  │     executing         │ ← 正在执行中
                  └──────┬───────────┬───┘
                         │           │
                   成功完成     失败/超时
                         │           │
                         ▼           ▼
                  ┌──────────┐ ┌──────────────┐
                  │completed │ │    failed     │
                  └──────────┘ │              │
                               │  重试次数用完?│
                               ├── 是 → dead   │
                               │  否 → pending │
                               └──────────────┘
                                         │
                                   人工接管
                                         │
                                         ▼
                               ┌──────────────────┐
                               │ manual_intervene  │
                               └──────────────────┘

import time
import uuid
from datetime import datetime, timedelta, timezone
from typing import Optional

class TaskScheduler:
    """轻量级任务调度器，适用于中小团队（<50 Agent）"""

    def __init__(self, db_pool, poll_interval: float = 1.0):
        self.db = db_pool
        self.poll_interval = poll_interval  # 秒
        self.running = False

    def tick(self):
        """一次调度心跳：超时检测 → 优先级入队 → 死信转移"""
        now = datetime.now(timezone.utc)

        # 1. 超时检测：执行超时的任务标记为失败
        self.db.execute("""
            UPDATE agent_tasks
            SET status = 'failed',
                retry_count = retry_count + 1,
                updated_at = NOW()
            WHERE status = 'executing'
              AND locked_at < $1 - $2::interval
              AND lock_token IS NOT NULL
        """, now, timedelta(seconds=300))  # 默认超时 300s

        # 2. 优先级入队：把 pending 任务按规定数量入队
        self.db.execute("""
            UPDATE agent_tasks
            SET status = 'queued',
                updated_at = NOW()
            WHERE id IN (
                SELECT id FROM agent_tasks
                WHERE status = 'pending'
                  AND retry_count < max_retries
                ORDER BY priority ASC, created_at ASC
                LIMIT $1   -- 每次最多入队 N 个
                FOR UPDATE SKIP LOCKED
            )
        """, self.max_concurrent)

        # 3. 死信转移：重试次数超限的任务
        self.db.execute("""
            UPDATE agent_tasks
            SET status = 'dead',
                updated_at = NOW()
            WHERE status = 'failed'
              AND retry_count >= max_retries
        """)

    def run_forever(self):
        """启动调度循环"""
        self.running = True
        while self.running:
            self.tick()
            time.sleep(self.poll_interval)

    def stop(self):
        self.running = False

# 等得越久，优先级越高（aging）
有效优先级 = 静态优先级 - floor(等待时间 / aging_interval)
aging_interval = 10 分钟

# 例：一个 P3 任务等了 30 分钟
# 有效优先级 = 5 - floor(30/10) = 5 - 3 = 2 → 相当于 P1

def claim_task(self) -> Optional[dict]:
    """原子领取一个 queued 任务"""
    lock = uuid.uuid4()
    now = datetime.now(timezone.utc)

    result = self.db.fetchone("""
        UPDATE agent_tasks
        SET status = 'executing',
            lock_token = $1,
            locked_at = $2,
            assigned_to = $3,
            updated_at = $2
        WHERE id = (
            SELECT id FROM agent_tasks
            WHERE status = 'queued'
            ORDER BY priority ASC, created_at ASC
            LIMIT 1
            FOR UPDATE SKIP LOCKED
        )
        RETURNING id, title, task_type, payload, timeout_secs, callback_url
    """, lock, now, self.executor_id)

    return result

def complete_task(self, task_id: str, lock_token: str, result: dict):
    """完成任务并回传结果"""
    self.db.execute("""
        UPDATE agent_tasks
        SET status = 'completed',
            result = $1::jsonb,
            lock_token = NULL,
            locked_at = NULL,
            updated_at = NOW()
        WHERE id = $2
          AND lock_token = $3   -- 验证锁持有者
    """, json.dumps(result), task_id, lock_token)


def fail_task(self, task_id: str, lock_token: str, error: str):
    """标记任务失败（如果还有重试次数，调度器会自动重试）"""
    self.db.execute("""
        UPDATE agent_tasks
        SET status = 'failed',
            lock_token = NULL,
            locked_at = NULL,
            result = jsonb_build_object('error', $1),
            updated_at = NOW()
        WHERE id = $2
          AND lock_token = $3
    """, error, task_id, lock_token)

import signal

class AgentExecutor:
    def execute_with_timeout(self, task: dict) -> dict:
        """带超时的 Agent 执行"""
        timeout_secs = task.get('timeout_secs', 300)

        # 方案一：SIGALRM 信号（仅 Unix）
        signal.signal(signal.SIGALRM, self._timeout_handler)
        signal.alarm(timeout_secs)
        try:
            result = self._run_agent(task)   # 实际调用 Agent
            return {'status': 'ok', 'data': result}
        except TimeoutError:
            # 超时后杀掉 Agent 子进程
            self._kill_agent_process()
            return {'status': 'timeout', 'error': f'超过 {timeout_secs}s 限制'}
        finally:
            signal.alarm(0)  # 取消闹钟


    def cancel_task(self, task_id: str):
        """由调度器或用户发起的取消"""
        # 1. 数据库标记取消
        self.db.execute("""
            UPDATE agent_tasks
            SET status = 'cancelled',
                lock_token = NULL,
                locked_at = NULL,
                updated_at = NOW()
            WHERE id = $1 AND status = 'executing'
        """, task_id)
        # 2. 杀死执行进程
        self._kill_agent_process()

TIMEOUT_MAP = {
    'bugfix':     120,   # 2 min
    'review':     180,   # 3 min
    'feature':    600,   # 10 min
    'test':       120,   # 2 min
    'docs':        60,   # 1 min
    'refactor':   900,   # 15 min
}

# 队列深度（按状态）
agent_queue_depth{status="pending"}    # 待入队
agent_queue_depth{status="queued"}     # 等待执行
agent_queue_depth{status="executing"}  # 正在执行
agent_queue_depth{status="dead"}       # 死信（重点关注）

# 任务延迟
agent_queue_latency_seconds   # 任务从创建到开始执行的延迟
agent_queue_completion_time   # 任务从开始到完成的耗时

# 任务结果
agent_tasks_total{status="completed"}  # 成功
agent_tasks_total{status="failed"}     # 失败
agent_tasks_total{status="timeout"}    # 超时
agent_retries_total                    # 重试次数