appdb/feedback.py
Rama main · réplica de lectura del repositorio original.
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Wrong-answer feedback and its triage -- admin panel phase 4.
``docs/admin-panel-architecture.md`` §3's tier 1 ("closes a loop nothing
else closes") and the frozen phase 4 spec are the design contract. This
module is the storage layer -- the same split ``appdb.roles``/
``appdb.key_store`` keep from the routes that call them: authorisation
(who may flag which turn, who may triage) is enforced by the caller
(``api/v2_routes.py`` for flagging, ``api/admin_feedback_routes.py`` for
triage), not here.
The join, not a copy (spec §2.2)
---------------------------------
A feedback row never carries the question, the generated SQL, or any
result data. Those already exist, keyed by the same ``session_id``/
``turn_id``, in the analyst audit log (``observability/audit.py``) --
phase 4.2.0 added those two fields to :class:`~observability.audit.AuditRecord`
precisely so this join is possible. :func:`submit_flag` resolves
``request_id`` and ``config_version_id`` from that same audit record *at
flag time* (never supplied by the client) and stores only those two
identifiers -- the same "an id, never the content it names" category
``AuditRecord.session_id`` already is. Triage reads (the admin routes, not
this module) perform the actual join for display.
Nothing here auto-applies (spec §3.2)
----------------------------------------
:func:`resolve_feedback` never creates, applies, or even touches a
``config_bundle_versions`` row -- for an ``"alias_fix"``/``"rule_fix"``
outcome, the admin makes the actual domain-knowledge edit through the
*existing*, unchanged phase 3 surface (``POST /admin/config/versions``),
which already runs it through validation, diff, dry-run and (for
``schema.yaml``) the operations/security draft split. Resolving a flag
only records *that* a fix was made and, optionally, which config version
id it was -- so "nothing auto-applies" is not a rule this module has to
be careful to uphold, it is a fact about what this module is capable of
doing at all: it has no code path that writes ``project_config/`` content
anywhere.
Promotion to a golden case (spec §4)
---------------------------------------
The golden set (``eval_data/golden.jsonl``) is a file outside the
application database and outside phase 3's versioning (architecture §3
Tier 1, "the golden set has no home in this design"). :func:`promote_to_golden_case`
resolves this the way the spec directs: the case itself is appended to
that file (the same file :mod:`eval.runner`/CI already read, so it never
diverges into a second store), written ``status="pending_expected"``
(:data:`eval.models.GoldenCaseStatus`) since the flagged answer was wrong
and there is, by definition, no confirmed expectation yet -- and this
module records only *that* the promotion happened, by whom, and which
case id resulted.
"""
from __future__ import annotations
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from sqlalchemy import delete, select
import config as cfg
from appdb.engine import get_app_engine
from appdb.models import turn_feedback
from observability.audit import find_record_by_turn
#: The closed set of categories an analyst may pick when flagging an
#: answer (spec §2.1). Free text is additionally offered, but only
#: alongside one of these, never instead of one -- see
#: ``web/js/render/feedback.js``.
FEEDBACK_CATEGORIES: tuple[str, ...] = (
"wrong_number",
"different_question",
"wrong_filter_or_period",
"other",
)
#: The closed set of triage outcomes (spec §3.1). Every flag must resolve
#: to exactly one of these -- there is no "dismiss without a reason".
RESOLUTION_OUTCOMES: tuple[str, ...] = (
"alias_fix",
"rule_fix",
"golden_case",
"not_a_defect",
)
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
class TurnNotAuditedError(LookupError):
"""No audit record names the given ``session_id``/``turn_id`` (see
:func:`~observability.audit.find_record_by_turn`) -- there is nothing
to join this flag against, so it is refused rather than stored with a
silently missing ``request_id``."""
class FeedbackNotFoundError(LookupError):
"""No ``turn_feedback`` row matches the given ``feedback_id``."""
class AlreadyResolvedError(RuntimeError):
""":func:`resolve_feedback` called on a flag whose ``status`` is
already ``"resolved"`` -- resolution is one-way, matching the
triage queue's own "every outcome writes a record naming who resolved
it and how" (spec §3.1): a second resolution would either silently
overwrite the first admin's decision or need its own audit trail to
avoid doing so, and neither is worth the complexity for what has never
been asked for."""
def _public(row: dict[str, Any]) -> dict[str, Any]:
return {
"feedback_id": row["feedback_id"],
"session_id": row["session_id"],
"turn_id": row["turn_id"],
"request_id": row["request_id"],
"reporter_principal_id": row["reporter_principal_id"],
"category": row["category"],
"note": row["note"],
"config_version_id": row["config_version_id"],
"created_at": row["created_at"],
"status": row["status"],
"resolution_outcome": row["resolution_outcome"],
"resolution_note": row["resolution_note"],
"resolution_config_version_id": row["resolution_config_version_id"],
"resolution_golden_case_id": row["resolution_golden_case_id"],
"resolved_by": row["resolved_by"],
"resolved_at": row["resolved_at"],
}
# ---------------------------------------------------------------------------
# Submit (analyst-facing; ownership is the caller's job -- see module
# docstring)
# ---------------------------------------------------------------------------
def submit_flag(
*,
session_id: str,
turn_id: str,
reporter_principal_id: str,
category: str,
note: str = "",
) -> dict[str, Any]:
"""Record one wrong-answer flag against *session_id*/*turn_id*.
Parameters
----------
session_id, turn_id:
The turn being flagged. Ownership (only the turn's own principal
may flag it) is the caller's responsibility -- see module
docstring; this function stores whatever it is given.
reporter_principal_id:
The flagging principal's id.
category:
One of :data:`FEEDBACK_CATEGORIES`.
note:
Optional free text (spec §2.1's "one optional question").
Returns
-------
dict
The newly created row's public representation.
Raises
------
ValueError
*category* is not one of :data:`FEEDBACK_CATEGORIES`.
TurnNotAuditedError
No audit record joins to *session_id*/*turn_id* -- see
:func:`~observability.audit.find_record_by_turn`.
"""
if category not in FEEDBACK_CATEGORIES:
raise ValueError(
f"category must be one of {FEEDBACK_CATEGORIES!r}, got {category!r}"
)
audit_record = find_record_by_turn(session_id, turn_id)
if audit_record is None:
raise TurnNotAuditedError(
f"no audit record names session_id={session_id!r} turn_id={turn_id!r} "
"-- nothing to join this flag against"
)
now = _now_iso()
engine = get_app_engine()
with engine.begin() as conn:
result = conn.execute(
turn_feedback.insert().values(
session_id=session_id,
turn_id=turn_id,
request_id=audit_record.get("request_id"),
reporter_principal_id=reporter_principal_id,
category=category,
note=note or None,
config_version_id=audit_record.get("config_version_id"),
created_at=now,
status="open",
resolution_outcome=None,
resolution_note=None,
resolution_config_version_id=None,
resolution_golden_case_id=None,
resolved_by=None,
resolved_at=None,
)
)
feedback_id = result.inserted_primary_key[0]
return get_feedback(feedback_id)
# ---------------------------------------------------------------------------
# Read
# ---------------------------------------------------------------------------
def get_feedback(feedback_id: int) -> dict[str, Any]:
"""One flag's stored row (never the joined question/SQL -- see
``api.admin_feedback_routes`` for the endpoint that adds those).
Raises
------
FeedbackNotFoundError
"""
engine = get_app_engine()
with engine.connect() as conn:
row = conn.execute(
select(turn_feedback).where(turn_feedback.c.feedback_id == feedback_id)
).mappings().first()
if row is None:
raise FeedbackNotFoundError(f"no feedback row {feedback_id!r}")
return _public(dict(row))
def list_feedback(
*, status: str | None = None, session_id: str | None = None,
) -> list[dict[str, Any]]:
"""Every flag matching the given filters, newest first.
Parameters
----------
status:
``"open"`` or ``"resolved"`` to filter, ``None`` (default) for
every flag regardless of status -- the triage queue's own default
view (spec §3: "newest first").
session_id:
Restrict to flags on one session -- used by the analyst-facing
"did I already flag this turn" read (``GET
/v2/sessions/{sid}/turns/{tid}/feedback``), never by the admin
triage queue, which deliberately shows every principal's flags.
"""
stmt = select(turn_feedback).order_by(turn_feedback.c.feedback_id.desc())
if status is not None:
stmt = stmt.where(turn_feedback.c.status == status)
if session_id is not None:
stmt = stmt.where(turn_feedback.c.session_id == session_id)
engine = get_app_engine()
with engine.connect() as conn:
rows = conn.execute(stmt).mappings().all()
return [_public(dict(row)) for row in rows]
def withdraw_open_flags(
*, session_id: str, turn_id: str, reporter_principal_id: str,
) -> int:
"""Delete the still-open flags *reporter_principal_id* raised on a turn.
The flag half of "the later action wins": when a principal confirms an
answer they had flagged (:mod:`appdb.confirmations`), the earlier flag
is removed so the turn does not sit in the triage queue under a verdict
its reporter has taken back. Only ``status == "open"`` rows go. A flag
an admin has already resolved stays: triage recorded a decision on it
(possibly a config version or a golden case id, see
:func:`resolve_feedback`), and deleting the row would orphan that
record. Another principal's flags on the same turn are never touched.
Parameters
----------
session_id, turn_id:
The turn.
reporter_principal_id:
Whose open flags to delete.
Returns
-------
int
How many flags were deleted (usually 0 or 1).
"""
engine = get_app_engine()
with engine.begin() as conn:
result = conn.execute(
delete(turn_feedback).where(
turn_feedback.c.session_id == session_id,
turn_feedback.c.turn_id == turn_id,
turn_feedback.c.reporter_principal_id == reporter_principal_id,
turn_feedback.c.status == "open",
)
)
return int(result.rowcount or 0)
# ---------------------------------------------------------------------------
# Triage (spec §3)
# ---------------------------------------------------------------------------
def resolve_feedback(
feedback_id: int,
*,
outcome: str,
actor_principal_id: str,
note: str = "",
config_version_id: int | None = None,
golden_case_id: str | None = None,
) -> dict[str, Any]:
"""Resolve flag *feedback_id* into exactly one outcome (spec §3.1).
This function never creates or applies a configuration version, and
never writes to the golden-set file -- see module docstring. For
``"alias_fix"``/``"rule_fix"``, *config_version_id* is purely
provenance: the id of a ``config_bundle_versions`` row the admin
already created through ``POST /admin/config/versions`` (unchanged by
this phase). For ``"golden_case"``, *golden_case_id* is the id
:func:`promote_to_golden_case` assigned when writing the case to
``eval_data/golden.jsonl`` -- callers resolve with that outcome only
after calling that function (see ``api.admin_feedback_routes`` for the
single request that does both).
Parameters
----------
feedback_id:
The flag to resolve.
outcome:
One of :data:`RESOLUTION_OUTCOMES`.
actor_principal_id:
The triaging admin's id.
note:
Required (non-blank) for ``"not_a_defect"`` (spec §3.1: "recorded
with a reason, not silently dropped"); optional otherwise.
config_version_id, golden_case_id:
See above -- provenance only, never validated against the config
version table or the golden file by this function.
Returns
-------
dict
The resolved row's public representation.
Raises
------
ValueError
*outcome* is not one of :data:`RESOLUTION_OUTCOMES`, or
*outcome* is ``"not_a_defect"`` and *note* is blank.
FeedbackNotFoundError
AlreadyResolvedError
"""
if outcome not in RESOLUTION_OUTCOMES:
raise ValueError(
f"outcome must be one of {RESOLUTION_OUTCOMES!r}, got {outcome!r}"
)
if outcome == "not_a_defect" and not note.strip():
raise ValueError(
"a 'not_a_defect' outcome requires a non-blank note -- recorded "
"with a reason, never silently dropped (spec §3.1)"
)
engine = get_app_engine()
with engine.begin() as conn:
row = conn.execute(
select(turn_feedback).where(turn_feedback.c.feedback_id == feedback_id)
).mappings().first()
if row is None:
raise FeedbackNotFoundError(f"no feedback row {feedback_id!r}")
if row["status"] == "resolved":
raise AlreadyResolvedError(
f"feedback {feedback_id!r} was already resolved as "
f"{row['resolution_outcome']!r} by {row['resolved_by']!r} at "
f"{row['resolved_at']!r}"
)
conn.execute(
turn_feedback.update()
.where(turn_feedback.c.feedback_id == feedback_id)
.values(
status="resolved",
resolution_outcome=outcome,
resolution_note=note or None,
resolution_config_version_id=config_version_id,
resolution_golden_case_id=golden_case_id,
resolved_by=actor_principal_id,
resolved_at=_now_iso(),
)
)
return get_feedback(feedback_id)
# ---------------------------------------------------------------------------
# Promotion to a golden case (spec §4)
# ---------------------------------------------------------------------------
def promote_to_golden_case(
feedback_id: int, *, tags: list[str] | None = None,
) -> dict[str, Any]:
"""Append a new, ``"pending_expected"`` golden case for *feedback_id*'s
turn to ``cfg.settings.eval_golden_path`` (spec §4).
The question is read from the audit log by re-joining on the flag's
own ``session_id``/``turn_id`` (never stored on the feedback row
itself -- see module docstring), so this is safe to call even long
after :func:`submit_flag` ran, as long as the audit log has not
rotated past that turn. ``expected_sql``/``expected_fingerprint`` are
left unset: the flagged answer was wrong, so there is, by definition,
no confirmed expectation yet -- someone must supply one and flip the
case's ``status`` back to ``"active"`` before the regression gate will
ever run it (:func:`eval.runner.run_golden_set` skips a
``"pending_expected"`` case entirely).
Parameters
----------
feedback_id:
The flag being promoted.
tags:
Optional tags for the new case's per-tag accuracy breakdown once
it is active. Defaults to none.
Returns
-------
dict
``{"case_id": str, "question": str}`` for the newly appended case.
Raises
------
FeedbackNotFoundError
TurnNotAuditedError
The flag's turn can no longer be joined to an audit record.
"""
from eval.models import GoldenCase
feedback = get_feedback(feedback_id)
audit_record = find_record_by_turn(feedback["session_id"], feedback["turn_id"])
if audit_record is None:
raise TurnNotAuditedError(
f"no audit record names session_id={feedback['session_id']!r} "
f"turn_id={feedback['turn_id']!r} -- cannot promote without the "
"question it asked"
)
case_id = f"feedback_{feedback_id}"
case = GoldenCase(
id=case_id,
question=audit_record["question"],
tags=list(tags or []),
status="pending_expected",
notes=(
f"Promoted from feedback #{feedback_id} "
f"(session {feedback['session_id']!r}, turn {feedback['turn_id']!r}); "
"awaiting a confirmed expected_sql/expected_fingerprint."
),
)
golden_path = Path(cfg.settings.eval_golden_path)
golden_path.parent.mkdir(parents=True, exist_ok=True)
with golden_path.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(case.to_dict(), ensure_ascii=False))
fh.write("\n")
return {"case_id": case_id, "question": case.question}
# ---------------------------------------------------------------------------
# Closing the loop visibly (spec §5)
# ---------------------------------------------------------------------------
def feedback_stats() -> dict[str, Any]:
"""Aggregate counts for the panel's "closing the loop visibly" view
(spec §5): total/open flags, outcomes by category, and the golden
set's size split by :data:`~eval.models.GoldenCaseStatus`.
Never re-runs the evaluation harness -- accuracy trend is the admin
routes' job (``api.admin_feedback_routes``), reading whatever baseline
``python -m eval.cli run --save-baseline`` last recorded
(:mod:`eval.baseline`), since running the harness live against a real
endpoint on every panel load would be exactly the "expensive, and
surprising" thing this read-only view must not do.
"""
rows = list_feedback()
open_count = sum(1 for r in rows if r["status"] == "open")
by_outcome: dict[str, int] = {outcome: 0 for outcome in RESOLUTION_OUTCOMES}
for r in rows:
if r["resolution_outcome"]:
by_outcome[r["resolution_outcome"]] = by_outcome.get(r["resolution_outcome"], 0) + 1
golden_path = Path(cfg.settings.eval_golden_path)
golden_total = 0
golden_pending = 0
if golden_path.exists():
from eval.runner import load_golden_cases
try:
cases = load_golden_cases(golden_path)
except ValueError:
cases = []
golden_total = len(cases)
golden_pending = sum(1 for c in cases if c.status == "pending_expected")
return {
"flags_total": len(rows),
"flags_open": open_count,
"outcomes_by_category": by_outcome,
"golden_set_size": golden_total,
"golden_set_pending": golden_pending,
}