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.
72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
"""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
|
|
]
|