71543f042a
将「已删即 404」语义收进 IntelligentEvalRepository 单一接缝，消除三处独立裁决； 任务监控开始隐藏已删评估的任务（本 Phase 唯一刻意行为变化）。 - repository.py 新增 visible() 谓词与 require_live_eval() 服务接缝； 六处裸谓词统一走它，get()/get_including_deleted() 语义不变。 - decision_logs.py 删除本地 _require_eval，三处调用迁至 repository 接缝； count_decisions 由 len(.all()) 改为 func.count。 - task_queue.py list_tasks 与 stats 过滤已删评估的任务（行为变化）。 - web/routers/intelligent_evals.py: _require_eval_exists → _require_live_eval， 把 LookupError 翻译为 404；expired 会话 Markdown 标注下沉至 read_model.report_markdown_by_eval；配置快照 11 字段序列化收至 config_snapshot.snapshot_to_dict 单一出口。 - AGENTS.md 登记可见性纪律（已知陷阱 #6）。 - 补 characterization 测试锁定四处契约；更新 task_queue 测试以使用 真实 eval_id（可见性过滤后字面 eval_id 不再可见）。
365 lines
13 KiB
Python
365 lines
13 KiB
Python
"""Read projections for intelligent evaluations.
|
||
|
||
The lifecycle module owns writes and state transitions. This module is the
|
||
read seam used by HTTP adapters: it turns domain entities into stable,
|
||
purpose-specific projections and keeps session-count/detail assembly out of
|
||
routers. Message bodies are intentionally not part of the detail projection;
|
||
transcripts remain available through the dedicated messages endpoint.
|
||
"""
|
||
|
||
from collections.abc import Sequence
|
||
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, Field
|
||
from sqlmodel import Session
|
||
|
||
from agenteval.intelligent_eval.domain import parse_time_slot
|
||
from agenteval.intelligent_eval.models import (
|
||
IntelligentEval,
|
||
IntelligentEvalMessage,
|
||
IntelligentEvalSession,
|
||
IntelligentEvalSessionStatus,
|
||
IntelligentEvalStatus,
|
||
)
|
||
from agenteval.intelligent_eval.report import render_report_markdown
|
||
from agenteval.intelligent_eval.repository import (
|
||
IntelligentEvalMessageRepository,
|
||
IntelligentEvalRepository,
|
||
IntelligentEvalSessionRepository,
|
||
)
|
||
from agenteval.storage.db import as_utc, utc_now
|
||
|
||
|
||
class IntelligentEvalProjection(BaseModel):
|
||
"""Fields shared by list and detail representations."""
|
||
|
||
id: str
|
||
name: str
|
||
target_id: str
|
||
status: IntelligentEvalStatus
|
||
goal: str
|
||
seeds: dict[str, Any] = Field(default_factory=dict)
|
||
intent: str
|
||
role_description: str
|
||
plan: dict[str, Any] | None = None
|
||
plan_feedback: str | None = None
|
||
time_window_hours: int
|
||
report: dict[str, Any] | None = None
|
||
created_at: datetime | None = None
|
||
updated_at: datetime | None = None
|
||
started_at: datetime | None = None
|
||
completed_at: datetime | None = None
|
||
|
||
|
||
class IntelligentEvalListItem(IntelligentEvalProjection):
|
||
"""Compact projection used by collection and mutation responses."""
|
||
|
||
session_count: int = 0
|
||
completed_sessions: int = 0
|
||
|
||
|
||
class IntelligentEvalSessionSummary(BaseModel):
|
||
"""Session metadata shown in an evaluation detail view.
|
||
|
||
This deliberately contains no chat messages. Callers that need a
|
||
transcript must use ``GET /{eval_id}/sessions/{session_id}/messages``.
|
||
"""
|
||
|
||
id: str
|
||
eval_id: str
|
||
target_id: str
|
||
persona: dict[str, Any] = Field(default_factory=dict)
|
||
goal: str
|
||
dimension: str | None = None
|
||
status: IntelligentEvalSessionStatus
|
||
verdict: dict[str, Any] | None = None
|
||
turn_count: int = 0
|
||
created_at: datetime | None = None
|
||
closed_at: datetime | None = None
|
||
|
||
|
||
class IntelligentEvalDetail(IntelligentEvalListItem):
|
||
"""Full evaluation projection, including session metadata only."""
|
||
|
||
sessions: list[IntelligentEvalSessionSummary] = Field(default_factory=list)
|
||
|
||
|
||
def _require_id(eval_obj: IntelligentEval) -> str:
|
||
if eval_obj.id is None:
|
||
raise ValueError("intelligent evaluation projection requires an id")
|
||
return eval_obj.id
|
||
|
||
|
||
def _base_fields(eval_obj: IntelligentEval) -> dict[str, Any]:
|
||
"""Copy domain data into the stable projection field set."""
|
||
|
||
return {
|
||
"id": _require_id(eval_obj),
|
||
"name": eval_obj.name,
|
||
"target_id": eval_obj.target_id,
|
||
"status": eval_obj.status,
|
||
"goal": eval_obj.goal,
|
||
"seeds": eval_obj.seeds,
|
||
"intent": eval_obj.intent,
|
||
"role_description": eval_obj.role_description,
|
||
"plan": eval_obj.plan,
|
||
"plan_feedback": eval_obj.plan_feedback,
|
||
"time_window_hours": eval_obj.time_window_hours,
|
||
"report": eval_obj.report,
|
||
"created_at": eval_obj.created_at,
|
||
"updated_at": eval_obj.updated_at,
|
||
"started_at": eval_obj.started_at,
|
||
"completed_at": eval_obj.completed_at,
|
||
}
|
||
|
||
|
||
def _session_summary(session_obj: IntelligentEvalSession) -> IntelligentEvalSessionSummary:
|
||
if session_obj.id is None:
|
||
raise ValueError("intelligent evaluation session projection requires an id")
|
||
return IntelligentEvalSessionSummary.model_validate(session_obj.model_dump())
|
||
|
||
|
||
def project_list_item(
|
||
eval_obj: IntelligentEval,
|
||
sessions: Sequence[IntelligentEvalSession],
|
||
) -> IntelligentEvalListItem:
|
||
"""Build a list projection from an evaluation and its session metadata."""
|
||
|
||
return IntelligentEvalListItem(
|
||
**_base_fields(eval_obj),
|
||
session_count=len(sessions),
|
||
completed_sessions=sum(1 for item in sessions if item.status is IntelligentEvalSessionStatus.COMPLETED),
|
||
)
|
||
|
||
|
||
def project_detail(
|
||
eval_obj: IntelligentEval,
|
||
sessions: Sequence[IntelligentEvalSession],
|
||
) -> IntelligentEvalDetail:
|
||
"""Build a detail projection without embedding transcript messages."""
|
||
|
||
item = project_list_item(eval_obj, sessions)
|
||
return IntelligentEvalDetail(
|
||
**item.model_dump(),
|
||
sessions=[_session_summary(item) for item in sessions],
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Execution progress (执行过程视图)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class ExecutionSlotProgress(BaseModel):
|
||
"""Plan-vs-actual progress for one time-distribution slot."""
|
||
|
||
time_slot: str
|
||
planned: int = 0
|
||
created: int = 0
|
||
completed: int = 0
|
||
is_current: bool = False
|
||
is_past: bool = False
|
||
|
||
|
||
class ExecutionProgress(BaseModel):
|
||
"""Server-authoritative "where is it stuck and why" projection.
|
||
|
||
current_stage uses the four-step axis: planning / approval / executing /
|
||
done. cancelled/failed never appear as a stage; they surface as
|
||
abnormal_outcome with the stage marking the node where the eval stopped.
|
||
"""
|
||
|
||
current_stage: str
|
||
abnormal_outcome: str | None = None
|
||
blocker: str | None = None
|
||
next_action: str | None = None
|
||
slots: list[ExecutionSlotProgress] = Field(default_factory=list)
|
||
|
||
|
||
def _naive(moment: datetime) -> datetime:
|
||
return as_utc(moment).replace(tzinfo=None)
|
||
|
||
|
||
def _derive_stage(eval_obj: IntelligentEval) -> tuple[str, str | None]:
|
||
"""Map the 7-state machine onto the 4-step axis; abnormal outcomes mark the node."""
|
||
|
||
status = eval_obj.status
|
||
if status in (IntelligentEvalStatus.DRAFT, IntelligentEvalStatus.PLANNING):
|
||
return "planning", None
|
||
if status is IntelligentEvalStatus.PENDING_APPROVAL:
|
||
return "approval", None
|
||
if status is IntelligentEvalStatus.EXECUTING:
|
||
return "executing", None
|
||
if status is IntelligentEvalStatus.COMPLETED:
|
||
return "done", None
|
||
abnormal = status.value # cancelled / failed
|
||
if eval_obj.plan is None:
|
||
return "planning", abnormal
|
||
if eval_obj.started_at is None:
|
||
return "approval", abnormal
|
||
return "executing", abnormal
|
||
|
||
|
||
def _project_slots(
|
||
eval_obj: IntelligentEval,
|
||
sessions: Sequence[IntelligentEvalSession],
|
||
now: datetime,
|
||
) -> list[ExecutionSlotProgress]:
|
||
"""Bucket sessions into plan slots by created_at, mirroring domain.py 口径."""
|
||
|
||
if not eval_obj.plan or eval_obj.started_at is None:
|
||
return []
|
||
time_distribution = eval_obj.plan.get("time_distribution", [])
|
||
if not isinstance(time_distribution, list):
|
||
return []
|
||
started = _naive(eval_obj.started_at)
|
||
offset_hours = (_naive(now) - started).total_seconds() / 3600
|
||
slots: list[ExecutionSlotProgress] = []
|
||
for slot in time_distribution:
|
||
parsed = parse_time_slot(str(slot.get("time_slot", "")))
|
||
if parsed is None:
|
||
continue
|
||
start_hour, end_hour = parsed
|
||
window_start = started + timedelta(hours=start_hour)
|
||
window_end = started + timedelta(hours=end_hour)
|
||
in_window = [
|
||
item
|
||
for item in sessions
|
||
if item.created_at is not None and window_start <= _naive(item.created_at) < window_end
|
||
]
|
||
slots.append(
|
||
ExecutionSlotProgress(
|
||
time_slot=str(slot.get("time_slot", "")),
|
||
planned=int(slot.get("sessions", 0) or 0),
|
||
created=len(in_window),
|
||
completed=sum(1 for item in in_window if item.status is IntelligentEvalSessionStatus.COMPLETED),
|
||
is_current=start_hour <= offset_hours < end_hour,
|
||
is_past=offset_hours >= end_hour,
|
||
)
|
||
)
|
||
return slots
|
||
|
||
|
||
def project_execution_progress(
|
||
eval_obj: IntelligentEval,
|
||
sessions: Sequence[IntelligentEvalSession],
|
||
now: datetime | None = None,
|
||
) -> ExecutionProgress:
|
||
"""Derive stage, blocker and next action entirely server-side."""
|
||
|
||
moment = now or utc_now()
|
||
stage, abnormal = _derive_stage(eval_obj)
|
||
slots = _project_slots(eval_obj, sessions, moment)
|
||
blocker: str | None = None
|
||
next_action: str | None = None
|
||
|
||
if abnormal == "cancelled":
|
||
blocker = "评估已取消"
|
||
elif abnormal == "failed":
|
||
blocker = f"评估已失败：{eval_obj.plan_feedback}" if eval_obj.plan_feedback else "评估已失败"
|
||
elif stage == "planning":
|
||
next_action = (
|
||
"等待 OpenClaw 规划师提交粗计划"
|
||
if eval_obj.status is IntelligentEvalStatus.PLANNING
|
||
else "提交评估后进入 AI 规划"
|
||
)
|
||
elif stage == "approval":
|
||
blocker = "等待人工审批粗计划"
|
||
next_action = "审批通过后开始执行"
|
||
elif stage == "executing":
|
||
due_planned = sum(item.planned for item in slots if item.is_past or item.is_current)
|
||
deficit = max(0, due_planned - sum(item.created for item in slots))
|
||
running = sum(1 for item in sessions if item.status is IntelligentEvalSessionStatus.RUNNING)
|
||
if deficit > 0:
|
||
next_action = f"等待平台触发 worker 补足当前欠账 {deficit} 个会话"
|
||
elif running > 0:
|
||
next_action = f"等待 {running} 个进行中的会话完成"
|
||
elif any(not item.is_past for item in slots):
|
||
next_action = "当前时段计划已完成，等待下一时段到期"
|
||
else:
|
||
next_action = "等待 analyst 汇总评估报告"
|
||
|
||
return ExecutionProgress(
|
||
current_stage=stage,
|
||
abnormal_outcome=abnormal,
|
||
blocker=blocker,
|
||
next_action=next_action,
|
||
slots=slots,
|
||
)
|
||
|
||
|
||
class IntelligentEvalReadModel:
|
||
"""Read interface for stable intelligent-evaluation projections.
|
||
|
||
The current implementation uses the existing session repository. Query
|
||
batching can be added behind this seam without changing Router callers.
|
||
"""
|
||
|
||
def __init__(self, session: Session):
|
||
self._evals = IntelligentEvalRepository(session)
|
||
self._sessions = IntelligentEvalSessionRepository(session)
|
||
self._messages = IntelligentEvalMessageRepository(session)
|
||
|
||
def list_item(self, eval_obj: IntelligentEval) -> IntelligentEvalListItem:
|
||
return project_list_item(eval_obj, self._sessions.list_by_eval(_require_id(eval_obj)))
|
||
|
||
def list_items(self, evals: Sequence[IntelligentEval]) -> list[IntelligentEvalListItem]:
|
||
"""Project a collection after one batched session lookup."""
|
||
|
||
sessions_by_eval = self._sessions.list_by_evals([_require_id(item) for item in evals])
|
||
return [project_list_item(item, sessions_by_eval[item.id]) for item in evals if item.id is not None]
|
||
|
||
def detail_by_id(self, eval_id: str) -> IntelligentEvalDetail | None:
|
||
"""Read one evaluation and its summaries with one snapshot query."""
|
||
|
||
snapshot = self._evals.get_with_sessions(eval_id)
|
||
return project_detail(*snapshot) if snapshot is not None else None
|
||
|
||
def execution_progress_by_id(self, eval_id: str) -> ExecutionProgress | None:
|
||
"""Project the execution-process view (stage / blocker / slots)."""
|
||
|
||
snapshot = self._evals.get_with_sessions(eval_id)
|
||
return project_execution_progress(*snapshot) if snapshot is not None else None
|
||
|
||
def sessions_by_eval(self, eval_id: str) -> list[IntelligentEvalSession] | None:
|
||
"""Return session summaries, or ``None`` when the parent is unknown."""
|
||
|
||
if self._evals.get(eval_id) is None:
|
||
return None
|
||
return self._sessions.list_by_eval(eval_id)
|
||
|
||
def messages_by_session(self, eval_id: str, session_id: str) -> list[IntelligentEvalMessage] | None:
|
||
"""Read a transcript only when the session belongs to the evaluation."""
|
||
|
||
if self._evals.get(eval_id) is None:
|
||
return None
|
||
session_obj = self._sessions.get(session_id)
|
||
if session_obj is None or session_obj.eval_id != eval_id:
|
||
return None
|
||
return self._messages.list_by_session(session_id)
|
||
|
||
def report_by_eval(self, eval_id: str) -> tuple[str, dict[str, Any]] | None:
|
||
"""Return the report payload with its evaluation name for renderers."""
|
||
|
||
eval_obj = self._evals.get(eval_id)
|
||
if eval_obj is None or eval_obj.report is None:
|
||
return None
|
||
return eval_obj.name, eval_obj.report
|
||
|
||
def report_markdown_by_eval(self, eval_id: str) -> str | None:
|
||
"""报告的 Markdown 投影（含 expired 会话的不完整证据标注，ADR-0011）。"""
|
||
|
||
report = self.report_by_eval(eval_id)
|
||
if report is None:
|
||
return None
|
||
name, payload = report
|
||
markdown = render_report_markdown(payload, name=name, eval_id=eval_id)
|
||
expired = [s for s in self._sessions.list_by_eval(eval_id) if s.status is IntelligentEvalSessionStatus.EXPIRED]
|
||
if expired:
|
||
lines = ["", "## 不完整证据会话", ""]
|
||
for s in expired:
|
||
lines.append(f"- 会话 `{s.id}`（{s.turn_count} 轮）：60 分钟无新轮次过期，证据不完整")
|
||
markdown = markdown.rstrip() + "\n" + "\n".join(lines) + "\n"
|
||
return markdown
|