"""Executing a suite: ask, check, judge, record. One *attempt* is one case asked once. It runs the production Feishu path through :func:`~app.assistant.conversation.simulate_employee_turn`, which substitutes exactly one thing — a dry-run proposal dispatcher — so an evaluation can neither send a confirmation card to a colleague nor file a request in anyone's name. Nothing here writes to a chat, an outbox, or a transcript. Three properties this module exists to hold: **Every attempt is durable the moment it finishes.** Results are committed one at a time, in their own session, rather than gathered and written at the end. A run is minutes-to-hours of paid provider calls; losing all of it because the process was redeployed at minute fifty would make the harness something people avoid running. **A run is resumable, and resuming is the same code path.** Pending work is derived by subtracting the ``(case, attempt)`` pairs that already have rows, so re-entering an interrupted run continues it and re-entering a finished one is a no-op. This is also what lets a wall-clock budget be a normal outcome rather than a failure: stop, mark ``partial``, say what is left. **Concurrency is deliberately small and each task owns its session.** An ``AsyncSession`` is not safe to share between concurrent tasks, and the provider is the same one answering live employee questions — a regression run must never be the reason somebody's chat is slow. """ from __future__ import annotations import asyncio import logging import time import uuid from dataclasses import dataclass from datetime import UTC, datetime from typing import TYPE_CHECKING, Any, cast from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.assistant.evaluation import checks as checks_mod from app.assistant.evaluation.constants import ( ATTEMPT_TIMEOUT_SECONDS, CASE_ACTIVE, CHECK_MISSING_ACTION, CHECK_MISSING_TOOL, CHECK_UNEXPECTED_ACTION, CHECK_UNGROUNDED_CITATION, DEFAULT_RUN_BUDGET_SECONDS, DIMENSION_FAIL_AT_OR_BELOW, FAIL_ACTION, FAIL_JUDGE_ERROR, FAIL_OUT_OF_SCOPE, FAIL_RETRIEVAL, FAIL_TOOL_FIELD_OR_DATA, FAIL_UNCLASSIFIED, JUDGE_DIMENSIONS, JUDGE_HARD_DIMENSIONS, LANGS, MAX_CASES_PER_RUN, RUN_CANCELLED, RUN_FAILED, RUN_PARTIAL, RUN_RUNNING, RUN_SUCCEEDED, VERDICT_ERROR, VERDICT_FAIL, VERDICT_PASS, ) from app.assistant.evaluation.judge import judge_attempt from app.assistant.evaluation.models import ( AssistantEvalCase, AssistantEvalResult, AssistantEvalRun, AssistantEvalSuite, ) if TYPE_CHECKING: from app.core.locale import Lang logger = logging.getLogger(__name__) #: How much of a cited article's text travels with an attempt. #: #: The same list is both stored on the result row and handed to the judge, #: so this has to sit at or above the judge's own per-document budget #: (``judge._DOC_CHARS`` = 1800) — below it, capping here would quietly #: shrink what the reviewer is allowed to check the answer against, and #: nothing would say so. _STORED_BODY_CHARS = 2_000 class EvalRunError(RuntimeError): """A run could not start. Carries a code the API maps to a 4xx.""" def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code def _naive_now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) @dataclass(frozen=True) class _Attempt: case_id: uuid.UUID attempt: int # --------------------------------------------------------------------------- # Persona # --------------------------------------------------------------------------- async def resolve_persona( db: AsyncSession, *, case: AssistantEvalCase, suite: AssistantEvalSuite, ): """Whose identity this case is asked under. Order: the case's own pin, then the suite's default, then a deterministic fallback. The fallback exists so a freshly seeded suite is runnable at all, but it is the weak option and the console says so: the assistant's tools are caller-scoped, so an unpinned persona means the correct answer can change when that person's own devices, seats or manager change. Departed people are refused rather than silently substituted. The assistant gates them out, so a run under a leaver measures the leaver gate, not the assistant. """ from app.identity.person_models import IdentityPerson # noqa: PLC0415 for candidate_id in (case.persona_person_id, suite.default_persona_person_id): if candidate_id is None: continue person = await db.get(IdentityPerson, candidate_id) if person is None: raise EvalRunError( "ASSISTANT_EVAL_PERSONA_GONE", f"用例指定的模拟员工 {candidate_id} 已不存在。", ) # 🔴 The refusal the docstring above promises, actually performed. # Only the FALLBACK filtered on employment status, so a pinned persona # who had since left ran anyway — and that run measures the leaver gate # rather than the assistant, while putting a departed employee's data # into an eval row an auditor can read. if person.employment_status != "active": raise EvalRunError( "ASSISTANT_EVAL_PERSONA_DEPARTED", f"用例指定的模拟员工 {candidate_id} 已离职" f"(状态:{person.employment_status})—— 用离职的人跑,量到的是" "离职闸,不是助手。请改指定一位在职员工。", ) return person person = ( await db.execute( select(IdentityPerson) .where(IdentityPerson.employment_status == "active") .order_by(IdentityPerson.id.asc()) .limit(1) ) ).scalar_one_or_none() if person is None: raise EvalRunError( "ASSISTANT_EVAL_NO_PERSONA", "没有可用的在职员工可供模拟,无法运行评测。", ) return person # --------------------------------------------------------------------------- # One attempt # --------------------------------------------------------------------------- async def _load_source_bodies( db: AsyncSession, docs: list[dict[str, Any]], ) -> list[dict[str, Any]]: """Attach each cited article's real text to the source list. The simulation captures titles and links, which is all an administrator reading a trace needs. A judge needs the words: "does the reply agree with the document" is unanswerable from a title, and a judge asked to answer it anyway will fall back to plausibility — the exact failure this harness is built to detect. """ from app.domains.support.models import SupportKbArticle # noqa: PLC0415 ids: list[uuid.UUID] = [] for doc in docs: raw = str((doc or {}).get("id") or "").strip() if not raw: continue try: ids.append(uuid.UUID(raw)) except (TypeError, ValueError): continue if not ids: return list(docs) rows = ( await db.execute( select(SupportKbArticle.id, SupportKbArticle.body).where( SupportKbArticle.id.in_(ids) ) ) ).all() bodies = {str(row[0]): row[1] or "" for row in rows} enriched: list[dict[str, Any]] = [] for doc in docs: item = dict(doc or {}) body = bodies.get(str(item.get("id") or "")) if body: # Bounded, because this list is STORED on the result row, not just # handed to the judge. Eight uncapped articles at a few thousand # characters each, times thirty attempts, is megabytes of duplicated # KB text per run — and the article itself is one click away, so the # row only has to carry enough to recognise and check against. item["body"] = body[:_STORED_BODY_CHARS] if len(body) > _STORED_BODY_CHARS: item["body_truncated"] = True enriched.append(item) return enriched def _failure_type_from_findings(findings: list[checks_mod.Finding]) -> str | None: """Map a blocking deterministic finding onto the shared fix taxonomy. Only the unambiguous ones. A missing phrase or a wrong language does not say *why* on its own, so those are left for the judge to classify — a confident wrong bucket is worse than an empty one, because it routes the work to the wrong person. """ codes = {f.code for f in findings if f.severity == checks_mod.SEVERITY_BLOCKING} if CHECK_UNGROUNDED_CITATION in codes: # A citation to something never retrieved is the model asserting data. # It is filed as a data defect rather than a retrieval one because the # retrieval did not go wrong — it was bypassed. return FAIL_TOOL_FIELD_OR_DATA if CHECK_MISSING_TOOL in codes: return FAIL_RETRIEVAL if CHECK_UNEXPECTED_ACTION in codes or CHECK_MISSING_ACTION in codes: return FAIL_ACTION return None def combine_verdict( *, findings: list[checks_mod.Finding], judge, ) -> tuple[str, str | None]: """The two halves into one ``(verdict, failure_type)``. Precedence, highest first: 1. An ``error`` finding — the turn never happened. Not a quality signal. 2. A ``blocking`` finding — a fact about the transcript. The judge cannot vote it back up; that is the whole reason the checks run first. 3. The judge being unavailable or unusable — ``error``, never ``fail``. An unconfigured judge says nothing about the assistant, and recording it as failure would put the harness's own outage on the quality trend. 4. A hard dimension at or below the floor — enforced here rather than trusted to the judge's ``verdict`` field, so "pass" can never mean "passed while scoring groundedness 1". 5. The judge's own verdict. """ severity = checks_mod.worst_severity(findings) if severity == checks_mod.SEVERITY_ERROR: return VERDICT_ERROR, None judge_failure = getattr(judge, "failure_type", None) if severity == checks_mod.SEVERITY_BLOCKING: return VERDICT_FAIL, judge_failure or _failure_type_from_findings(findings) if judge is None or not judge.available or not judge.scores: return VERDICT_ERROR, FAIL_JUDGE_ERROR # 🔴 A hard dimension the judge did not score is an ERROR, not a free pass. # # The sentinel below is right for the SOFT dimensions — a partially scored # reply must not be failed for what it left blank. It was wrong here: the # "no usable scores" branch above only fires when EVERY dimension is # missing, so a judge that returned ``{"grounded": null, "accurate": null, # "complete": 4, …, "verdict": "pass"}`` cleared a floor whose entire # purpose is that 「pass」 cannot mean 「the model said pass while scoring # groundedness 1」. The floor was bypassable by the judge simply not # answering. Unscored is unmeasured, and unmeasured is an error. unscored = [dim for dim in JUDGE_HARD_DIMENSIONS if judge.scores.get(dim) is None] if unscored: return VERDICT_ERROR, FAIL_JUDGE_ERROR floored = [ dim for dim in JUDGE_HARD_DIMENSIONS if judge.scores.get(dim, DIMENSION_MAX_SENTINEL) <= DIMENSION_FAIL_AT_OR_BELOW ] if floored: return VERDICT_FAIL, judge_failure or FAIL_TOOL_FIELD_OR_DATA if judge.verdict == VERDICT_FAIL: return VERDICT_FAIL, judge_failure or FAIL_OUT_OF_SCOPE if judge.verdict == VERDICT_PASS: return VERDICT_PASS, None # The judge scored every dimension but never wrote a verdict. Derive one # rather than discarding a usable measurement: every hard dimension already # cleared the floor above, so this is a pass. return VERDICT_PASS, None #: The default for ``scores.get(dim, …)`` in the hard-dimension floor. #: #: It can no longer be reached: a missing hard dimension is caught one branch #: earlier and recorded as an error, because the floor cannot assert anything #: about a number nobody produced. Kept as an explicit, non-tripping default #: rather than removed — the comprehension needs one, and a default of 0 there #: would fail every attempt the instant that earlier branch is ever relaxed. DIMENSION_MAX_SENTINEL = 5 async def run_one_attempt( db: AsyncSession, *, case: AssistantEvalCase, suite: AssistantEvalSuite, run: AssistantEvalRun, attempt: int, ) -> AssistantEvalResult: """Ask once, score once, return the (unflushed) result row.""" from app.assistant import messages as canned # noqa: PLC0415 from app.assistant.conversation import simulate_employee_turn # noqa: PLC0415 from app.core.locale import resolve_lang # noqa: PLC0415 row = AssistantEvalResult( run_id=run.id, case_id=case.id, attempt=attempt, question=case.question, category=case.category, lang=case.lang, ) try: person = await resolve_persona(db, case=case, suite=suite) except EvalRunError as exc: row.verdict = VERDICT_ERROR row.failure_type = None row.error = str(exc) return row row.persona_person_id = person.id # The case's stored language is a plain column and the channel's is a # closed literal. Narrowing here rather than trusting the column keeps a # hand-edited row (or a future language) from reaching the prompt as an # unhandled value: it falls back to the persona's own setting, which is # what an employee would actually get. stored_lang = (case.lang or "").strip().lower() lang: Lang = ( cast("Lang", stored_lang) if stored_lang in LANGS else resolve_lang(person) ) try: result = await asyncio.wait_for( simulate_employee_turn( db, person=person, question=case.question, lang=lang, setup_turns=case.setup_turns or None, capture_evidence=True, ), timeout=ATTEMPT_TIMEOUT_SECONDS, ) except TimeoutError: row.verdict = VERDICT_ERROR row.error = f"这一轮超过 {ATTEMPT_TIMEOUT_SECONDS:.0f} 秒仍未返回。" return row except Exception as exc: # noqa: BLE001 - one bad case must not end the run logger.exception("eval: attempt crashed case=%s attempt=%s", case.id, attempt) row.verdict = VERDICT_ERROR row.error = f"{type(exc).__name__}: {exc}"[:2000] return row sources = await _load_source_bodies(db, result.docs or []) row.reply = result.reply or "" row.tool_calls = list(result.tool_calls or []) row.sources = sources row.proposed_actions = list(result.proposed_actions or []) row.policy_conflicts = list(result.policy_conflicts or []) row.provider = result.provider row.latency_ms = result.latency_ms findings = checks_mod.run_checks( reply=result.reply or "", tool_calls=list(result.tool_calls or []), sources=sources, proposed_actions=list(result.proposed_actions or []), policy_conflicts=list(result.policy_conflicts or []), # The pre-fetched documents are evidence too, and for a policy question # answered without a tool call they are the ONLY evidence. evidence=[*(result.evidence or []), *_docs_as_evidence(sources)], expectations=case.expectations, expected_lang=lang, turn_failed=result.turn_failed, available=result.available, canned_replies=(canned.failed(lang), canned.kernel_unavailable(lang)), ) row.findings = [f.as_dict() for f in findings] judge = None if checks_mod.worst_severity(findings) != checks_mod.SEVERITY_ERROR: try: judge = await judge_attempt( db, question=case.question, setup_turns=case.setup_turns or None, reply=result.reply or "", sources=sources, tool_calls=list(result.tool_calls or []), proposed_actions=list(result.proposed_actions or []), expectations=case.expectations, lang=lang, category=case.category, ) except Exception as exc: # noqa: BLE001 - a judge outage is not a fail logger.exception("eval: judge crashed case=%s", case.id) row.error = f"评委异常:{type(exc).__name__}: {exc}"[:2000] if judge is not None: row.scores = dict(judge.scores) or None row.judge_model = judge.model reasoning = judge.reasoning if judge.hallucinations: reasoning = ( reasoning + "\n\n无据可查的说法:\n" + "\n".join(f"- {item}" for item in judge.hallucinations) ).strip() row.judge_reasoning = reasoning or None if judge.error and not row.error: row.error = judge.error verdict, failure_type = combine_verdict(findings=findings, judge=judge) row.verdict = verdict row.failure_type = failure_type return row def _docs_as_evidence(sources: list[dict[str, Any]]) -> list[str]: """Retrieved documents rendered as evidence blobs for the citation check. A URL the server itself put in front of the model is retrieved, not invented — so it must be in the evidence set even when the model never called a tool to get it. """ blobs: list[str] = [] for doc in sources: parts = [str(doc.get(key) or "") for key in ("id", "title", "url", "body")] joined = "\n".join(part for part in parts if part) if joined: blobs.append(joined[:20_000]) return blobs # --------------------------------------------------------------------------- # One run # --------------------------------------------------------------------------- async def _pending_attempts( db: AsyncSession, *, run: AssistantEvalRun, cases: list[AssistantEvalCase], ) -> list[_Attempt]: """Everything this run still owes, ordered case-major. Case-major rather than attempt-major so an interrupted run has covered a prefix of the SUITE rather than one repeat of everything: a partial result over 30 of 60 cases is readable, one attempt of each of 60 tells you almost nothing about flakiness, which is the reason repeats exist. """ done = { (row[0], row[1]) for row in ( await db.execute( select( AssistantEvalResult.case_id, AssistantEvalResult.attempt ).where(AssistantEvalResult.run_id == run.id) ) ).all() } pending: list[_Attempt] = [] for case in cases: for attempt in range(max(1, run.repeats)): if (case.id, attempt) not in done: pending.append(_Attempt(case_id=case.id, attempt=attempt)) return pending async def _aggregate(db: AsyncSession, run: AssistantEvalRun) -> None: """Recompute every summary column on the run from its result rows. Derived rather than incremented so a resumed run's numbers are correct without any of them having to be transactionally consistent with the attempt that produced them. """ rows = ( ( await db.execute( select(AssistantEvalResult).where(AssistantEvalResult.run_id == run.id) ) ) .scalars() .all() ) passed = sum(1 for r in rows if r.verdict == VERDICT_PASS) failed = sum(1 for r in rows if r.verdict == VERDICT_FAIL) errored = sum(1 for r in rows if r.verdict == VERDICT_ERROR) scored = passed + failed run.attempts_total = len(rows) run.attempts_passed = passed run.attempts_failed = failed run.attempts_errored = errored # Errors are excluded from the denominator: a provider outage is not a # measurement of answer quality, and counting it as failure turns an # incident into what looks like a regression. run.pass_rate = round(passed * 100.0 / scored, 2) if scored else None by_case: dict[uuid.UUID, list[str]] = {} for row in rows: by_case.setdefault(row.case_id, []).append(row.verdict) run.cases_run = len(by_case) run.cases_flaky = sum( 1 for verdicts in by_case.values() if len({v for v in verdicts if v != VERDICT_ERROR}) > 1 ) dim_totals: dict[str, list[int]] = {dim: [] for dim in JUDGE_DIMENSIONS} for row in rows: for dim, value in (row.scores or {}).items(): if dim in dim_totals and isinstance(value, (int, float)): dim_totals[dim].append(int(value)) run.dimension_means = { dim: round(sum(values) / len(values), 2) for dim, values in dim_totals.items() if values } or None categories: dict[str, dict[str, Any]] = {} for row in rows: bucket = categories.setdefault( row.category, {"attempts": 0, "passed": 0, "failed": 0, "errored": 0} ) bucket["attempts"] += 1 if row.verdict == VERDICT_PASS: bucket["passed"] += 1 elif row.verdict == VERDICT_FAIL: bucket["failed"] += 1 else: bucket["errored"] += 1 for bucket in categories.values(): denom = bucket["passed"] + bucket["failed"] bucket["pass_rate"] = round(bucket["passed"] * 100.0 / denom, 2) if denom else None run.category_breakdown = categories or None failures: dict[str, int] = {} errors: dict[str, int] = {} for row in rows: if row.verdict == VERDICT_ERROR and row.failure_type: # 🔴 「没量到」 also has a cause, and it decides who is called. # # A judge that started returning prose instead of JSON turns every # attempt into an error, and so does an unreachable provider — the # first is a harness defect and the second is an outage, and they # need different people. Both used to arrive as an unexplained # ``attempts_errored`` bump with nothing anywhere saying which. # Kept in its OWN map rather than folded into ``failure_breakdown``: # an error is deliberately not a failure, and merging them would put # an outage back into the quality number one level up. errors[row.failure_type] = errors.get(row.failure_type, 0) + 1 if row.verdict == VERDICT_FAIL: # 🔴 Every failure lands in a bucket, ``unclassified`` included. # # ``_failure_type_from_findings`` deliberately returns None for the # blocking checks that do not say WHY on their own (a missing # phrase, the wrong language), and the judge — blind to the checks — # usually votes pass on those, so it contributes no type either. # Skipping the row left the run header reporting ten failures while # the panel beside it rendered 「这次没有失败的尝试。」 on the same # screen. A bucket that says "we could not classify these" is a # worse number than a clean taxonomy and a far better one than a # contradiction. # Named distinctly: ``bucket`` is already a ``dict`` in the loop # above, and reusing it makes mypy infer the wrong type for both. failure_type = row.failure_type or FAIL_UNCLASSIFIED failures[failure_type] = failures.get(failure_type, 0) + 1 run.failure_breakdown = failures or None run.error_breakdown = errors or None async def execute_run( run_id: uuid.UUID, *, budget_seconds: float | None = None, ) -> dict[str, Any]: """Run (or resume) one evaluation run to completion or to its budget. Opens its own sessions throughout — safe to call from a background task, a scheduled job, or a request that has already returned. Idempotent: attempts that already have a result row are skipped, so calling this twice on the same run finishes it once. A run already in a terminal state returns its summary untouched. """ from app.core.database import async_session_factory # noqa: PLC0415 from app.domains.support.copilot import ( # noqa: PLC0415 active_llm_identity, copilot_available, ) started = time.monotonic() async with async_session_factory() as db: run = await db.get(AssistantEvalRun, run_id) if run is None: raise EvalRunError("ASSISTANT_EVAL_RUN_NOT_FOUND", "评测运行不存在。") if run.status != RUN_RUNNING: return {"run_id": str(run.id), "status": run.status, "resumed": False} suite = await db.get(AssistantEvalSuite, run.suite_id) if suite is None: run.status = RUN_FAILED run.notes = "题库已被删除。" run.finished_at = _naive_now() await db.commit() return {"run_id": str(run.id), "status": RUN_FAILED} if not await copilot_available(db): run.status = RUN_FAILED run.notes = "助手未启用或未配置模型后端,无法评测。" run.finished_at = _naive_now() await db.commit() return {"run_id": str(run.id), "status": RUN_FAILED} provider, model = await active_llm_identity(db) run.provider = provider run.model = model run.judge_provider = provider run.judge_model = model cases = ( ( await db.execute( select(AssistantEvalCase) .where( AssistantEvalCase.suite_id == suite.id, AssistantEvalCase.status == CASE_ACTIVE, ) .order_by( AssistantEvalCase.ordinal.asc(), AssistantEvalCase.id.asc() ) .limit(MAX_CASES_PER_RUN) ) ) .scalars() .all() ) run.cases_total = len(cases) pending = await _pending_attempts(db, run=run, cases=list(cases)) await db.commit() # Plain values, not ORM instances: everything below runs after this # session is closed, and a detached instance is a lazy load waiting # to raise MissingGreenlet inside a worker. suite_id = suite.id case_count = len(cases) repeats = max(1, run.repeats) concurrency = max(1, run.concurrency) budget = ( budget_seconds if budget_seconds is not None else DEFAULT_RUN_BUDGET_SECONDS ) if not pending: return await _finalise(run_id, status=RUN_SUCCEEDED) semaphore = asyncio.Semaphore(concurrency) exhausted = asyncio.Event() completed = 0 async def _worker(item: _Attempt) -> None: nonlocal completed if exhausted.is_set(): return async with semaphore: if exhausted.is_set(): return if time.monotonic() - started > budget: exhausted.set() return # Its own session, and its own commit. An attempt that dies must # not take the run's other attempts down with it, and a result that # exists must survive the process going away a second later. async with async_session_factory() as attempt_db: try: case = await attempt_db.get(AssistantEvalCase, item.case_id) suite_row = await attempt_db.get(AssistantEvalSuite, suite_id) run_row = await attempt_db.get(AssistantEvalRun, run_id) if case is None or suite_row is None or run_row is None: return result = await run_one_attempt( attempt_db, case=case, suite=suite_row, run=run_row, attempt=item.attempt, ) attempt_db.add(result) await attempt_db.commit() completed += 1 except Exception: # noqa: BLE001 - never let one case end a run logger.exception( "eval: persisting attempt failed run=%s case=%s attempt=%s", run_id, item.case_id, item.attempt, ) await attempt_db.rollback() logger.info( "eval: run=%s starting %d attempt(s) over %d case(s) x%d, concurrency=%d", run_id, len(pending), case_count, repeats, concurrency, ) await asyncio.gather(*(_worker(item) for item in pending)) remaining = len(pending) - completed status = RUN_PARTIAL if (exhausted.is_set() or remaining > 0) else RUN_SUCCEEDED note = None if status == RUN_PARTIAL: note = ( f"预算内跑完 {completed}/{len(pending)} 次尝试,还剩 {remaining} 次未跑。" "再点一次「继续」会从断点接着跑(已完成的不会重跑)。" ) return await _finalise(run_id, status=status, note=note) async def _finalise( run_id: uuid.UUID, *, status: str, note: str | None = None, ) -> dict[str, Any]: from app.core.database import async_session_factory # noqa: PLC0415 async with async_session_factory() as db: run = await db.get(AssistantEvalRun, run_id) if run is None: return {"run_id": str(run_id), "status": RUN_FAILED} await _aggregate(db, run) # A cancellation that landed while the last attempts were finishing # wins: the operator's decision is not overwritten by the worker's. if run.status == RUN_RUNNING: # 🔴 A run that scored NOTHING did not succeed. # # ``completed`` counts attempts that PERSISTED, whatever their # verdict, so 120 attempts that all recorded ``error`` reached # ``remaining == 0`` and finished ``succeeded`` with # ``pass_rate = None``. The weekly cron only raises on ``failed``, # and the alert path reads ``drop is None`` as "within threshold" — # so a judge model that started returning prose instead of JSON, or # a provider that was never reachable, produced a green job and a # silent week. This module's own rule is that 「配置从来没生效过」 is # the failure it must not hide. scored = (run.attempts_passed or 0) + (run.attempts_failed or 0) if status == RUN_SUCCEEDED and not scored: status = RUN_FAILED note = ( ( f"{run.attempts_total} 次尝试全部是 error,一个判定都" "没拿到 —— 这不是「助手答得差」,是这次评测没有量到任何" "东西。先看模型/评委配置。" ) if run.attempts_total # ⚠️ The zero-attempt case is the SAME failure, and gating # the guard on ``attempts_total`` left it open: a suite # whose cases were all archived (or whose slug was renamed) # produced no pending attempts at all, finished # ``succeeded`` with ``pass_rate = None``, and the weekly # cron reported ok — green forever while measuring nothing, # which is the 「配置从来没生效过」 class this guard exists # for. else ( "这次运行一条用例都没跑 —— 题库里没有启用中的用例。" "评测没有量到任何东西,先看题库配置。" ) ) run.status = status run.finished_at = _naive_now() if note: run.notes = note await db.commit() return { "run_id": str(run.id), "status": run.status, "pass_rate": run.pass_rate, "attempts_total": run.attempts_total, "attempts_passed": run.attempts_passed, "attempts_failed": run.attempts_failed, "attempts_errored": run.attempts_errored, "cases_run": run.cases_run, "cases_flaky": run.cases_flaky, "notes": run.notes, } async def reap_stale_runs(db: AsyncSession, *, stale_after_seconds: float) -> int: """Close runs whose executor went away. A run executes in a background task. A redeploy kills it mid-flight and leaves a row saying ``running`` forever, which the console then shows as "in progress" indefinitely and which blocks nothing but misleads everyone. A run with results becomes ``partial`` (its numbers are real for what it reached); one with none becomes ``failed``. """ cutoff = _naive_now().timestamp() - stale_after_seconds candidates = ( ( await db.execute( select(AssistantEvalRun).where(AssistantEvalRun.status == RUN_RUNNING) ) ) .scalars() .all() ) reaped = 0 for run in candidates: last_seen = run.started_at newest = ( await db.execute( select(func.max(AssistantEvalResult.created_at)).where( AssistantEvalResult.run_id == run.id ) ) ).scalar() if newest is not None and newest > last_seen: last_seen = newest if last_seen.timestamp() > cutoff: continue await _aggregate(db, run) run.status = RUN_PARTIAL if run.attempts_total else RUN_FAILED run.finished_at = _naive_now() run.notes = ( "执行进程在中途消失(多半是发布重启),这一行由清理任务收尾。" + ( "已跑出的结果仍然有效,可以点「继续」把剩下的跑完。" if run.attempts_total else "一次尝试都没跑成。" ) ) reaped += 1 if reaped: await db.commit() return reaped async def cancel_run(db: AsyncSession, run: AssistantEvalRun) -> None: """Stop a run. In-flight attempts finish and are kept — they are paid for.""" await _aggregate(db, run) run.status = RUN_CANCELLED run.finished_at = _naive_now() run.notes = "由管理员取消。已完成的尝试保留。" __all__ = [ "EvalRunError", "cancel_run", "combine_verdict", "execute_run", "reap_stale_runs", "resolve_persona", "run_one_attempt", ]