"""Tables of the assistant evaluation harness. Four tables, one question: *is the Feishu assistant getting better or worse, and at what.* suite ──< case the specification: what we expect │ the assistant to handle └──< run ──< result the measurement: what it did, each time we asked **Why the results are stored rather than computed on demand.** A pass rate is only meaningful next to the one before it. Recomputing it later is impossible: the corpus moves, the prompt moves, the model behind ``qwen-max-latest`` moves. A stored attempt — with the reply, the sources it actually read, and the judge's reasoning — is the only artefact that can still be argued with in three months. **Why the reply text is kept in full.** Every other assistant-adjacent ledger in this codebase stores a *shape* rather than the words (see :class:`app.assistant.gap_verification.TurnShape`), because those rows are about real employees and their questions are their own data. These rows are not: an evaluation case is a question the IT team wrote down on purpose, asked under a simulated identity, through a dry-run path that can neither send a card nor file a request. Judging an answer requires reading it, so it is kept. Mined-from-production questions are the one exception and they enter as ``draft`` — a person promotes them, and that review is where the asker's data comes out. Naive-UTC datetime discipline (codebase-wide MySQL/TiDB convention). """ from __future__ import annotations import uuid from datetime import datetime from sqlalchemy import ( JSON, Boolean, DateTime, Float, ForeignKey, Index, Integer, String, Text, UniqueConstraint, func, ) from sqlalchemy.orm import Mapped, mapped_column, relationship from app.assistant.evaluation.constants import ( CASE_DRAFT, ORIGIN_CURATED, RUN_RUNNING, SUITE_ACTIVE, TRIGGER_MANUAL, ) from app.core.database import Base from app.core.sql_types import GUID as Uuid class AssistantEvalSuite(Base): """A named bank of questions run together. Suites exist so that "the weekly regression" and "everything we have ever thought of" can be different sizes. The regression suite is small enough to run on a schedule without becoming a provider bill; the full bank is where cases accumulate. """ __tablename__ = "assistant_eval_suites" __table_args__ = ( UniqueConstraint("slug", name="uq_assistant_eval_suite_slug"), Index("idx_assistant_eval_suite_status", "status"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) #: Stable machine name. The scheduled job and the MCP tools address a suite #: by slug, so a renamed display name cannot break either. slug: Mapped[str] = mapped_column(String(64), nullable=False) name: Mapped[str] = mapped_column(String(128), nullable=False) description: Mapped[str | None] = mapped_column(Text) status: Mapped[str] = mapped_column(String(16), nullable=False, default=SUITE_ACTIVE) #: Suites seeded by a migration are marked so the UI can warn before an #: edit that the next seed run will not restore. is_builtin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) #: Who the assistant thinks it is talking to, for every case that does not #: pin its own. The assistant's tools are caller-scoped, so the persona is #: part of the measurement, not an implementation detail: the same question #: asked as a manager and as a new hire has two different correct answers. #: Pinning it at the suite is what makes runs months apart comparable — #: without it the harness silently re-baselines every time the fallback #: employee changes. default_persona_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=True, ) created_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=True, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) cases: Mapped[list[AssistantEvalCase]] = relationship( back_populates="suite", cascade="all, delete-orphan", ) class AssistantEvalCase(Base): """One question, with what a good answer to it must contain. ``setup_turns`` is what makes this able to test the thing a single-question harness cannot: whether 「这台」 in the second message still refers to the laptop named in the first. Production chat is multi-turn; an evaluation that only ever asks one question measures a product nobody uses. """ __tablename__ = "assistant_eval_cases" __table_args__ = ( Index("idx_assistant_eval_case_suite", "suite_id", "status"), Index("idx_assistant_eval_case_category", "category"), Index("idx_assistant_eval_case_origin", "origin", "status"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) suite_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("assistant_eval_suites.id"), nullable=False, ) #: Display order within the suite. Not an identity — reordering must not #: break the join from a historical result back to its case. ordinal: Mapped[int] = mapped_column(Integer, nullable=False, default=0) #: The question under test — the LAST user message of the exchange. question: Mapped[str] = mapped_column(Text, nullable=False) #: Everything said before it, as ``[{"role": "user"|"assistant", #: "content": str}, ...]``. Empty/None for the common single-turn case. setup_turns: Mapped[list | None] = mapped_column(JSON) lang: Mapped[str] = mapped_column(String(8), nullable=False, default="zh") category: Mapped[str] = mapped_column(String(32), nullable=False) status: Mapped[str] = mapped_column(String(16), nullable=False, default=CASE_DRAFT) origin: Mapped[str] = mapped_column(String(24), nullable=False, default=ORIGIN_CURATED) #: Where a machine-produced case came from, for the reviewer: a message id, #: a deflection id, a prior result id. Free text because the referents live #: in four different tables and a polymorphic FK would buy nothing. origin_ref: Mapped[str | None] = mapped_column(String(128)) #: Whose identity to simulate. NULL means "the harness picks an ordinary #: employee" — see ``runner._resolve_persona``. Pinning it matters for #: cases about somebody's own data, where the expected answer depends on #: who is asking. persona_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=True, ) #: The deterministic half of the specification. Keys, all optional: #: ``must_mention`` list[str] — substrings the reply must contain #: ``must_not_mention`` list[str] — substrings that must be absent #: ``expect_tools`` list[str] — tool names that must have run #: ``expect_action`` bool — must (not) offer a confirmation card #: ``expect_escalate`` bool — must (not) route to a ticket #: ``reference_answer`` str — what a good answer says, for the judge #: ``notes`` str — why this case exists, for the reviewer #: A case with an empty object is still valid: the judge alone can carry it. expectations: Mapped[dict | None] = mapped_column(JSON) created_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=True, ) reviewed_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=True, ) reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) suite: Mapped[AssistantEvalSuite] = relationship(back_populates="cases") class AssistantEvalRun(Base): """One execution of one suite. A run is immutable once terminal. Re-running produces a new row, which is the entire point: the sequence of rows is the trend line. """ __tablename__ = "assistant_eval_runs" __table_args__ = ( Index("idx_assistant_eval_run_suite", "suite_id", "started_at"), Index("idx_assistant_eval_run_status", "status"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) suite_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("assistant_eval_suites.id"), nullable=False, ) status: Mapped[str] = mapped_column(String(16), nullable=False, default=RUN_RUNNING) trigger: Mapped[str] = mapped_column(String(16), nullable=False, default=TRIGGER_MANUAL) triggered_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=True, ) trace_id: Mapped[str | None] = mapped_column(String(64)) repeats: Mapped[int] = mapped_column(Integer, nullable=False, default=1) concurrency: Mapped[int] = mapped_column(Integer, nullable=False, default=1) #: Which backend answered, recorded per run because a provider switch is #: the most likely explanation of a step change in the trend. provider: Mapped[str | None] = mapped_column(String(32)) model: Mapped[str | None] = mapped_column(String(64)) judge_provider: Mapped[str | None] = mapped_column(String(32)) judge_model: Mapped[str | None] = mapped_column(String(64)) cases_total: Mapped[int] = mapped_column(Integer, nullable=False, default=0) cases_run: Mapped[int] = mapped_column(Integer, nullable=False, default=0) #: Cases whose repeats did not all agree. The number that says "do not read #: the pass rate as a point estimate". cases_flaky: Mapped[int] = mapped_column(Integer, nullable=False, default=0) attempts_total: Mapped[int] = mapped_column(Integer, nullable=False, default=0) attempts_passed: Mapped[int] = mapped_column(Integer, nullable=False, default=0) attempts_failed: Mapped[int] = mapped_column(Integer, nullable=False, default=0) attempts_errored: Mapped[int] = mapped_column(Integer, nullable=False, default=0) #: passed / (passed + failed), as a percentage. Errors are excluded from #: the denominator: a provider outage is not a quality measurement, and #: counting it as failure makes an incident look like a regression. #: NULL when nothing was scored. pass_rate: Mapped[float | None] = mapped_column(Float) #: ``{dimension: mean_score}`` over scored attempts. dimension_means: Mapped[dict | None] = mapped_column(JSON) #: ``{category: {"attempts": n, "passed": n, "pass_rate": float}}``. category_breakdown: Mapped[dict | None] = mapped_column(JSON) #: ``{failure_type: count}`` over failed attempts. failure_breakdown: Mapped[dict | None] = mapped_column(JSON) #: Why the attempts that could not be scored could not be scored, e.g. #: ``{"judge_error": 60}`` (migration 595). Deliberately its own column #: rather than a key inside ``failure_breakdown``: an ``error`` is not a #: failure — that separation is what keeps a provider outage out of the #: quality number — but 「没量到」 still has a cause, and the cause decides #: who is called. A judge emitting prose instead of JSON is a harness #: defect; an unreachable provider is an outage. ``None`` on rows written #: before the column existed, which is not the same as zero. error_breakdown: Mapped[dict | None] = mapped_column(JSON) #: Why the run ended where it did — budget exhausted, cancelled, crashed. notes: Mapped[str | None] = mapped_column(Text) started_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) results: Mapped[list[AssistantEvalResult]] = relationship( back_populates="run", cascade="all, delete-orphan", ) class AssistantEvalResult(Base): """One attempt at one case inside one run. ``attempt`` is 0-based and unique within ``(run_id, case_id)``. Storing every repeat rather than a per-case summary is what makes flakiness visible; a summary row would average away the only evidence that the assistant answers the same question two different ways. """ __tablename__ = "assistant_eval_results" __table_args__ = ( UniqueConstraint( "run_id", "case_id", "attempt", name="uq_assistant_eval_result_attempt", ), Index("idx_assistant_eval_result_run", "run_id", "verdict"), Index("idx_assistant_eval_result_case", "case_id", "created_at"), Index("idx_assistant_eval_result_failure", "run_id", "failure_type"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) run_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("assistant_eval_runs.id"), nullable=False, ) #: No cascade delete from the case: retiring a question must not erase the #: history of how the assistant used to answer it. The FK is RESTRICT and #: cases are retired, never deleted, once they have results. case_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("assistant_eval_cases.id"), nullable=False, ) attempt: Mapped[int] = mapped_column(Integer, nullable=False, default=0) #: Denormalised from the case at execution time. A case whose text is later #: edited must not silently rewrite what a historical run was measuring. question: Mapped[str] = mapped_column(Text, nullable=False) category: Mapped[str] = mapped_column(String(32), nullable=False) lang: Mapped[str] = mapped_column(String(8), nullable=False, default="zh") persona_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=True, ) reply: Mapped[str | None] = mapped_column(Text) #: Tool names in call order, as the simulation reported them. tool_calls: Mapped[list | None] = mapped_column(JSON) #: ``[{id, title, url, source, updated_at}]`` — what the turn actually read, #: captured from tool results rather than from the model's prose. sources: Mapped[list | None] = mapped_column(JSON) proposed_actions: Mapped[list | None] = mapped_column(JSON) policy_conflicts: Mapped[list | None] = mapped_column(JSON) provider: Mapped[str | None] = mapped_column(String(32)) latency_ms: Mapped[int | None] = mapped_column(Integer) verdict: Mapped[str] = mapped_column(String(16), nullable=False) failure_type: Mapped[str | None] = mapped_column(String(32)) #: ``{dimension: 0..5}``. NULL when the judge never ran. scores: Mapped[dict | None] = mapped_column(JSON) #: The judge's own words. The reason a failure is diagnosable rather than #: merely counted. judge_reasoning: Mapped[str | None] = mapped_column(Text) judge_model: Mapped[str | None] = mapped_column(String(64)) #: ``[{code, severity, detail}]`` from the model-free checks. findings: Mapped[list | None] = mapped_column(JSON) #: Populated when the attempt could not be scored at all. error: Mapped[str | None] = mapped_column(Text) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) run: Mapped[AssistantEvalRun] = relationship(back_populates="results") __all__ = [ "AssistantEvalCase", "AssistantEvalResult", "AssistantEvalRun", "AssistantEvalSuite", ]