2dd023fdd9
All checks were successful
CI / test (push) Successful in 4m2s
方案③落地后，智能评估执行机制从'常驻 cron 每分钟自唤醒'改为'平台每 60s 扫描入队 + 按需触发无状态 headless agent'（触发式执行）。对齐领域语言： - CONTEXT.md：Cron 池/工作单元（Worker）标 deprecated；新增触发式执行词条； 修正时间窗口（cron 自唤醒→平台扫描时段到期）、任务队列（消费端）、决策日志 - ADR-0009 新增：记录触发式执行取代 cron 池的决策（原因：cron 需外部 channel， OpenClaw webchat 非 channel 账号无法 delivery）；ADR-0007 标 superseded - 代码标 deprecated：cron_pool / fault_tolerance / openclaw_cron_pool 路由 / CronPoolMonitor 页（导航入口已从 App.tsx 移除，监控由 TaskQueueMonitor 承担） 895 passed, vitest 19 passed
327 lines
10 KiB
Python
327 lines
10 KiB
Python
"""Cron pool management for intelligent evaluations (Cron 池管理).
|
||
|
||
DEPRECATED (ADR-0009): 智能评估已改为触发式执行（平台扫描入队 + 触发 headless
|
||
agent），cron 池不再使用，initialize_pool 不进 lifespan，t480 worker cron 已禁用。
|
||
遗留保留仅供回溯；卡死检测由 ``task_queue.requeue_stale_assigned_tasks`` 承担。
|
||
|
||
Platform manages a pool of OpenClaw crons (5-20) that can process any
|
||
intelligent evaluation. Pool automatically scales up/down based on load.
|
||
"""
|
||
|
||
import logging
|
||
from datetime import timedelta
|
||
|
||
from sqlmodel import Session, select
|
||
|
||
from agenteval.intelligent_eval.openclaw_client import OpenClawClient
|
||
from agenteval.intelligent_eval.task_queue import requeue_stuck_task
|
||
from agenteval.storage.db import OpenClawCronPoolDB, utc_now
|
||
|
||
_logger = logging.getLogger("agenteval")
|
||
|
||
# Pool configuration
|
||
MIN_POOL_SIZE = 5
|
||
MAX_POOL_SIZE = 20
|
||
SCALE_UP_THRESHOLD = 0.8 # busy/total > 0.8 triggers scale up
|
||
SCALE_DOWN_THRESHOLD = 2 # idle > min_size * 2 triggers scale down
|
||
STUCK_THRESHOLD_MINUTES = 10
|
||
|
||
|
||
async def initialize_pool(session: Session, client: OpenClawClient) -> int:
|
||
"""Initialize cron pool on startup.
|
||
|
||
Creates MIN_POOL_SIZE crons if pool is empty.
|
||
|
||
Returns:
|
||
Number of crons created
|
||
"""
|
||
# Check if pool already initialized
|
||
existing = session.exec(select(OpenClawCronPoolDB)).all()
|
||
if existing:
|
||
_logger.info(f"Cron pool already initialized with {len(existing)} crons")
|
||
return 0
|
||
|
||
# Create MIN_POOL_SIZE crons
|
||
created = 0
|
||
for i in range(MIN_POOL_SIZE):
|
||
try:
|
||
cron_id = await client.create_cron(
|
||
name=f"intelligent-eval-worker-{i}",
|
||
schedule="* * * * *", # Every minute
|
||
skill="agenteval-intelligent-worker",
|
||
state={"status": "idle"},
|
||
)
|
||
|
||
# Record in DB
|
||
cron_db = OpenClawCronPoolDB(
|
||
openclaw_cron_id=cron_id,
|
||
status="idle",
|
||
last_active_at=utc_now(),
|
||
)
|
||
session.add(cron_db)
|
||
created += 1
|
||
except Exception as exc:
|
||
_logger.error(f"Failed to create cron {i}: {exc}")
|
||
|
||
session.commit()
|
||
_logger.info(f"Initialized cron pool with {created} crons")
|
||
return created
|
||
|
||
|
||
async def scale_up(count: int, session: Session, client: OpenClawClient) -> int:
|
||
"""Scale up the pool by creating new crons.
|
||
|
||
Args:
|
||
count: Number of crons to create
|
||
session: Database session
|
||
client: OpenClaw client
|
||
|
||
Returns:
|
||
Number of crons created
|
||
"""
|
||
# Check current pool size
|
||
current_size = len(session.exec(select(OpenClawCronPoolDB)).all())
|
||
if current_size >= MAX_POOL_SIZE:
|
||
_logger.warning(f"Pool already at max size ({MAX_POOL_SIZE})")
|
||
return 0
|
||
|
||
# Limit count to not exceed max size
|
||
count = min(count, MAX_POOL_SIZE - current_size)
|
||
|
||
created = 0
|
||
for i in range(count):
|
||
try:
|
||
cron_id = await client.create_cron(
|
||
name=f"intelligent-eval-worker-{current_size + i}",
|
||
schedule="* * * * *",
|
||
skill="agenteval-intelligent-worker",
|
||
state={"status": "idle"},
|
||
)
|
||
|
||
cron_db = OpenClawCronPoolDB(
|
||
openclaw_cron_id=cron_id,
|
||
status="idle",
|
||
last_active_at=utc_now(),
|
||
)
|
||
session.add(cron_db)
|
||
created += 1
|
||
except Exception as exc:
|
||
_logger.error(f"Failed to create cron during scale up: {exc}")
|
||
|
||
session.commit()
|
||
_logger.info(f"Scaled up pool by {created} crons (total: {current_size + created})")
|
||
return created
|
||
|
||
|
||
async def scale_down(count: int, session: Session, client: OpenClawClient) -> int:
|
||
"""Scale down the pool by deleting idle crons.
|
||
|
||
Args:
|
||
count: Number of crons to delete
|
||
session: Database session
|
||
client: OpenClaw client
|
||
|
||
Returns:
|
||
Number of crons deleted
|
||
"""
|
||
# Check current pool size
|
||
current_size = len(session.exec(select(OpenClawCronPoolDB)).all())
|
||
if current_size <= MIN_POOL_SIZE:
|
||
_logger.warning(f"Pool already at min size ({MIN_POOL_SIZE})")
|
||
return 0
|
||
|
||
# Limit count to not go below min size
|
||
count = min(count, current_size - MIN_POOL_SIZE)
|
||
|
||
# Find idle crons to delete
|
||
idle_crons = session.exec(
|
||
select(OpenClawCronPoolDB)
|
||
.where(OpenClawCronPoolDB.status == "idle")
|
||
.order_by(OpenClawCronPoolDB.last_active_at)
|
||
.limit(count)
|
||
).all()
|
||
|
||
deleted = 0
|
||
for cron in idle_crons:
|
||
try:
|
||
await client.delete_cron(cron.openclaw_cron_id)
|
||
session.delete(cron)
|
||
deleted += 1
|
||
except Exception as exc:
|
||
_logger.error(f"Failed to delete cron {cron.openclaw_cron_id}: {exc}")
|
||
|
||
session.commit()
|
||
_logger.info(f"Scaled down pool by {deleted} crons (total: {current_size - deleted})")
|
||
return deleted
|
||
|
||
|
||
async def auto_scale(session: Session, client: OpenClawClient) -> tuple[int, int]:
|
||
"""Automatically scale pool based on load.
|
||
|
||
Returns:
|
||
(scaled_up, scaled_down) counts
|
||
"""
|
||
crons = session.exec(select(OpenClawCronPoolDB)).all()
|
||
total = len(crons)
|
||
|
||
if total == 0:
|
||
# Pool not initialized
|
||
return (0, 0)
|
||
|
||
busy = sum(1 for c in crons if c.status == "busy")
|
||
idle = sum(1 for c in crons if c.status == "idle")
|
||
|
||
scaled_up = 0
|
||
scaled_down = 0
|
||
|
||
# Scale up if busy/total > threshold and not at max
|
||
if busy / total > SCALE_UP_THRESHOLD and total < MAX_POOL_SIZE:
|
||
scaled_up = await scale_up(1, session, client)
|
||
|
||
# Scale down if idle > min_size * threshold and not at min
|
||
elif idle > MIN_POOL_SIZE * SCALE_DOWN_THRESHOLD and total > MIN_POOL_SIZE:
|
||
scaled_down = await scale_down(1, session, client)
|
||
|
||
return (scaled_up, scaled_down)
|
||
|
||
|
||
def get_pool_status(session: Session) -> dict:
|
||
"""Get current pool status.
|
||
|
||
Returns:
|
||
Dict with pool stats
|
||
"""
|
||
crons = session.exec(select(OpenClawCronPoolDB)).all()
|
||
total = len(crons)
|
||
idle = sum(1 for c in crons if c.status == "idle")
|
||
busy = sum(1 for c in crons if c.status == "busy")
|
||
stuck = sum(1 for c in crons if c.status == "stuck")
|
||
|
||
return {
|
||
"total": total,
|
||
"idle": idle,
|
||
"busy": busy,
|
||
"stuck": stuck,
|
||
"min_size": MIN_POOL_SIZE,
|
||
"max_size": MAX_POOL_SIZE,
|
||
}
|
||
|
||
|
||
async def sync_cron_states(session: Session, client: OpenClawClient) -> int:
|
||
"""Sync cron states from OpenClaw to platform DB.
|
||
|
||
Returns:
|
||
Number of crons synced
|
||
"""
|
||
# Get all crons from OpenClaw
|
||
openclaw_crons = await client.list_crons()
|
||
|
||
synced = 0
|
||
for oc_cron in openclaw_crons:
|
||
# Find corresponding DB record
|
||
db_cron = session.exec(
|
||
select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == oc_cron.id)
|
||
).first()
|
||
|
||
if db_cron is None:
|
||
# New cron, add to DB
|
||
db_cron = OpenClawCronPoolDB(
|
||
openclaw_cron_id=oc_cron.id,
|
||
status="idle" if oc_cron.enabled else "disabled",
|
||
last_active_at=utc_now(),
|
||
)
|
||
session.add(db_cron)
|
||
synced += 1
|
||
else:
|
||
# Update existing record
|
||
if oc_cron.state:
|
||
new_status = oc_cron.state.get("status", "idle")
|
||
if db_cron.status != new_status:
|
||
db_cron.status = new_status
|
||
db_cron.updated_at = utc_now()
|
||
synced += 1
|
||
|
||
session.commit()
|
||
return synced
|
||
|
||
|
||
def detect_stuck_crons(session: Session) -> list[OpenClawCronPoolDB]:
|
||
"""Detect stuck crons (busy but not active for > 10 minutes).
|
||
|
||
Returns:
|
||
List of stuck crons
|
||
"""
|
||
threshold = utc_now() - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
|
||
stuck = session.exec(
|
||
select(OpenClawCronPoolDB).where(
|
||
OpenClawCronPoolDB.status == "busy",
|
||
OpenClawCronPoolDB.last_active_at < threshold,
|
||
)
|
||
).all()
|
||
return list(stuck)
|
||
|
||
|
||
async def handle_stuck_cron(cron: OpenClawCronPoolDB, session: Session, client: OpenClawClient) -> None:
|
||
"""Handle a stuck cron: mark as stuck, requeue task, delete cron, create new one.
|
||
|
||
Args:
|
||
cron: Stuck cron
|
||
session: Database session
|
||
client: OpenClaw client
|
||
"""
|
||
_logger.warning(f"Handling stuck cron {cron.openclaw_cron_id}")
|
||
|
||
# Mark as stuck
|
||
cron.status = "stuck"
|
||
cron.updated_at = utc_now()
|
||
|
||
# Requeue task if any
|
||
if cron.current_eval_id:
|
||
# P1 deepening (S4): settlement delegated to task_queue.
|
||
requeue_stuck_task(cron.current_eval_id, cron.openclaw_cron_id, session)
|
||
|
||
# Delete stuck cron
|
||
try:
|
||
await client.delete_cron(cron.openclaw_cron_id)
|
||
session.delete(cron)
|
||
except Exception as exc:
|
||
_logger.error(f"Failed to delete stuck cron {cron.openclaw_cron_id}: {exc}")
|
||
|
||
# Create new cron to replace
|
||
await scale_up(1, session, client)
|
||
|
||
session.commit()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# P3 deepening (S2) — router logic moved into service
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def heartbeat(cron_id: str, status: str, current_eval_id: str | None, session: Session) -> None:
|
||
"""Record a cron heartbeat. Raises ``LookupError`` if cron not found."""
|
||
cron = session.exec(
|
||
select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == cron_id)
|
||
).first()
|
||
if cron is None:
|
||
raise LookupError(f"cron {cron_id} not found")
|
||
cron.last_active_at = utc_now()
|
||
cron.status = status
|
||
cron.current_eval_id = current_eval_id
|
||
cron.updated_at = utc_now()
|
||
session.commit()
|
||
|
||
|
||
async def scale_to(target_size: int, session: Session, client: OpenClawClient) -> dict:
|
||
"""Scale the pool to *target_size*. Returns a dict suitable for the HTTP response."""
|
||
current_status = get_pool_status(session)
|
||
current = current_status["total"]
|
||
|
||
if target_size > current:
|
||
created = await scale_up(target_size - current, session, client)
|
||
return {"success": True, "scaled_up": created, "current_size": current + created}
|
||
if target_size < current:
|
||
deleted = await scale_down(current - target_size, session, client)
|
||
return {"success": True, "scaled_down": deleted, "current_size": current - deleted}
|
||
return {"success": True, "current_size": current, "message": "already at target size"}
|