# SPDX-License-Identifier: BUSL-1.1 # Copyright (c) 2024-2026 Ali Sadeghi Aghili """Analyst confirmations: "this answer is correct" (پاسخ درست است). The positive twin of :mod:`appdb.feedback`. An analyst who has just seen a correct answer confirms it with one click; each confirmation becomes evidence in two places: a reviewed golden-set candidate (``scripts/harvest_golden.py``) and the real-use quality report (:mod:`observability.quality`). This module is the storage layer, split from its callers the same way :mod:`appdb.feedback` is: authorisation (only the turn's own principal may confirm it) is enforced by ``api/v2_routes.py``, not here. Identifiers only ----------------- A row never carries the question, the generated SQL or any result data -- the rule :mod:`appdb.feedback` states in its "The join, not a copy" section. Those already exist, keyed by the same ``session_id``/``turn_id``, in the analyst audit log. :func:`confirm_turn` reads that record to decide whether the turn may be confirmed at all, resolves ``request_id`` from it (never from the client), and stores only the identifiers: ``session_id``, ``turn_id``, ``principal_id``, ``request_id`` and ``created_at``. Rules ----- * **Only a successful answer can be confirmed.** The turn's audit record must show SQL that ran and returned a result (:func:`observability.audit.turn_executed_successfully`). A rejected, declined or failed turn, and a turn with no audit record to check, raise :class:`TurnNotConfirmableError`. * **At most one confirmation per (turn, principal).** Enforced by a unique constraint on the table and handled here, so two concurrent clicks store one row. Confirming again is idempotent: the existing row is returned and nothing changes (``created_at`` keeps the first click's time). * **A principal withdraws only their own.** :func:`withdraw_confirmation` is keyed by the caller's principal id; another principal's confirmation of the same turn is never touched. The row is deleted -- nothing downstream needs the history of a click that was taken back. * **Flag and confirmation exclude each other, per principal.** A principal who confirms a turn they had flagged, or flags one they had confirmed, has changed their mind: the later action wins and the earlier one is removed. The routes in ``api/v2_routes.py`` apply this with :func:`withdraw_confirmation` and :func:`appdb.feedback.withdraw_open_flags`; it is per principal, so two analysts disagreeing about one turn leave one confirmation and one flag. """ from __future__ import annotations from datetime import datetime, timezone from typing import Any from sqlalchemy import delete, select from sqlalchemy.exc import IntegrityError from appdb.engine import get_app_engine from appdb.models import turn_confirmations from observability.audit import find_record_by_turn, turn_executed_successfully def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() class TurnNotConfirmableError(RuntimeError): """The turn cannot be confirmed: its audit record does not show a successful execution, or there is no audit record to check. Attributes ---------- reason: ``"not_audited"`` (no record names the turn -- e.g. the audit log has since rotated past it) or ``"not_successful"`` (the record shows a rejection, a decline, an error, or no result). """ def __init__(self, message: str, *, reason: str) -> None: super().__init__(message) self.reason = reason def _public(row: dict[str, Any]) -> dict[str, Any]: return { "confirmation_id": row["confirmation_id"], "session_id": row["session_id"], "turn_id": row["turn_id"], "principal_id": row["principal_id"], "request_id": row["request_id"], "created_at": row["created_at"], } def get_confirmation( *, session_id: str, turn_id: str, principal_id: str, ) -> dict[str, Any] | None: """The *principal_id*'s confirmation of the turn, or ``None``. Parameters ---------- session_id, turn_id: The turn. principal_id: Whose confirmation to look up. Returns ------- dict | None The row's public representation. """ engine = get_app_engine() with engine.connect() as conn: row = conn.execute( select(turn_confirmations).where( turn_confirmations.c.session_id == session_id, turn_confirmations.c.turn_id == turn_id, turn_confirmations.c.principal_id == principal_id, ) ).mappings().first() return None if row is None else _public(dict(row)) def confirm_turn( *, session_id: str, turn_id: str, principal_id: str, ) -> tuple[dict[str, Any], bool]: """Record that *principal_id* confirmed *session_id*/*turn_id* as correct. Idempotent: when this principal has already confirmed the turn, the existing row is returned and nothing is written. Ownership (only the turn's own principal may confirm it) is the caller's responsibility -- see the module docstring; this function stores whatever principal it is given. Parameters ---------- session_id, turn_id: The turn being confirmed. principal_id: The confirming principal's id. Returns ------- tuple[dict, bool] ``(row, created)``: the row's public representation, and whether this call created it (``False`` when it already existed). Raises ------ TurnNotConfirmableError The audit log has no record for the turn, or the record does not show a successful execution. Examples -------- >>> import tempfile, os >>> import config as cfg >>> from datetime import datetime >>> import observability.audit as audit >>> from appdb.engine import dispose_app_engine >>> d = tempfile.mkdtemp() >>> audit._AUDIT_LOG_FILE = os.path.join(d, "audit_log.jsonl") >>> audit.save_audit_record(audit.AuditRecord( ... timestamp=datetime(2026, 1, 1), request_id="r_1", question="q", ... generated_sql="SELECT 1", guard={"verdict": "allowed"}, row_count=1, ... columns=["n"], session_id="s_1", turn_id="t_1")) >>> with cfg.override_settings(app_db_url="sqlite:///" + os.path.join(d, "app.db")): ... dispose_app_engine() ... row, created = confirm_turn(session_id="s_1", turn_id="t_1", principal_id="analyst-1") ... again, created_again = confirm_turn(session_id="s_1", turn_id="t_1", principal_id="analyst-1") ... dispose_app_engine() >>> created, created_again, row["request_id"], again["confirmation_id"] == row["confirmation_id"] (True, False, 'r_1', True) >>> audit._AUDIT_LOG_FILE = "" """ audit_record = find_record_by_turn(session_id, turn_id) if audit_record is None: raise TurnNotConfirmableError( f"no audit record names session_id={session_id!r} turn_id={turn_id!r} " "-- there is nothing to show this answer was a successful one", reason="not_audited", ) if not turn_executed_successfully(audit_record): raise TurnNotConfirmableError( "only an answer whose query ran and returned a result can be confirmed; " "this turn was rejected, declined or failed", reason="not_successful", ) existing = get_confirmation( session_id=session_id, turn_id=turn_id, principal_id=principal_id, ) if existing is not None: return existing, False created = True engine = get_app_engine() try: with engine.begin() as conn: conn.execute( turn_confirmations.insert().values( session_id=session_id, turn_id=turn_id, principal_id=principal_id, request_id=audit_record.get("request_id"), created_at=_now_iso(), ) ) except IntegrityError: # A concurrent click won the race to the unique constraint: this # call created nothing, which is the same outcome as a repeat. created = False row = get_confirmation(session_id=session_id, turn_id=turn_id, principal_id=principal_id) assert row is not None # this insert, or the racing one, stored it return row, created def withdraw_confirmation( *, session_id: str, turn_id: str, principal_id: str, ) -> bool: """Delete *principal_id*'s own confirmation of the turn, if there is one. Never touches another principal's confirmation of the same turn. Parameters ---------- session_id, turn_id: The turn. principal_id: Whose confirmation to withdraw. Returns ------- bool ``True`` when a confirmation was deleted, ``False`` when there was none (withdrawing twice is not an error). """ engine = get_app_engine() with engine.begin() as conn: result = conn.execute( delete(turn_confirmations).where( turn_confirmations.c.session_id == session_id, turn_confirmations.c.turn_id == turn_id, turn_confirmations.c.principal_id == principal_id, ) ) return bool(result.rowcount) def list_confirmations( *, session_id: str | None = None, principal_id: str | None = None, since: str | None = None, ) -> list[dict[str, Any]]: """Every confirmation matching the given filters, newest first. Parameters ---------- session_id: Restrict to one session -- the analyst-facing read ("which of my turns here did I confirm") passes this together with *principal_id*. principal_id: Restrict to one principal's confirmations. since: ISO-8601 timestamp (inclusive); compared as plain strings, which is safe because every timestamp stored here is ``datetime.now(timezone.utc).isoformat()``. Returns ------- list[dict] Public representations, ``created_at`` descending. """ stmt = select(turn_confirmations).order_by( turn_confirmations.c.created_at.desc(), turn_confirmations.c.confirmation_id.desc(), ) if session_id is not None: stmt = stmt.where(turn_confirmations.c.session_id == session_id) if principal_id is not None: stmt = stmt.where(turn_confirmations.c.principal_id == principal_id) if since is not None: stmt = stmt.where(turn_confirmations.c.created_at >= since) engine = get_app_engine() with engine.connect() as conn: rows = conn.execute(stmt).mappings().all() return [_public(dict(row)) for row in rows]