3852c6f87d
Some checks failed
CI / test (push) Failing after 4m16s
P3 deepening (issue #9): remove direct ORM from router handlers. - decision_logs.py (new): create_decision_log / list_decision_logs service - cron_pool.heartbeat: encapsulate heartbeat cron lookup + update + commit - cron_pool.scale_to: encapsulate scale direction decision (if/elif/else) - task_queue.get_next_task_with_eval: encapsulate eval-loading + dict-building - Router endpoints now delegate to services, only handling HTTP-level validation (status codes, 404 translation via LookupError). No observable behaviour change — 873 passed + 5 xfailed unchanged. T3 router ORM contract guards (5/5) continue to pass.
323 lines
10 KiB
Python
323 lines
10 KiB
Python
"""Cron pool management for intelligent evaluations (Cron 池管理).
|
|
|
|
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"}
|