From 3852c6f87d4c19d450c21722d2d3b2ddc00219b4 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Thu, 13 Aug 2026 13:57:34 +0800 Subject: [PATCH] refactor(intelligent-eval): router logic down to service layer (P3, S2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../agenteval/intelligent_eval/cron_pool.py | 33 ++++++ .../intelligent_eval/decision_logs.py | 71 ++++++++++++ .../agenteval/intelligent_eval/task_queue.py | 36 +++++++ .../web/routers/intelligent_evals.py | 101 +++--------------- .../web/routers/openclaw_cron_pool.py | 46 ++------ 5 files changed, 161 insertions(+), 126 deletions(-) create mode 100644 backend/agenteval/intelligent_eval/decision_logs.py diff --git a/backend/agenteval/intelligent_eval/cron_pool.py b/backend/agenteval/intelligent_eval/cron_pool.py index bd95693..cb95d1b 100644 --- a/backend/agenteval/intelligent_eval/cron_pool.py +++ b/backend/agenteval/intelligent_eval/cron_pool.py @@ -287,3 +287,36 @@ async def handle_stuck_cron(cron: OpenClawCronPoolDB, session: Session, client: 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"} diff --git a/backend/agenteval/intelligent_eval/decision_logs.py b/backend/agenteval/intelligent_eval/decision_logs.py new file mode 100644 index 0000000..6c7f133 --- /dev/null +++ b/backend/agenteval/intelligent_eval/decision_logs.py @@ -0,0 +1,71 @@ +"""Decision-log service (P3 deepening, S2). + +Pulled out of ``web/routers/intelligent_evals.py`` so the router only handles +HTTP validation and error translation. The ORM writes and reads now live here. +""" +from typing import Any + +from sqlmodel import Session, select + +from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalDecisionLogDB + + +def create_decision_log( + eval_id: str, + decision_type: str, + reason: str, + cron_id: str, + context: dict[str, Any], + session: Session, +) -> dict: + """Create a decision log entry. Raises ``LookupError`` if eval not found.""" + eval_db = session.get(IntelligentEvalDB, eval_id) + if eval_db is None: + raise LookupError(f"intelligent eval {eval_id} not found") + + log = IntelligentEvalDecisionLogDB( + eval_id=eval_id, + decision_type=decision_type, + reason=reason, + cron_id=cron_id, + ) + log.set_context(context) + session.add(log) + session.commit() + session.refresh(log) + + return { + "id": log.id, + "eval_id": log.eval_id, + "decision_type": log.decision_type, + "reason": log.reason, + "context": log.get_context(), + "cron_id": log.cron_id, + "created_at": log.created_at.isoformat() if log.created_at else None, + } + + +def list_decision_logs(eval_id: str, session: Session) -> list[dict]: + """List decision logs for an eval. Raises ``LookupError`` if eval not found.""" + eval_db = session.get(IntelligentEvalDB, eval_id) + if eval_db is None: + raise LookupError(f"intelligent eval {eval_id} not found") + + logs = session.exec( + select(IntelligentEvalDecisionLogDB) + .where(IntelligentEvalDecisionLogDB.eval_id == eval_id) + .order_by(IntelligentEvalDecisionLogDB.created_at.desc()) + ).all() + + return [ + { + "id": log.id, + "eval_id": log.eval_id, + "decision_type": log.decision_type, + "reason": log.reason, + "context": log.get_context(), + "cron_id": log.cron_id, + "created_at": log.created_at.isoformat() if log.created_at else None, + } + for log in logs + ] diff --git a/backend/agenteval/intelligent_eval/task_queue.py b/backend/agenteval/intelligent_eval/task_queue.py index 17f5a11..5bb2749 100644 --- a/backend/agenteval/intelligent_eval/task_queue.py +++ b/backend/agenteval/intelligent_eval/task_queue.py @@ -179,3 +179,39 @@ def requeue_stuck_task(eval_id: str, cron_id: str, session: Session) -> bool: ) session.add(new_task) return True + + +# --------------------------------------------------------------------------- +# P3 deepening (S2) — get_next_task with embedded eval info +# --------------------------------------------------------------------------- + + +def get_next_task_with_eval(session: Session) -> Optional[dict]: + """Return the next pending task with its eval details, or None. + + P3 deepening (S2): the eval-loading + dict-building that previously lived in + ``web/routers/intelligent_evals.py::get_next_task`` now lives here. + """ + task = get_next_task(session) + if task is None: + return None + + eval_db = session.get(IntelligentEvalDB, task.eval_id) + if eval_db is None: + return None + + return { + "task": { + "id": task.id, + "eval_id": task.eval_id, + "priority": task.priority, + "reason": task.reason, + "eval": { + "id": eval_db.id, + "name": eval_db.name, + "status": eval_db.status, + "plan": eval_db.get_plan(), + "started_at": eval_db.started_at.isoformat() if eval_db.started_at else None, + }, + }, + } diff --git a/backend/agenteval/web/routers/intelligent_evals.py b/backend/agenteval/web/routers/intelligent_evals.py index 867e26d..4c1785e 100644 --- a/backend/agenteval/web/routers/intelligent_evals.py +++ b/backend/agenteval/web/routers/intelligent_evals.py @@ -239,38 +239,11 @@ async def list_messages(eval_id: str, session_id: str, session: Session = Depend @router.get("/tasks/next") async def get_next_task(session: Session = Depends(get_db)) -> dict: - """Get next pending task for OpenClaw workers. + """Get next pending task for OpenClaw workers.""" + from agenteval.intelligent_eval.task_queue import get_next_task_with_eval - Returns the highest-priority pending task, or None if no tasks available. - """ - from agenteval.intelligent_eval import task_queue - - task = task_queue.get_next_task(session) - if task is None: - return {"task": None} - - # Load eval details - from agenteval.storage.db import IntelligentEvalDB - - eval_db = session.get(IntelligentEvalDB, task.eval_id) - if eval_db is None: - return {"task": None} - - return { - "task": { - "id": task.id, - "eval_id": task.eval_id, - "priority": task.priority, - "reason": task.reason, - "eval": { - "id": eval_db.id, - "name": eval_db.name, - "status": eval_db.status, - "plan": eval_db.get_plan(), - "started_at": eval_db.started_at.isoformat() if eval_db.started_at else None, - }, - } - } + result = get_next_task_with_eval(session) + return result if result is not None else {"task": None} @router.post("/tasks/{task_id}/assign") @@ -314,69 +287,23 @@ async def create_decision_log( session: Session = Depends(get_db), ) -> dict: """Create a decision log entry for an intelligent eval.""" - # Verify eval exists - from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalDecisionLogDB + from agenteval.intelligent_eval.decision_logs import create_decision_log as _create - eval_db = session.get(IntelligentEvalDB, eval_id) - if eval_db is None: - raise HTTPException(status_code=404, detail=f"intelligent eval {eval_id} not found") - - # Create decision log - log = IntelligentEvalDecisionLogDB( - eval_id=eval_id, - decision_type=request.decision_type, - reason=request.reason, - cron_id=request.cron_id, - ) - log.set_context(request.context) - session.add(log) - session.commit() - session.refresh(log) - - return { - "id": log.id, - "eval_id": log.eval_id, - "decision_type": log.decision_type, - "reason": log.reason, - "context": log.get_context(), - "cron_id": log.cron_id, - "created_at": log.created_at.isoformat() if log.created_at else None, - } + try: + return _create(eval_id, request.decision_type, request.reason, request.cron_id, request.context, session) + except LookupError as e: + raise HTTPException(status_code=404, detail=str(e)) from e @router.get("/{eval_id}/decision-logs") async def list_decision_logs(eval_id: str, session: Session = Depends(get_db)) -> dict: """List all decision logs for an evaluation.""" - from sqlmodel import select + from agenteval.intelligent_eval.decision_logs import list_decision_logs as _list - from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalDecisionLogDB - - # Verify eval exists - eval_db = session.get(IntelligentEvalDB, eval_id) - if eval_db is None: - raise HTTPException(status_code=404, detail=f"intelligent eval {eval_id} not found") - - # Get all decision logs - logs = session.exec( - select(IntelligentEvalDecisionLogDB) - .where(IntelligentEvalDecisionLogDB.eval_id == eval_id) - .order_by(IntelligentEvalDecisionLogDB.created_at.desc()) - ).all() - - return { - "logs": [ - { - "id": log.id, - "eval_id": log.eval_id, - "decision_type": log.decision_type, - "reason": log.reason, - "context": log.get_context(), - "cron_id": log.cron_id, - "created_at": log.created_at.isoformat() if log.created_at else None, - } - for log in logs - ] - } + try: + return {"logs": _list(eval_id, session)} + except LookupError as e: + raise HTTPException(status_code=404, detail=str(e)) from e @router.get("/{eval_id}/config-snapshots") diff --git a/backend/agenteval/web/routers/openclaw_cron_pool.py b/backend/agenteval/web/routers/openclaw_cron_pool.py index d50b6b7..e6da309 100644 --- a/backend/agenteval/web/routers/openclaw_cron_pool.py +++ b/backend/agenteval/web/routers/openclaw_cron_pool.py @@ -3,11 +3,10 @@ from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field -from sqlmodel import Session, select +from sqlmodel import Session from agenteval.intelligent_eval import cron_pool from agenteval.intelligent_eval.openclaw_client import OpenClawClient -from agenteval.storage.db import OpenClawCronPoolDB, utc_now from agenteval.web.deps import get_db router = APIRouter() @@ -32,24 +31,8 @@ async def get_cron_pool_status(session: Session = Depends(get_db)) -> dict: @router.post("/cron-pool/scale") async def scale_cron_pool(request: ScaleRequest, session: Session = Depends(get_db)) -> dict: """Manually scale cron pool to target size.""" - current_status = cron_pool.get_pool_status(session) - current_size = current_status["total"] - target_size = request.target_size - client = OpenClawClient() - - if target_size > current_size: - # Scale up - count = target_size - current_size - created = await cron_pool.scale_up(count, session, client) - return {"success": True, "scaled_up": created, "current_size": current_size + created} - elif target_size < current_size: - # Scale down - count = current_size - target_size - deleted = await cron_pool.scale_down(count, session, client) - return {"success": True, "scaled_down": deleted, "current_size": current_size - deleted} - else: - return {"success": True, "current_size": current_size, "message": "already at target size"} + return await cron_pool.scale_to(request.target_size, session, client) @router.post("/cron-pool/sync") @@ -78,26 +61,11 @@ async def report_heartbeat( request: HeartbeatRequest, session: Session = Depends(get_db), ) -> dict: - """Report cron heartbeat. - - Updates the cron's last_active_at timestamp and current status. - """ - # Find cron by openclaw_cron_id - cron = session.exec( - select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == cron_id) - ).first() - - if cron is None: - raise HTTPException(status_code=404, detail=f"cron {cron_id} not found") - - # Update heartbeat - cron.last_active_at = utc_now() - cron.status = request.status - cron.current_eval_id = request.current_eval_id - cron.updated_at = utc_now() - - session.commit() - + """Report cron heartbeat. Updates last_active_at and current status.""" + try: + cron_pool.heartbeat(cron_id, request.status, request.current_eval_id, session) + except LookupError as e: + raise HTTPException(status_code=404, detail=str(e)) from e return {"success": True}